hls_runtime/server/engine.rs
1//! [`HlsOrigin`] — the LL-HLS origin [`ServedEgress`] (plan step 4):
2//! blocking-reload/part-availability *decision* logic and playlist rendering,
3//! rendered directly from a shared [`Trunk`] instead of the deleted
4//! `MediaStore` push-fed rolling window.
5//!
6//! Master/media playlist tags are RFC 8216 §4.3.4 (`#EXT-X-STREAM-INF`) and
7//! §4.3.3 (`#EXTM3U`/`#EXT-X-VERSION`, rendered by [`MediaPlaylist::to_m3u8`]);
8//! the blocking reload query parameters (`_HLS_msn`/`_HLS_part`) are the
9//! Blocking Playlist Reload mechanism of RFC 8216bis §6.2.5.2 — the client
10//! asks the origin to hold the response open until the requested Media
11//! Sequence Number/part is available, bounded by the caller's own
12//! [`AwaitPolicy`] so the origin never hangs indefinitely.
13//!
14//! # What comes straight from the `Trunk`, with no cache at all
15//!
16//! Every part-availability and blocking-reload decision reads the `Trunk`
17//! directly, `&self`-shaped, every call:
18//!
19//! - **Live parts of the open segment** — [`Trunk::part_bytes`]/
20//! [`Trunk::parts_in_segment`] (step 3b-iv's live-part log). This is the
21//! whole reason step 3b-iv exists: before it, nothing in `Trunk` could
22//! answer "does part 3 of the segment currently being written exist",
23//! which is exactly what forced `MediaStore` to keep its own
24//! `live_parts`/`recent_parts` buffers in the first place.
25//! - **Whether a segment has closed** — [`Trunk::last_closed_segment`].
26//! - **The "in-progress-or-last-active segment" `MediaStore::latest_progress`
27//! used to track as a push-fed field** — [`HlsOrigin::live_edge`] derives
28//! it from the two queries above alone (`last_closed_segment() + 1`, probed
29//! via `parts_in_segment`), needing no field of its own. See that method's
30//! doc for the derivation and why it is exact, not a heuristic.
31//! - **A just-closed segment's final part still resolving** — falls out of
32//! [`Trunk::part_bytes`] for free: [`media_plane::trunk::SegmentWriter::publish_segment`]
33//! deliberately never touches the live-part log (see `trunk`'s own module
34//! doc, "The live-part log"), so this crate no longer needs `MediaStore`'s
35//! separate `recent_parts` buffer at all — that buffer existed *only* to
36//! simulate exactly the guarantee the `Trunk` now gives natively.
37//!
38//! # The one thing that genuinely cannot come from the `Trunk` alone
39//!
40//! [`Trunk::subscribe_segments`] hands back a moving, single-consumer
41//! [`SegmentCursor`] — there is no snapshot query over the segment log the
42//! way [`Trunk::events_between`] gives the event log (see
43//! `media_plane::egress`'s own module doc, "`ServedEgress::resolve` does not
44//! take `&Trunk`", which anticipated exactly this). Rendering a Media
45//! Playlist needs the **window** of currently-advertised closed segments
46//! (their bytes, durations, and discontinuity bits), plus two numbers that
47//! must survive eviction from that window: the lifetime-max segment
48//! duration (RFC 8216bis §4.4.3.1's `TARGETDURATION` MUST) and the
49//! cumulative discontinuity count that has rolled off the front
50//! (`#EXT-X-DISCONTINUITY-SEQUENCE`, RFC 8216 §4.3.3.3). None of that is
51//! answerable by a fresh `&self` call on `Trunk` — it has to be assembled by
52//! draining a cursor over time.
53//!
54//! `Window` is that assembly, and it is **not** a second `MediaStore`: it
55//! holds only bytes/duration/discontinuity-bit for the segments currently in
56//! the advertised window, fed by exactly **one** [`SegmentCursor`] this
57//! `HlsOrigin` owns — precisely the shape `media_plane::egress`'s module
58//! doc prescribes ("a `ServedEgress` implementation... keeps its own
59//! resolvable window in sync by draining [cursors]... `resolve` only ever
60//! reads that already-synced state"). It carries none of `MediaStore`'s
61//! other fields (`health`, `track_specs`, `created_at`, `window_segments()`
62//! diagnostics) — those served `multimux`'s DASH/ll-DASH outputs, not
63//! LL-HLS rendering, and are out of this step's scope (Step 5's problem, if
64//! still needed once `multimux` is rewritten).
65//!
66//! The fMP4 **init segment** bytes are the other thing this module holds
67//! outside the `Trunk`: an init segment is neither a sample, a finished
68//! segment, an event, nor a live part — it is produced once by the
69//! segmenter and never changes, so it was never in scope for any of
70//! `Trunk`'s four rings. [`HlsOrigin::set_init`] is the (small, honest) side
71//! channel for it — not a duplicate of anything `Trunk` holds.
72
73use super::ids::Generation;
74use media_plane::ids::{PartIndex, SegmentSeq};
75use std::collections::{BTreeMap, VecDeque};
76use std::num::NonZeroUsize;
77use std::sync::atomic::{AtomicU32, AtomicU64, Ordering};
78use std::sync::{Arc, Mutex, MutexGuard, PoisonError, Weak};
79use std::time::Duration;
80use transmux::TrackId;
81
82use broadcast_common::{Parse, Timestamp, Unpackage};
83use broadcast_hls::{
84 ByteRange, ClosedCaptionsRef, DecimalSeconds, FrameRate, LowLatencyConfig, MapTag,
85 MasterPlaylist, MediaPlaylist, MediaRendition, MediaSegment, MediaType, OpenSegment, PartSpec,
86 Variant, VideoRange,
87};
88use bytes::Bytes;
89use media_plane::egress::{AwaitPolicy, CachePolicy, EgressResponse, ServedEgress};
90use media_plane::trunk::{PartEntry, SegmentCursor, SegmentCursorItem, SegmentEntry, Trunk};
91
92/// [`HlsOrigin`]'s `drained_through` before any drain has run: no real head
93/// (`u32::MAX + 1` at most) equals it, so the first drain always happens.
94const UNDRAINED: u64 = u64::MAX;
95
96/// Process-wide high-water mark of [`next_instance_token`].
97static LAST_INSTANCE: AtomicU64 = AtomicU64::new(0);
98
99/// A token that differs for every [`HlsOrigin`] ever built, in this process
100/// **and across process restarts**: the wall clock in milliseconds, forced
101/// strictly above the previous token handed out (two origins built in the
102/// same millisecond, or after the clock stepped back, still differ). It is
103/// part of every resource name served `immutable` (see
104/// [`HlsOrigin::instance`]).
105fn next_instance_token() -> u64 {
106 let now_ms = std::time::SystemTime::now()
107 .duration_since(std::time::UNIX_EPOCH)
108 .ok()
109 .and_then(|d| u64::try_from(d.as_millis()).ok())
110 .unwrap_or(0);
111 let mut previous = LAST_INSTANCE.load(Ordering::Relaxed);
112 loop {
113 let token = now_ms.max(previous.saturating_add(1));
114 match LAST_INSTANCE.compare_exchange_weak(
115 previous,
116 token,
117 Ordering::AcqRel,
118 Ordering::Relaxed,
119 ) {
120 Ok(_) => return token,
121 Err(actual) => previous = actual,
122 }
123 }
124}
125
126/// Lock `m`, recovering the guard from a poisoned mutex.
127///
128/// Every critical section in this module leaves its data structurally
129/// consistent at each step (plain pushes/pops/assignments), so a panic in
130/// some other thread that held the lock must not turn into a panic in every
131/// later request (audit r09-O4 / T6, issues #1089/#1134).
132fn locked<T>(m: &Mutex<T>) -> MutexGuard<'_, T> {
133 m.lock().unwrap_or_else(PoisonError::into_inner)
134}
135
136/// Track id for the single rendition served per stream (no multi-track/
137/// multi-rendition support yet).
138pub const DEFAULT_TRACK_ID: TrackId = TrackId::new(1);
139
140/// Which container [`HlsOrigin`] serves segments/parts as — orthogonal to
141/// whether LL-HLS is enabled ([`HlsOriginBuilder::low_latency`]); issue #873.
142///
143/// RFC 8216bis §3.1.1 / §3.1.2 give the two containers different
144/// `#EXT-X-MAP` obligations:
145///
146/// - fMP4 (§3.1.2): "Each fMP4 Segment in a Media Playlist MUST have an
147/// `EXT-X-MAP` tag applied to it" — unconditional, so [`Container::Fmp4`]
148/// always emits one.
149/// - MPEG-2 TS (§3.1.1): "Each Transport Stream Segment MUST contain a PAT
150/// and a PMT, **or** have an `EXT-X-MAP` tag applied to it" — a
151/// disjunction, not a container restriction. `EXT-X-MAP` is legal for TS;
152/// it is not required when the segments carry their own PAT/PMT.
153///
154/// [`Container::MpegTs`] omits `#EXT-X-MAP` **by default**, on the
155/// assumption that segments come from a self-initialising source (e.g.
156/// `transmux`'s TS segmenter, which re-emits PAT+PMT at the head of every
157/// segment) — it does not *forbid* the tag; a future caller feeding
158/// pre-segmented TS without in-band PSI would need a way to opt back in,
159/// which is not implemented here (out of scope for issue #873; the current
160/// wiring never calls `set_init` from a `MpegTs`-configured pipeline, so the
161/// gap has no live caller yet).
162#[derive(Debug, Clone, Copy, PartialEq, Eq)]
163#[non_exhaustive]
164pub enum Container {
165 /// Fragmented MP4 / CMAF segments (`.m4s`), with an fMP4 init segment
166 /// (`.mp4`) referenced by an always-present `#EXT-X-MAP`.
167 Fmp4,
168 /// Whole MPEG-2 Transport Stream segments (`.ts`), self-initialising
169 /// (in-band PAT/PMT) — no `#EXT-X-MAP`, no init segment served.
170 MpegTs,
171 /// WebVTT subtitle segments (`.vtt`), each starting with its own WebVTT
172 /// header (RFC 8216 §3.5), so no `#EXT-X-MAP` and no init segment. With a
173 /// `low_latency` setting a WebVTT origin serves Partial Segments exactly
174 /// like the other containers: each part is itself a complete WebVTT
175 /// document (RFC 8216bis §3.1.4).
176 WebVtt,
177}
178
179impl Container {
180 /// The spec token for this container — `"fmp4"` / `"mpeg-ts"` / `"webvtt"`.
181 pub fn name(&self) -> &'static str {
182 match self {
183 Container::Fmp4 => "fmp4",
184 Container::MpegTs => "mpeg-ts",
185 Container::WebVtt => "webvtt",
186 }
187 }
188
189 /// The dynamic-filename extension (without the leading `.`) this
190 /// container's segments/parts are named with.
191 fn segment_extension(self) -> &'static str {
192 match self {
193 Container::Fmp4 => "m4s",
194 Container::MpegTs => "ts",
195 Container::WebVtt => "vtt",
196 }
197 }
198}
199
200broadcast_common::impl_spec_display!(Container);
201
202impl Default for Container {
203 /// [`Container::Fmp4`] — preserves every pre-#873 caller's behaviour.
204 fn default() -> Self {
205 Container::Fmp4
206 }
207}
208
209/// How many superseded init segments stay resolvable by their versioned name
210/// (`init-{track}-{instance}-{generation}.mp4`): segments and playlists that name an
211/// older generation are still cached by players and CDNs for a while after a
212/// mid-stream codec change.
213const INIT_HISTORY: usize = 8;
214
215/// Nanoseconds per second, for the integer bitrate arithmetic.
216const NANOS_PER_SEC: u128 = 1_000_000_000;
217
218/// RFC 8216bis §6.2.2: the server MUST NOT remove a Media Segment from a live
219/// Playlist if that would leave a Playlist shorter than three times the
220/// Target Duration. A segment is cut at (about) the target duration, so a
221/// window of fewer than three segments cannot satisfy that; the advertised
222/// window is never shorter than this.
223const MIN_WINDOW_SEGMENTS: usize = 3;
224
225/// RFC 8216bis §6.2.2: `EXT-X-PART` tags SHOULD be removed once they are
226/// greater than three Target Durations from the end of the Playlist (and the
227/// part must stay downloadable for as long again).
228const PART_RETENTION_TARGET_DURATIONS: f64 = 3.0;
229
230/// RFC 8216bis §6.2.6 (SHOULD): "Not Found" for a resource the server cannot
231/// find and that is not specified by an `EXT-X-PRELOAD-HINT`. The live-edge
232/// (open) segment is the one after the last closed segment, and a hinted part
233/// is in it or in the one right after it — so a part more than this many
234/// segments past the last closed one is refused rather than held open.
235const FUTURE_PART_SEGMENT_BOUND: u32 = 2;
236
237/// RFC 8216bis §6.2.5.2: the Advance Part Limit is "three divided by the Part
238/// Target Duration if the Part Target Duration is less than one second, or
239/// three otherwise". Expressed against the part target in milliseconds.
240const ADVANCE_PART_LIMIT_NUMERATOR: u64 = 3;
241const MILLIS_PER_SEC: u64 = 1_000;
242
243/// RFC 8216bis §6.2.5.2 (SHOULD): a `_HLS_msn` greater than "the Media
244/// Sequence Number of the last Media Segment in the current Playlist plus
245/// two" should be rejected rather than always blocking to the caller's
246/// timeout — a legitimate client only ever asks for the segment/part right
247/// after the one it already has, so anything more than two segments beyond
248/// the current last closed segment is either a malfunctioning client or abuse.
249const ABUSE_MSN_FUTURE_BOUND: u64 = 2;
250
251/// RFC 8216bis / Apple LL-HLS §4.4.3.7: `#EXT-X-SERVER-CONTROL`'s
252/// `PART-HOLD-BACK` attribute MUST be at least 3x the part target duration
253/// (`#EXT-X-PART-INF`'s `PART-TARGET`).
254const PART_HOLD_BACK_MULTIPLIER: f64 = 3.0;
255
256/// RFC 8216bis §4.4.3.8: `HOLD-BACK` MUST be at least three times the Target
257/// Duration.
258const HOLD_BACK_MULTIPLIER: f64 = 3.0;
259
260/// RFC 8216bis §4.4.3.8: the Skip Boundary is at least six times the Target
261/// Duration (advertised as `CAN-SKIP-UNTIL`).
262const SKIP_BOUNDARY_MULTIPLIER: f64 = 6.0;
263
264/// A minimal single-variant master playlist pointing at `media_playlist_name`
265/// (the caller's configured media-playlist filename — e.g. multimux's
266/// `Config::playlist_name`, defaulting to `"media.m3u8"`) — the same
267/// regardless of any stream state (no multi-rendition support yet), so this
268/// takes no `Trunk`/origin argument.
269pub fn master_playlist_m3u8(media_playlist_name: &str, bandwidth_bps: u32) -> String {
270 format!("#EXTM3U\n#EXT-X-STREAM-INF:BANDWIDTH={bandwidth_bps}\n{media_playlist_name}\n")
271}
272
273/// Typed value of the `_HLS_skip` Delivery Directive (RFC 8216bis §6.2.5.1):
274/// `YES` requests a delta update that skips older Media Segments, `v2` also
275/// skips older `EXT-X-DATERANGE` tags.
276#[derive(Debug, Clone, Copy, PartialEq, Eq)]
277#[non_exhaustive]
278pub enum SkipDirective {
279 /// Skip older Media Segments only (`_HLS_skip=YES`).
280 Yes,
281 /// Skip older Media Segments and older `EXT-X-DATERANGE` tags
282 /// (`_HLS_skip=v2`).
283 V2,
284}
285
286impl SkipDirective {
287 /// The spec token for this directive value.
288 pub fn name(&self) -> &'static str {
289 match self {
290 SkipDirective::Yes => "YES",
291 SkipDirective::V2 => "v2",
292 }
293 }
294}
295
296broadcast_common::impl_spec_display!(SkipDirective);
297
298/// HTTP Priority signalling data for an LL-HLS response (RFC 8216bis §6.2.6
299/// / §6.3): the extensible priority scheme (RFC 9218) urgency the adapter
300/// should attach to the response. Playlists MUST have urgency 1; Media
301/// Segments urgencies 2 through 6; Partial Segments share the segment's
302/// urgency band.
303#[derive(Debug, Clone, Copy, PartialEq, Eq)]
304pub struct HttpPriority {
305 /// RFC 9218 urgency value (lower is more urgent).
306 pub urgency: u8,
307 /// Whether the response may be delivered incrementally.
308 pub incremental: bool,
309}
310
311impl HttpPriority {
312 /// Urgency for a Media Playlist response.
313 pub const PLAYLIST: Self = Self {
314 urgency: 1,
315 incremental: false,
316 };
317 /// Urgency for a Partial Segment or whole Media Segment response.
318 pub const SEGMENT: Self = Self {
319 urgency: 2,
320 incremental: true,
321 };
322}
323
324/// Blocking playlist reload query parameters (RFC 8216bis §6.2.5.2) — the
325/// sans-IO counterpart of an adapter's own (likely serde-`Deserialize`)
326/// query-string type; the adapter maps its wire query params into this.
327#[derive(Debug, Default, Clone, PartialEq, Eq)]
328pub struct BlockingQuery {
329 /// The Media Sequence Number the client already has, plus one — the
330 /// origin should not respond until a segment/part beyond this is ready.
331 pub hls_msn: Option<u64>,
332 /// The part index (within `hls_msn`) the client is waiting for.
333 pub hls_part: Option<u32>,
334 /// The `_HLS_skip` Delivery Directive (RFC 8216bis §6.2.5.1), if any.
335 pub hls_skip: Option<SkipDirective>,
336 /// The `_HLS_primary` / `_HLS_primary_id` value (RFC 8216bis Appendix D.4),
337 /// carried opaquely; the origin does not alter playlist rendering for it.
338 pub hls_primary: Option<String>,
339}
340
341/// [`ServedEgress::Request`] for [`HlsOrigin`]: which wire resource is
342/// being asked for. A data-carrying dispatch ADT (matches this crate's
343/// `client::action::Action`/`ResourceId` convention) — see
344/// `tests/label_coverage.rs`'s SKIP list.
345#[derive(Debug, Clone, PartialEq, Eq)]
346#[non_exhaustive]
347pub enum HlsRequest {
348 /// `GET <media playlist>`, optionally carrying a blocking-reload query.
349 Playlist {
350 /// The track id to render the playlist for (a naming parameter only —
351 /// see [`DEFAULT_TRACK_ID`]).
352 track_id: TrackId,
353 /// The blocking-reload query parameters, if any.
354 query: BlockingQuery,
355 },
356 /// `GET` a dynamic origin resource by its wire filename (`init-{track}.mp4`,
357 /// `seg-{track}-{seq}.m4s`, `part-{track}-{seq}.{idx}.m4s`).
358 Resource {
359 /// The requested filename, exactly as it appeared in the request path.
360 name: String,
361 },
362 /// `GET <iframe playlist>` for `track_id`, optionally carrying a
363 /// blocking-reload query.
364 IFramePlaylist {
365 /// The track id to render the I-frame playlist for.
366 track_id: TrackId,
367 /// The blocking-reload query parameters, if any.
368 query: BlockingQuery,
369 /// Filename of the primary (media) playlist this I-frame playlist's
370 /// `#EXT-X-RENDITION-REPORT` points at (RFC 8216bis §4.4.5.4).
371 media_playlist_name: String,
372 },
373}
374
375/// [`ServedEgress::Body`] for [`HlsOrigin`]: the resolved body, typed by
376/// which [`HlsRequest`] produced it. A data-carrying ADT — see
377/// `tests/label_coverage.rs`'s SKIP list.
378#[derive(Debug, Clone, PartialEq, Eq)]
379#[non_exhaustive]
380pub enum HlsBody {
381 /// A rendered Media Playlist (`#EXTM3U` text).
382 Playlist(String),
383 /// Resolved resource bytes (init/segment/part).
384 Resource(Bytes),
385}
386
387/// One playlist-window-resident **closed** segment's identity/bytes —
388/// `Window`'s per-entry shape. Deliberately narrower than the old
389/// `MediaStore`'s `SegmentInfo`-derived window entries: this crate only ever
390/// needs bytes + duration + the discontinuity bit to render a Media
391/// Playlist, so that is all this holds.
392#[derive(Clone)]
393struct WindowSegment {
394 sequence_number: SegmentSeq,
395 /// The init-segment generation this segment was cut against (see
396 /// [`InitStore`]).
397 init_gen: Generation,
398 bytes: Bytes,
399 duration_secs: f64,
400 discontinuous: bool,
401 /// Byte length of the leading I-frame slice of this segment's first
402 /// video sample, when the segment is usable in an I-frame playlist;
403 /// `None` otherwise (audio-first, non-sync first sample, parse failure,
404 /// or no video track).
405 iframe_len: Option<u64>,
406 /// This segment's `SegmentEntry::timeline_position`, in nanoseconds —
407 /// carried through so [`HlsOrigin::closed_segments`] can hand it to a
408 /// caller doing time-based windowing over a *different* source of
409 /// segments (e.g. multimux's DVR archive, issue #900), without that
410 /// caller needing a second cursor of its own just to learn it.
411 start_ns: u64,
412 /// Mean video frame rate of this segment in millihertz, measured when the
413 /// stream's SPS carries no VUI timing; `None` when not measurable.
414 video_millihertz: Option<u32>,
415}
416
417/// One closed segment's identity/metadata as tracked by this origin's
418/// `Window` — [`HlsOrigin::closed_segments`]'s return shape. Deliberately
419/// excludes the segment's bytes: a caller merging this with another
420/// segment source (issue #900) fetches bytes through the normal
421/// [`ServedEgress::resolve`] resource path when it needs them, not through
422/// this snapshot.
423#[derive(Debug, Clone, Copy, PartialEq)]
424#[non_exhaustive]
425pub struct ClosedSegment {
426 /// This segment's sequence number (`_HLS_msn`-addressable).
427 pub sequence_number: SegmentSeq,
428 /// `SegmentEntry::timeline_position`, in nanoseconds — the `Trunk`'s
429 /// absolute timeline, for time-based windowing across sources that
430 /// share the same `Trunk` (e.g. multimux's DVR archive, whose own
431 /// `IndexEntry::start_pts_ns` is exactly this same clock).
432 pub start_ns: u64,
433 /// Segment duration in seconds.
434 pub duration_secs: f64,
435 /// Whether `#EXT-X-DISCONTINUITY` precedes this segment (RFC 8216
436 /// §4.3.4.3).
437 pub discontinuous: bool,
438 /// The init-segment generation this segment was cut against (see
439 /// [`HlsOrigin::init_name`]).
440 pub init_generation: Generation,
441}
442
443impl ClosedSegment {
444 /// The same segment, cut against init generation `generation`.
445 #[must_use]
446 pub fn with_init_generation(mut self, generation: Generation) -> Self {
447 self.init_generation = generation;
448 self
449 }
450
451 /// Construct a [`ClosedSegment`] — needed because the type is
452 /// `#[non_exhaustive]` (a struct-literal outside this crate does not
453 /// typecheck), for a caller building test fixtures over the shape
454 /// [`HlsOrigin::closed_segments`] returns (e.g. multimux's own
455 /// catch-up merge tests, issue #900) without depending on this
456 /// module's private `Window`.
457 pub fn new(
458 sequence_number: SegmentSeq,
459 start_ns: u64,
460 duration_secs: f64,
461 discontinuous: bool,
462 ) -> Self {
463 ClosedSegment {
464 sequence_number,
465 start_ns,
466 duration_secs,
467 discontinuous,
468 init_generation: Generation::new(1),
469 }
470 }
471}
472
473/// The small per-[`HlsOrigin`] synced window this module's own doc
474/// ("The one thing that genuinely cannot come from the `Trunk` alone")
475/// explains the need for — fed by draining exactly one [`SegmentCursor`],
476/// never pushed into directly.
477struct Window {
478 segments: VecDeque<WindowSegment>,
479 capacity: usize,
480 /// Largest segment duration ever drained, surviving window eviction —
481 /// RFC 8216bis §4.4.3.1's `TARGETDURATION` MUST holds for *every*
482 /// segment this origin has ever advertised, not just the ones still in
483 /// the window (mirrors the deleted `MediaStore::max_segment_duration`).
484 max_segment_duration_secs: f64,
485 /// Highest per-segment bitrate (bits/second, rounded up) ever drained,
486 /// surviving eviction — what a master playlist's `BANDWIDTH` (RFC 8216
487 /// §4.3.4.2: an upper bound of the peak segment bitrate) is built from.
488 /// `0` until a segment with a measurable duration has been drained.
489 peak_bandwidth_bps: u64,
490 /// Cumulative bytes of every segment ever drained, surviving window
491 /// eviction — the numerator of the master playlist's
492 /// `AVERAGE-BANDWIDTH` (RFC 8216bis §4.4.6.2, Apple authoring item 9.14:
493 /// the average segment bit rate over the whole stream, not just the
494 /// window).
495 total_bytes: u64,
496 /// Cumulative duration of every segment ever drained whose duration was
497 /// non-zero, in nanoseconds, surviving eviction — the denominator of
498 /// `AVERAGE-BANDWIDTH`. `0` until such a segment has been drained.
499 total_duration_nanos: u128,
500 /// Cumulative peak per-I-frame bitrate (bits/second, rounded up) over
501 /// every I-frame-playlist-usable segment ever drained, surviving window
502 /// eviction — `BANDWIDTH` of an `#EXT-X-I-FRAME-STREAM-INF`.
503 iframe_peak_bandwidth_bps: u64,
504 /// Cumulative byte length of every I-frame slice ever drained, surviving
505 /// window eviction — the numerator (in bytes, converted to bits) of an
506 /// I-frame variant's `AVERAGE-BANDWIDTH`.
507 iframe_total_bytes: u64,
508 /// Cumulative duration of every I-frame-playlist-usable segment ever
509 /// drained whose duration was non-zero, in nanoseconds, surviving
510 /// window eviction — the denominator of an I-frame variant's
511 /// `AVERAGE-BANDWIDTH`.
512 iframe_total_duration_nanos: u128,
513 /// Cumulative count of discontinuities that have rolled off the front of
514 /// the window — RFC 8216 §4.3.3.3's `#EXT-X-DISCONTINUITY-SEQUENCE`.
515 /// Incremented exactly once per **evicted** entry whose
516 /// [`WindowSegment::discontinuous`] was `true`; a discontinuity still
517 /// inside the window is rendered as a per-segment `#EXT-X-DISCONTINUITY`
518 /// tag instead (see [`MediaPlaylist::to_m3u8`]), never double-counted
519 /// here.
520 discontinuity_sequence: u64,
521 /// The cursor reported lost segments (`Lagged`/`Gap`) since the last
522 /// push: the next segment does not follow the window's last one.
523 gap_pending: bool,
524}
525
526impl Window {
527 fn new(capacity: NonZeroUsize) -> Self {
528 Window {
529 segments: VecDeque::new(),
530 capacity: capacity.get().max(MIN_WINDOW_SEGMENTS),
531 max_segment_duration_secs: 0.0,
532 peak_bandwidth_bps: 0,
533 total_bytes: 0,
534 total_duration_nanos: 0,
535 iframe_peak_bandwidth_bps: 0,
536 iframe_total_bytes: 0,
537 iframe_total_duration_nanos: 0,
538 discontinuity_sequence: 0,
539 gap_pending: false,
540 }
541 }
542
543 /// Record that the cursor lost segments: whatever is pushed next does not
544 /// follow the current window contents.
545 fn note_gap(&mut self) {
546 self.gap_pending = true;
547 }
548
549 fn max_video_millihertz(&self) -> Option<u32> {
550 self.segments
551 .iter()
552 .filter_map(|segment| segment.video_millihertz)
553 .max()
554 }
555
556 /// Drop every window entry (they roll off the front, so their
557 /// discontinuities count towards the discontinuity sequence).
558 fn clear_rolling_off(&mut self) {
559 let discontinuities = self.segments.iter().filter(|s| s.discontinuous).count();
560 self.discontinuity_sequence = self
561 .discontinuity_sequence
562 .saturating_add(u64::try_from(discontinuities).unwrap_or(u64::MAX));
563 self.segments.clear();
564 }
565
566 /// Absorb one drained [`SegmentEntry`], evicting the oldest window entry
567 /// first if already at `capacity` — same evict-then-push shape as every
568 /// ring in `trunk.rs` itself.
569 ///
570 /// A playlist numbers its segments implicitly (`EXT-X-MEDIA-SEQUENCE` +
571 /// index, RFC 8216 §4.3.3.2), so the window must hold consecutive
572 /// sequence numbers. An entry that does not directly follow the last one
573 /// (numbers skipped, a repeat or a decrease), or that follows reported
574 /// lost segments, starts a new run: the old entries are cleared and the
575 /// entry is marked discontinuous (RFC 8216 §4.3.4.3) rather than being
576 /// appended under a wrong implied number.
577 fn push(
578 &mut self,
579 entry: SegmentEntry,
580 init_gen: Generation,
581 iframe_len: Option<u64>,
582 video_millihertz: Option<u32>,
583 ) {
584 let follows = self
585 .segments
586 .back()
587 .is_none_or(|last| last.sequence_number.checked_next() == Some(entry.sequence_number));
588 let mut discontinuous = entry.meta.discontinuous;
589 if !follows || self.gap_pending {
590 self.clear_rolling_off();
591 discontinuous = true;
592 }
593 self.gap_pending = false;
594 // RFC 8216bis §4.4.4.5: a different Media Initialization Section
595 // starts a new timeline segment — `EXT-X-DISCONTINUITY` precedes the
596 // first segment that uses it (audit r09-C2, issue #1030).
597 if self
598 .segments
599 .back()
600 .is_some_and(|last| last.init_gen != init_gen)
601 {
602 discontinuous = true;
603 }
604
605 let duration_secs = entry.duration.as_secs_f64();
606 self.max_segment_duration_secs = self.max_segment_duration_secs.max(duration_secs);
607 if let Some(bps) = segment_bandwidth_bps(entry.bytes.len(), entry.duration) {
608 self.peak_bandwidth_bps = self.peak_bandwidth_bps.max(bps);
609 }
610 // Totals are cumulative over every drained segment and are never
611 // reset on eviction (or on a discontinuity): `AVERAGE-BANDWIDTH`
612 // describes the whole stream. A zero-duration segment has no
613 // measurable bit rate, so it contributes to neither total.
614 if !entry.duration.is_zero() {
615 self.total_bytes = self
616 .total_bytes
617 .saturating_add(u64::try_from(entry.bytes.len()).unwrap_or(u64::MAX));
618 self.total_duration_nanos = self
619 .total_duration_nanos
620 .saturating_add(entry.duration.as_nanos());
621 }
622 // I-frame accounting: per usable entry, bits = iframe_len * 8 over the
623 // segment's EXTINF duration. Cumulative over every drained segment,
624 // surviving eviction (mirrors the media totals above).
625 if let Some(len) = iframe_len {
626 if let Some(bps) =
627 segment_bandwidth_bps(usize::try_from(len).unwrap_or(usize::MAX), entry.duration)
628 {
629 self.iframe_peak_bandwidth_bps = self.iframe_peak_bandwidth_bps.max(bps);
630 }
631 if !entry.duration.is_zero() {
632 self.iframe_total_bytes = self.iframe_total_bytes.saturating_add(len);
633 self.iframe_total_duration_nanos = self
634 .iframe_total_duration_nanos
635 .saturating_add(entry.duration.as_nanos());
636 }
637 }
638 if self.segments.len() >= self.capacity
639 && let Some(evicted) = self.segments.pop_front()
640 && evicted.discontinuous
641 {
642 self.discontinuity_sequence = self.discontinuity_sequence.saturating_add(1);
643 }
644 self.segments.push_back(WindowSegment {
645 sequence_number: entry.sequence_number,
646 init_gen,
647 bytes: entry.bytes,
648 duration_secs,
649 discontinuous,
650 start_ns: entry.timeline_position.as_nanos(),
651 iframe_len,
652 video_millihertz,
653 });
654 }
655
656 /// The bytes of segment `sequence_number`. The window holds consecutive
657 /// sequence numbers (see [`Self::push`]), so the number identifies at
658 /// most one entry; it is addressed by offset from the front and the
659 /// entry's own number is re-checked.
660 fn bytes_of(&self, sequence_number: SegmentSeq) -> Option<Bytes> {
661 let front = self.segments.front()?.sequence_number;
662 let index = usize::try_from(sequence_number.checked_distance_from(front)?).ok()?;
663 let segment = self.segments.get(index)?;
664 (segment.sequence_number == sequence_number).then(|| segment.bytes.clone())
665 }
666}
667
668/// A copy of the window state `render_playlist` reads (see
669/// [`HlsOrigin::window_view`]).
670struct WindowView {
671 segments: VecDeque<WindowSegment>,
672 max_segment_duration_secs: f64,
673 discontinuity_sequence: u64,
674}
675
676/// The init segments this origin has served, by generation.
677///
678/// A resource name must always map to the same bytes for `Immutable` to be
679/// true (audit r09-C2, issue #1030), so a changed init segment gets a new
680/// generation and a new versioned name,
681/// `init-{track}-{instance}-{generation}.mp4`; the
682/// bare `init-{track}.mp4` always means "the current one" and is served
683/// [`CachePolicy::NoCache`].
684#[derive(Default)]
685struct InitStore {
686 /// Generation of `current` (`0` = no init set yet); starts at `1`.
687 generation: Generation,
688 current: Option<Bytes>,
689 /// The last [`INIT_HISTORY`] generations' bytes, oldest first.
690 history: VecDeque<(Generation, Bytes)>,
691 /// `(first origin sequence number cut against it, generation)`, oldest
692 /// first: segment `seq` belongs to the last entry whose first number is
693 /// `<= seq`.
694 starts: VecDeque<(SegmentSeq, Generation)>,
695}
696
697impl InitStore {
698 /// Record `bytes`; a new generation only if they differ from the current
699 /// init. `next_seq` is the first segment not yet closed.
700 fn set(&mut self, bytes: Bytes, next_seq: SegmentSeq) {
701 if self.current.as_ref() == Some(&bytes) {
702 return;
703 }
704 self.generation = Generation::new(self.generation.value().saturating_add(1));
705 // The very first init covers every segment ever seen.
706 let first_seq = if self.current.is_none() {
707 SegmentSeq::new(0)
708 } else {
709 next_seq
710 };
711 self.current = Some(bytes.clone());
712 self.history.push_back((self.generation, bytes));
713 self.starts.push_back((first_seq, self.generation));
714 while self.history.len() > INIT_HISTORY {
715 self.history.pop_front();
716 }
717 while self.starts.len() > INIT_HISTORY {
718 self.starts.pop_front();
719 }
720 }
721
722 /// The generation segment `seq` was cut against (`1` before any init).
723 fn generation_for(&self, seq: SegmentSeq) -> Generation {
724 self.starts
725 .iter()
726 .rev()
727 .find(|(first, _)| *first <= seq)
728 .or_else(|| self.starts.front())
729 .map_or(Generation::new(1), |&(_, generation)| generation)
730 }
731
732 fn bytes_of(&self, generation: Generation) -> Option<Bytes> {
733 self.history
734 .iter()
735 .find(|(g, _)| *g == generation)
736 .map(|(_, b)| b.clone())
737 }
738}
739
740/// Whether a codec config is a video codec (mirrors
741/// `transmux::CodecConfig::is_video`, which is crate-private).
742/// `SUPPLEMENTAL-CODECS` compatibility brand of a Dolby Vision stream whose
743/// base layer is HDR10 (`dv_bl_signal_compatibility_id` 1)
744/// (`private/specs-verified/hls/apple/hls-authoring-spec.md`, "The
745/// SUPPLEMENTAL-CODECS attribute").
746const DV_BRAND_HDR10: &str = "db1p";
747/// Brand for an HLG-compatible base layer (id 4), same table.
748const DV_BRAND_HLG: &str = "db4h";
749/// `SUPPLEMENTAL-CODECS` compatibility brand announcing HDR10+ dynamic
750/// metadata (`private/specs-verified/hls/apple/hls-authoring-spec.md`, "The
751/// SUPPLEMENTAL-CODECS attribute": "You can indicate the presence of HDR10+
752/// metadata by adding a brand of 'cdm4'").
753const HDR10PLUS_BRAND: &str = "cdm4";
754/// ITU-T H.273 `transfer_characteristics` of SMPTE ST 2084 (PQ).
755const TRANSFER_PQ: u8 = 16;
756const MILLIHERTZ_PER_HZ: u128 = 1000;
757
758const MIN_SAMPLES_FOR_RATE: u64 = 2;
759
760/// Implementation-defined sanity cap (not a spec limit): a per-segment
761/// measurement above this is broken timing, not a frame rate, and is dropped
762/// so one bad segment cannot advertise it in `FRAME-RATE`.
763const MAX_PLAUSIBLE_FRAME_RATE_MILLIHERTZ: u32 = 300 * 1000;
764const SUPPLEMENTAL_CODECS_ATTR: &str = "SUPPLEMENTAL-CODECS";
765
766/// How a Dolby Vision video track is announced in the multivariant playlist.
767struct DolbyVisionHls {
768 codecs: String,
769 supplemental_codecs: Option<String>,
770 video_range: VideoRange,
771}
772
773/// Apple HLS authoring spec "The SUPPLEMENTAL-CODECS attribute": a
774/// compatible base layer keeps its own codec string in `CODECS` and names the
775/// Dolby Vision codec plus brand in `SUPPLEMENTAL-CODECS`, with `VIDEO-RANGE`
776/// matching the brand (HDR10 base: PQ, HLG base: HLG). A base layer that is not
777/// SDR/HDR compliant (id 0, profile 5) is announced as `dvh1.PP.LL` in `CODECS`
778/// with `VIDEO-RANGE=PQ`
779/// (`private/specs-verified/dolby/vision/dolby-vision-streams-in-hls-v2.0-2018-11.md`
780/// section 3 examples). Other compatibility ids have no brand in the verified
781/// text and are not signalled.
782fn dolby_vision_hls(spec: &transmux::TrackSpec) -> Option<DolbyVisionHls> {
783 use transmux::DoviBlSignalCompatibilityId as Ccid;
784 let record = spec.dovi.as_ref()?;
785 if !record.is_single_layer() || !record.is_hevc_profile() {
786 return None;
787 }
788 let dv = record.codec_string()?;
789 match record.bl_signal_compatibility_id {
790 Ccid::None => Some(DolbyVisionHls {
791 codecs: dv,
792 supplemental_codecs: None,
793 video_range: VideoRange::Pq,
794 }),
795 Ccid::Hdr10 => Some(DolbyVisionHls {
796 codecs: transmux::rfc6381_codec_string(&spec.config).ok()?,
797 supplemental_codecs: Some(format!("{dv}/{DV_BRAND_HDR10}")),
798 video_range: VideoRange::Pq,
799 }),
800 Ccid::Hlg => Some(DolbyVisionHls {
801 codecs: transmux::rfc6381_codec_string(&spec.config).ok()?,
802 supplemental_codecs: Some(format!("{dv}/{DV_BRAND_HLG}")),
803 video_range: VideoRange::Hlg,
804 }),
805 _ => None,
806 }
807}
808
809/// `SUPPLEMENTAL-CODECS` value of an HDR10+ HEVC track: the track's own HEVC
810/// codec string plus the `cdm4` brand. Apple prints this form only for AV1
811/// (`av01.0.05M.10.0.112/cdm4`); applying the same rule to HEVC
812/// (`hvc1.2.4.L153.B0/cdm4`) is an INFERENCE from the rule, not an Apple
813/// example (`private/specs-verified/smpte/hdr10plus/hdr10plus-manifest-signalling.md`).
814/// Only a PQ stream is announced, since HDR10+ rides on an HDR10 base layer
815/// and HLS has no separate `VIDEO-RANGE` value for it.
816fn hdr10plus_supplemental_codecs(spec: &transmux::TrackSpec) -> Option<String> {
817 if !spec.hdr10plus || !matches!(spec.config, transmux::CodecConfig::Hevc { .. }) {
818 return None;
819 }
820 let transfer = spec
821 .display
822 .as_ref()?
823 .video_signal
824 .as_ref()?
825 .colour
826 .as_ref()?
827 .transfer
828 .code();
829 if transfer != TRANSFER_PQ {
830 return None;
831 }
832 let codec = transmux::rfc6381_codec_string(&spec.config).ok()?;
833 Some(format!("{codec}/{HDR10PLUS_BRAND}"))
834}
835
836fn is_video_codec(config: &transmux::CodecConfig) -> bool {
837 matches!(
838 config,
839 transmux::CodecConfig::Avc { .. }
840 | transmux::CodecConfig::Hevc { .. }
841 | transmux::CodecConfig::Vvc { .. }
842 | transmux::CodecConfig::Av1 { .. }
843 | transmux::CodecConfig::Vp9 { .. }
844 | transmux::CodecConfig::Vp8 { .. }
845 | transmux::CodecConfig::Mpeg2Video { .. }
846 )
847}
848
849fn dedupe_rendition_names(mut media: Vec<MediaRendition>) -> Vec<MediaRendition> {
850 let mut seen: std::collections::BTreeSet<(String, String)> = std::collections::BTreeSet::new();
851 for m in &mut media {
852 let base = m.name.clone();
853 let mut n = 1u32;
854 while !seen.insert((m.group_id.clone(), m.name.clone())) {
855 n += 1;
856 m.name = format!("{base} ({n})");
857 }
858 }
859 media
860}
861
862fn channel_count_string(config: &transmux::CodecConfig) -> Option<String> {
863 let n = match config {
864 transmux::CodecConfig::Aac { channel_count, .. }
865 | transmux::CodecConfig::Ac3 { channel_count, .. }
866 | transmux::CodecConfig::Eac3 { channel_count, .. }
867 | transmux::CodecConfig::Opus { channel_count, .. }
868 | transmux::CodecConfig::Flac { channel_count, .. }
869 | transmux::CodecConfig::Ac4 { channel_count, .. }
870 | transmux::CodecConfig::MpegH { channel_count, .. }
871 | transmux::CodecConfig::MpegAudio { channel_count, .. } => *channel_count,
872 _ => return None,
873 };
874 Some(n.to_string())
875}
876
877/// The top-level `moof` of `segment` and its offset.
878fn first_moof(segment: &[u8]) -> Option<(usize, transmux::movie_fragment::MovieFragmentBox)> {
879 // Locate the (single) top-level `moof` box. Its offset is measured from
880 // the start of `segment`, matching `frag_offsets::sample_ranges_in`.
881 let mut offset = 0usize;
882 let mut moof_off: Option<usize> = None;
883 let mut moof_total: usize = 0;
884 let mut moof_hdr: usize = 0;
885 while offset + transmux::box_types::BOX_HEADER_MIN_SIZE <= segment.len() {
886 let rest = &segment[offset..];
887 let header = transmux::box_types::BoxHeader::parse(rest).ok()?;
888 let hdr_sz = header.header_size();
889 if hdr_sz > rest.len() {
890 break;
891 }
892 let total = if header.size == u64::from(transmux::box_types::SIZE_TO_EOF) {
893 rest.len()
894 } else {
895 match usize::try_from(header.size) {
896 Ok(t) if t >= hdr_sz => t,
897 _ => break,
898 }
899 };
900 if header.box_type.is(b"moof") {
901 moof_off = Some(offset);
902 moof_total = total.min(rest.len());
903 moof_hdr = hdr_sz;
904 break;
905 }
906 let step = total.min(rest.len());
907 if step == 0 {
908 break;
909 }
910 offset += step;
911 }
912 let moof_off = moof_off?;
913 let body = &segment[moof_off + moof_hdr..moof_off + moof_total];
914 let moof = transmux::movie_fragment::MovieFragmentBox::parse_body(body).ok()?;
915 Some((moof_off, moof))
916}
917
918/// Mean video frame rate of one segment in millihertz, rounded to three
919/// decimal places (RFC 8216bis section 4.4.6.2): `samples * timescale / ticks`.
920fn segment_frame_rate_millihertz(samples: u64, ticks: u64, timescale: u32) -> Option<u32> {
921 if samples < MIN_SAMPLES_FOR_RATE || ticks == 0 || timescale == 0 {
922 return None;
923 }
924 let ticks = u128::from(ticks);
925 let millihertz =
926 (u128::from(samples) * u128::from(timescale) * MILLIHERTZ_PER_HZ + ticks / 2) / ticks;
927 u32::try_from(millihertz)
928 .ok()
929 .filter(|mh| *mh <= MAX_PLAUSIBLE_FRAME_RATE_MILLIHERTZ)
930}
931
932/// Mean video frame rate of an fMP4 `segment` from its `trun` durations, or
933/// `None` when the segment cannot be parsed or holds no video samples.
934fn fmp4_video_rate(segment: &[u8], video_track_id: TrackId, timescale: u32) -> Option<u32> {
935 let (moof_off, moof) = first_moof(segment)?;
936 let mdats = transmux::frag_offsets::mdat_ranges(segment);
937 let samples =
938 transmux::frag_offsets::sample_ranges_in(segment, &mdats, moof_off, &moof, video_track_id)
939 .ok()?;
940 let ticks: u64 = samples.iter().map(|r| u64::from(r.duration)).sum();
941 segment_frame_rate_millihertz(u64::try_from(samples.len()).ok()?, ticks, timescale)
942}
943
944/// Mean video frame rate of a self-initialising MPEG-TS `segment` from the
945/// demuxed video samples' DTS deltas (PTS when a sample has no DTS), which
946/// `TsDemux` unwraps through `broadcast_common::clock33`.
947fn ts_video_rate(segment: &[u8]) -> Option<u32> {
948 let media = transmux::ts_demux::TsDemux::new().demux(segment).ok()?;
949 let track = media
950 .tracks
951 .iter()
952 .find(|track| is_video_codec(&track.spec.config))?;
953 let ticks: u64 = track
954 .samples
955 .iter()
956 .filter_map(|sample| sample.duration)
957 .map(u64::from)
958 .sum();
959 segment_frame_rate_millihertz(
960 u64::try_from(track.samples.len()).ok()?,
961 ticks,
962 track.spec.timescale,
963 )
964}
965
966/// Compute the byte length of the leading I-frame slice of `segment`'s first
967/// video sample, or `None` when the segment cannot back an I-frame playlist
968/// entry (audio-first layout, non-sync first sample, parse failure, or no
969/// usable mdat/moof).
970fn compute_iframe_len(segment: &[u8], video_track_id: TrackId) -> Option<u64> {
971 let (moof_off, moof) = first_moof(segment)?;
972 let mdats = transmux::frag_offsets::mdat_ranges(segment);
973 let samples =
974 transmux::frag_offsets::sample_ranges_in(segment, &mdats, moof_off, &moof, video_track_id)
975 .ok()?;
976 if samples.is_empty() {
977 return None;
978 }
979 let first = samples[0];
980 // A sample is a sync sample when bit 16 of its flags is clear.
981 if first.flags & (1 << 16) != 0 {
982 return None;
983 }
984 // The first video sample must start exactly at an mdat payload start: no
985 // other track's bytes may precede it.
986 let mdat_payload_start = mdat_payload_start_for(segment, first.start)?;
987 if first.start != mdat_payload_start {
988 return None;
989 }
990 u64::try_from(first.end).ok()
991}
992
993/// Byte offset of the `mdat` payload that contains `sample_start`, if any.
994fn mdat_payload_start_for(segment: &[u8], sample_start: usize) -> Option<usize> {
995 let mut offset = 0usize;
996 while offset + transmux::box_types::BOX_HEADER_MIN_SIZE <= segment.len() {
997 let rest = &segment[offset..];
998 let header = transmux::box_types::BoxHeader::parse(rest).ok()?;
999 let hdr_sz = header.header_size();
1000 if hdr_sz > rest.len() {
1001 break;
1002 }
1003 let total = if header.size == u64::from(transmux::box_types::SIZE_TO_EOF) {
1004 rest.len()
1005 } else {
1006 match usize::try_from(header.size) {
1007 Ok(t) if t >= hdr_sz => t,
1008 _ => break,
1009 }
1010 };
1011 if header.box_type.is(b"mdat") {
1012 let payload = offset + hdr_sz;
1013 let end = offset + total;
1014 if sample_start >= payload && sample_start < end {
1015 return Some(payload);
1016 }
1017 }
1018 let step = total.min(rest.len());
1019 if step == 0 {
1020 break;
1021 }
1022 offset += step;
1023 }
1024 None
1025}
1026
1027/// The mean segment bitrate in bits/second over the whole stream — RFC
1028/// 8216bis §4.4.6.2 `AVERAGE-BANDWIDTH` (Apple authoring item 9.14) — rounded
1029/// to nearest, or `None` when no measurable segment has been drained
1030/// (`total_duration_nanos == 0`) or the figure does not fit
1031/// a `u32`.
1032///
1033/// Exact integer arithmetic: `bits * 1e9` is computed in `u128` (a `u64` byte
1034/// total times 8e9 cannot overflow it) before the single division.
1035fn average_bandwidth_bps(total_bytes: u64, total_duration_nanos: u128) -> Option<u32> {
1036 if total_duration_nanos == 0 {
1037 return None;
1038 }
1039 let bits_nanos = u128::from(total_bytes)
1040 .checked_mul(8)?
1041 .checked_mul(NANOS_PER_SEC)?;
1042 let mean = bits_nanos.checked_add(total_duration_nanos / 2)? / total_duration_nanos;
1043 u32::try_from(mean).ok()
1044}
1045
1046/// A segment's bitrate in bits/second, rounded up (so a `BANDWIDTH` built
1047/// from it is never below the true value), or `None` for a zero duration or
1048/// an overflowing figure.
1049fn segment_bandwidth_bps(len: usize, duration: Duration) -> Option<u64> {
1050 let nanos = duration.as_nanos();
1051 if nanos == 0 {
1052 return None;
1053 }
1054 let bits = u128::try_from(len).ok()?.checked_mul(8)?;
1055 let bps = bits.checked_mul(NANOS_PER_SEC)?.div_ceil(nanos);
1056 u64::try_from(bps).ok()
1057}
1058
1059/// Parse `part-{track}-{instance}-{seq}.{idx}.{ext}` (instance-named, the form
1060/// the playlist uses) or the legacy `part-{track}-{seq}.{idx}.{ext}` into
1061/// `(instance, seq, idx)`; `instance` is `None` for the legacy form. `None` if it
1062/// isn't a part filename in `container`'s extension or a field does not parse.
1063fn parse_part(file: &str, container: Container) -> Option<(Option<u64>, u64, PartIndex)> {
1064 let suffix = format!(".{}", container.segment_extension());
1065 let rest = file.strip_prefix("part-")?.strip_suffix(suffix.as_str())?;
1066 let (names, idx) = rest.rsplit_once('.')?;
1067 let (epoch, seq) = parse_epoch_seq(names)?;
1068 Some((epoch, seq, PartIndex::new(idx.parse().ok()?)))
1069}
1070
1071/// `{track}-{seq}` or `{track}-{instance}-{seq}` -> `(instance, seq)`; `{track}` is
1072/// validated but unused (an [`HlsOrigin`] holds a single track).
1073fn parse_epoch_seq(names: &str) -> Option<(Option<u64>, u64)> {
1074 let mut fields = names.split('-');
1075 fields.next()?.parse::<u32>().ok()?;
1076 let first = fields.next()?;
1077 match fields.next() {
1078 None => Some((None, first.parse().ok()?)),
1079 Some(seq) => {
1080 if fields.next().is_some() {
1081 return None;
1082 }
1083 Some((Some(first.parse().ok()?), seq.parse().ok()?))
1084 }
1085 }
1086}
1087
1088/// Wire prefix for an I-frame (trick-play) segment resource (issue #1422).
1089///
1090/// An I-frame playlist references its segment bytes under this prefix rather
1091/// than the media playlist's `seg-` prefix, so the bytes behind an I-frame
1092/// URI are *always* the stored segment bytes: `multimux`'s DASH `emsg`
1093/// splice (and the in-progress chunked fallback) key off `seg-` only and
1094/// never touch `iframe-seg-` URIs, which keeps the advertised
1095/// `#EXT-X-BYTERANGE` length (measured on the stored bytes by
1096/// `compute_iframe_len`) from drifting away from the served bytes.
1097pub const IFRAME_SEGMENT_PREFIX: &str = "iframe-seg-";
1098
1099/// What [`parse_immediate`] recognised.
1100enum ImmediateResource {
1101 /// `init-{track}.mp4`: the current init (changes, never `immutable`).
1102 CurrentInit,
1103 /// `init-{track}-{instance}-{generation}.mp4` (`epoch` is the instance
1104 /// token).
1105 VersionedInit { epoch: u64, generation: Generation },
1106 /// `seg-{track}-{seq}.{ext}` (`epoch: None`, legacy) or
1107 /// `seg-{track}-{instance}-{seq}.{ext}` (`epoch` is the instance token).
1108 Segment { epoch: Option<u64>, seq: u64 },
1109}
1110
1111/// Parse an `init-…`/`seg-…` dynamic filename; `part-…` filenames are handled
1112/// separately by [`parse_part`] (they can block until available).
1113///
1114/// The init forms are only recognised under [`Container::Fmp4`] — a `MpegTs`
1115/// origin's segments are self-initialising (see [`Container`]'s own doc), so
1116/// `init-*.mp4` under `MpegTs` is `None` whether or not
1117/// [`HlsOrigin::set_init`] was ever called (issue #873's cross-container
1118/// refusal: advertised == servable).
1119fn parse_immediate(file: &str, container: Container) -> Option<ImmediateResource> {
1120 if container == Container::Fmp4
1121 && let Some(rest) = file.strip_prefix("init-")
1122 {
1123 let rest = rest.strip_suffix(".mp4")?;
1124 let mut fields = rest.split('-');
1125 fields.next()?.parse::<u32>().ok()?;
1126 return match (fields.next(), fields.next(), fields.next()) {
1127 (None, ..) => Some(ImmediateResource::CurrentInit),
1128 (Some(epoch), Some(generation), None) => Some(ImmediateResource::VersionedInit {
1129 epoch: epoch.parse().ok()?,
1130 generation: Generation::new(generation.parse().ok()?),
1131 }),
1132 _ => None,
1133 };
1134 }
1135 // `iframe-seg-…` resolves to the *same* stored segment bytes as
1136 // `seg-…` (issue #1422): the I-frame playlist advertises a byte range
1137 // measured on the stored bytes, so its URI must never be transformed by
1138 // `multimux`'s emsg splice. Recognising the prefix here makes the
1139 // origin serve the unmodified bytes for it.
1140 let seg_rest = file
1141 .strip_prefix("seg-")
1142 .or_else(|| file.strip_prefix(IFRAME_SEGMENT_PREFIX));
1143 if let Some(rest) = seg_rest {
1144 let suffix = format!(".{}", container.segment_extension());
1145 let (epoch, seq) = parse_epoch_seq(rest.strip_suffix(suffix.as_str())?)?;
1146 return Some(ImmediateResource::Segment { epoch, seq });
1147 }
1148 None
1149}
1150
1151/// Error returned by [`HlsOriginBuilder::build`] when a required field was
1152/// never set — never a silently-defaulted value (issue #873).
1153#[derive(Debug, Clone, Copy, PartialEq, Eq, thiserror::Error)]
1154#[non_exhaustive]
1155pub enum HlsOriginBuildError {
1156 /// [`HlsOriginBuilder::target_duration_secs`] was never called.
1157 #[error("HlsOrigin::builder(...).target_duration_secs(...) is required but was never called")]
1158 MissingTargetDurationSecs,
1159 /// [`HlsOriginBuilder::window_segments`] was never called.
1160 #[error("HlsOrigin::builder(...).window_segments(...) is required but was never called")]
1161 MissingWindowSegments,
1162 /// [`HlsOriginBuilder::target_duration_secs`] was NaN, infinite, zero or
1163 /// negative: `EXT-X-TARGETDURATION` must be a positive integer (RFC
1164 /// 8216bis §4.4.3.1).
1165 #[error("HlsOrigin::builder(...).target_duration_secs(...) must be finite and positive")]
1166 InvalidTargetDuration,
1167 /// [`HlsOriginBuilder::low_latency`] was given a part target of `0` ms:
1168 /// `#EXT-X-PART-INF`'s `PART-TARGET` must be a positive duration (RFC
1169 /// 8216bis §4.4.3.7), and the Advance Part Limit divides by it.
1170 #[error("HlsOrigin::builder(...).low_latency(0): the part target must be positive")]
1171 ZeroPartTarget,
1172 /// [`HlsOriginBuilder::media_sequence_offset`] is so large that
1173 /// `offset + u32::MAX` (the largest origin sequence number) would not fit
1174 /// a `u64` Media Sequence Number.
1175 #[error(
1176 "HlsOrigin::builder(...).media_sequence_offset(...) leaves no room for u32 sequence numbers"
1177 )]
1178 MediaSequenceOffsetTooLarge,
1179 /// [`HlsOriginBuilder::declared_bandwidth_bps`] was given `0` or a value
1180 /// greater than `u32::MAX`.
1181 #[error("HlsOrigin::builder(...).declared_bandwidth_bps(...) must be > 0 and <= u32::MAX")]
1182 InvalidDeclaredBandwidth,
1183 /// An audio rendition's `group_id` is empty.
1184 #[error("HlsOrigin::builder(...).audio_renditions(...): group_id must be non-empty")]
1185 EmptyAudioRenditionGroupId,
1186 /// An audio rendition's `name` is empty.
1187 #[error("HlsOrigin::builder(...).audio_renditions(...): name must be non-empty")]
1188 EmptyAudioRenditionName,
1189 /// Two audio renditions in one group share a `NAME`.
1190 #[error("HlsOrigin::builder(...).audio_renditions(...): duplicate NAME in a group")]
1191 DuplicateAudioRenditionName,
1192 /// An audio rendition group does not have exactly one `DEFAULT=YES` member.
1193 #[error(
1194 "HlsOrigin::builder(...).audio_renditions(...): a group must have exactly one DEFAULT=YES member"
1195 )]
1196 AudioRenditionDefaultCount,
1197 /// A `DEFAULT=YES` audio rendition is not `AUTOSELECT=YES` (RFC 8216bis §4.4.6.1).
1198 #[error("HlsOrigin::builder(...).audio_renditions(...): DEFAULT=YES requires AUTOSELECT=YES")]
1199 AudioRenditionDefaultNotAutoselect,
1200 /// An audio rendition's `language` is empty.
1201 #[error("HlsOrigin::builder(...).audio_renditions(...): language must be non-empty")]
1202 EmptyAudioRenditionLanguage,
1203 /// An audio rendition's `uri` is empty.
1204 #[error("HlsOrigin::builder(...).audio_renditions(...): uri must be non-empty")]
1205 EmptyAudioRenditionUri,
1206 /// Audio rendition groups do not all have the same members (RFC 8216bis §4.4.6.1).
1207 #[error(
1208 "HlsOrigin::builder(...).audio_renditions(...): every group must have the same members, differing only in URI and CHANNELS"
1209 )]
1210 AudioRenditionGroupsDiffer,
1211 /// A subtitle rendition's `name` is empty.
1212 #[error("HlsOrigin::builder(...).subtitle_renditions(...): name must be non-empty")]
1213 EmptySubtitleRenditionName,
1214 /// A subtitle rendition's `language` is empty (Apple HLS Authoring Spec 4.7).
1215 #[error("HlsOrigin::builder(...).subtitle_renditions(...): language must be non-empty")]
1216 EmptySubtitleRenditionLanguage,
1217 /// A subtitle rendition's `uri` is empty.
1218 #[error("HlsOrigin::builder(...).subtitle_renditions(...): uri must be non-empty")]
1219 EmptySubtitleRenditionUri,
1220 /// Two subtitle renditions share a `NAME`.
1221 #[error("HlsOrigin::builder(...).subtitle_renditions(...): duplicate NAME")]
1222 DuplicateSubtitleRenditionName,
1223 /// Two subtitle renditions share a `URI`.
1224 #[error("HlsOrigin::builder(...).subtitle_renditions(...): duplicate URI")]
1225 DuplicateSubtitleRenditionUri,
1226 /// A `DEFAULT=YES` subtitle rendition is not `AUTOSELECT=YES` (RFC 8216 §4.3.4.1.1).
1227 #[error(
1228 "HlsOrigin::builder(...).subtitle_renditions(...): DEFAULT=YES requires AUTOSELECT=YES"
1229 )]
1230 SubtitleRenditionDefaultNotAutoselect,
1231 /// More than one subtitle rendition is `DEFAULT=YES` (RFC 8216 §4.3.4.1.1).
1232 #[error("HlsOrigin::builder(...).subtitle_renditions(...): at most one DEFAULT=YES member")]
1233 MultipleDefaultSubtitleRenditions,
1234 /// `subtitle_sibling_origins` is non-empty but not one per declared rendition.
1235 #[error(
1236 "HlsOrigin::builder(...).subtitle_sibling_origins(...): one origin per declared subtitle rendition"
1237 )]
1238 SubtitleSiblingCountMismatch,
1239 /// A subtitle sibling origin is not a [`Container::WebVtt`] origin.
1240 #[error(
1241 "HlsOrigin::builder(...).subtitle_sibling_origins(...): sibling must be a WebVtt origin"
1242 )]
1243 SubtitleSiblingNotWebVtt,
1244 /// A subtitle sibling's `EXT-X-TARGETDURATION` differs from this origin's
1245 /// (Apple HLS Authoring Spec 5.6).
1246 #[error(
1247 "HlsOrigin::builder(...).subtitle_sibling_origins(...): sibling target duration must equal this origin's"
1248 )]
1249 SubtitleSiblingTargetDurationMismatch,
1250 /// A closed-caption override's `instream_id` is not `CC1`..`CC4` or
1251 /// `SERVICE1`..`SERVICE63`.
1252 #[error("HlsOrigin::builder(...).closed_caption_overrides(...): invalid INSTREAM-ID")]
1253 InvalidClosedCaptionInstreamId,
1254 /// Two closed-caption overrides name the same `INSTREAM-ID`.
1255 #[error("HlsOrigin::builder(...).closed_caption_overrides(...): duplicate INSTREAM-ID")]
1256 DuplicateClosedCaptionInstreamId,
1257}
1258
1259/// Error returned by [`HlsOrigin::master_playlist`].
1260#[derive(Debug, PartialEq, thiserror::Error)]
1261#[non_exhaustive]
1262pub enum HlsMasterError {
1263 /// No segment with a measurable duration has been published yet and no
1264 /// declared bandwidth was configured, so `BANDWIDTH` cannot be
1265 /// determined.
1266 #[error("BANDWIDTH unknown: no measured segment and no declared bandwidth")]
1267 BandwidthUnknown,
1268 /// The measured peak segment bitrate does not fit the `u32` `BANDWIDTH`
1269 /// of a master playlist variant.
1270 #[error("measured BANDWIDTH {bandwidth} exceeds u32")]
1271 BandwidthOverflow {
1272 /// The measured peak segment bitrate, bits per second.
1273 bandwidth: u64,
1274 },
1275 /// The master playlist could not be rendered (a field carries a
1276 /// character forbidden in an HLS quoted string).
1277 #[error("master playlist render failed: {0}")]
1278 Render(broadcast_hls::Error),
1279 /// An audio rendition's `uri` equals the media playlist name.
1280 #[error("audio rendition uri must not equal the media playlist name")]
1281 AudioRenditionUriEqualsMediaPlaylist,
1282}
1283
1284/// Validate a list of audio rendition declarations (RFC 8216bis §4.4.6.1).
1285pub fn validate_audio_renditions(decls: &[AudioRenditionDecl]) -> Result<(), HlsOriginBuildError> {
1286 use std::collections::{BTreeMap, BTreeSet};
1287 let mut groups: BTreeMap<&str, Vec<&AudioRenditionDecl>> = BTreeMap::new();
1288 for decl in decls {
1289 if decl.group_id.is_empty() {
1290 return Err(HlsOriginBuildError::EmptyAudioRenditionGroupId);
1291 }
1292 if decl.name.is_empty() {
1293 return Err(HlsOriginBuildError::EmptyAudioRenditionName);
1294 }
1295 if decl.language.is_empty() {
1296 return Err(HlsOriginBuildError::EmptyAudioRenditionLanguage);
1297 }
1298 if decl.uri.is_empty() {
1299 return Err(HlsOriginBuildError::EmptyAudioRenditionUri);
1300 }
1301 if decl.default && !decl.autoselect {
1302 return Err(HlsOriginBuildError::AudioRenditionDefaultNotAutoselect);
1303 }
1304 groups.entry(&decl.group_id).or_default().push(decl);
1305 }
1306 let mut signatures = BTreeSet::new();
1307 for members in groups.values() {
1308 let mut names = BTreeSet::new();
1309 if !members.iter().all(|m| names.insert(m.name.as_str())) {
1310 return Err(HlsOriginBuildError::DuplicateAudioRenditionName);
1311 }
1312 if members.iter().filter(|m| m.default).count() != 1 {
1313 return Err(HlsOriginBuildError::AudioRenditionDefaultCount);
1314 }
1315 let mut signature: Vec<_> = members
1316 .iter()
1317 .map(|m| {
1318 (
1319 m.name.as_str(),
1320 m.language.as_str(),
1321 m.default,
1322 m.autoselect,
1323 m.characteristics.as_slice(),
1324 )
1325 })
1326 .collect();
1327 signature.sort();
1328 signatures.insert(signature);
1329 }
1330 if signatures.len() > 1 {
1331 return Err(HlsOriginBuildError::AudioRenditionGroupsDiffer);
1332 }
1333 Ok(())
1334}
1335
1336/// Relaxed validation for split-audio routes where `LANGUAGE`/`NAME`/`CHANNELS`
1337/// may be left empty to be derived from the sibling audio track at render time
1338/// (issue #1428 slice 3). Only structural rules that do not depend on those
1339/// fields are enforced.
1340pub fn validate_audio_renditions_relaxed(
1341 decls: &[AudioRenditionDecl],
1342) -> Result<(), HlsOriginBuildError> {
1343 use std::collections::BTreeMap;
1344 for decl in decls {
1345 if decl.group_id.is_empty() {
1346 return Err(HlsOriginBuildError::EmptyAudioRenditionGroupId);
1347 }
1348 if decl.uri.is_empty() {
1349 return Err(HlsOriginBuildError::EmptyAudioRenditionUri);
1350 }
1351 if decl.default && !decl.autoselect {
1352 return Err(HlsOriginBuildError::AudioRenditionDefaultNotAutoselect);
1353 }
1354 }
1355 let mut groups: BTreeMap<&str, Vec<&AudioRenditionDecl>> = BTreeMap::new();
1356 for decl in decls {
1357 groups.entry(&decl.group_id).or_default().push(decl);
1358 }
1359 for members in groups.values() {
1360 if members.iter().filter(|m| m.default).count() != 1 {
1361 return Err(HlsOriginBuildError::AudioRenditionDefaultCount);
1362 }
1363 }
1364 Ok(())
1365}
1366
1367/// A typed declaration of one audio rendition for a multivariant playlist
1368/// (RFC 8216bis §4.4.6.1, Apple HLS authoring spec items 9.5-9.8). Consumed
1369/// by [`HlsOriginBuilder::audio_renditions`].
1370#[derive(Debug, Clone, PartialEq, Eq, Default)]
1371#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
1372pub struct AudioRenditionDecl {
1373 /// `GROUP-ID` of the audio rendition group.
1374 pub group_id: String,
1375 /// `NAME` of the rendition. On a split-audio route it may be left empty
1376 /// and is then derived from the track (the language, plus ` (AD)` for an
1377 /// audio-description track), made unique within its group.
1378 pub name: String,
1379 /// `LANGUAGE` of the rendition; must be non-empty (`"und"` for
1380 /// undetermined) unless split-audio derivation is active, which then
1381 /// fills it from the track's PMT `ISO_639_language_descriptor`,
1382 /// canonicalised to a BCP 47 (RFC 5646) primary language subtag via
1383 /// [`bcp47_language`] (ISO 639-2/B and 639-2/T are mapped to ISO 639-1
1384 /// when one exists).
1385 pub language: String,
1386 /// `DEFAULT` attribute.
1387 pub default: bool,
1388 /// `AUTOSELECT` attribute.
1389 pub autoselect: bool,
1390 /// `CHANNELS` attribute (e.g. `"2"` or `"JOC 3/2.0.2"`). `None` on a
1391 /// split-audio route derives the codec config's channel count.
1392 pub channels: Option<String>,
1393 /// URI of the audio media playlist.
1394 pub uri: String,
1395 /// `CHARACTERISTICS` MCTs. Empty on a split-audio route derives
1396 /// `public.accessibility.describes-video` for an audio-description
1397 /// track; a non-empty value always wins.
1398 pub characteristics: Vec<String>,
1399}
1400
1401/// `GROUP-ID` of the `EXT-X-MEDIA TYPE=SUBTITLES` group the master playlist
1402/// advertises for declared subtitle renditions.
1403pub const SUBTITLES_GROUP_ID: &str = "subs";
1404
1405/// `CHARACTERISTICS` of a subtitles track for deaf or hard-of-hearing viewers
1406/// (Apple HLS Authoring Spec item 4.5).
1407const HEARING_IMPAIRED_CHARACTERISTICS: [&str; 2] = [
1408 "public.accessibility.transcribes-spoken-dialog",
1409 "public.accessibility.describes-music-and-sound",
1410];
1411
1412/// `CODECS` token naming WebVTT subtitles (Apple HLS Authoring Spec item 5.10).
1413const WEBVTT_CODEC: &str = "wvtt";
1414
1415/// A typed declaration of one WebVTT subtitle rendition for a multivariant
1416/// playlist (RFC 8216 §4.3.4.1, Apple HLS Authoring Spec items 4.5-4.7, 5.11).
1417/// Consumed by [`HlsOriginBuilder::subtitle_renditions`]. All renditions share
1418/// the group [`SUBTITLES_GROUP_ID`].
1419///
1420/// `hearing_impaired` emits the item 4.5 `CHARACTERISTICS` and forces
1421/// `AUTOSELECT=YES` (item 4.6); `forced` emits `FORCED=YES` and forces
1422/// `AUTOSELECT=YES` (item 5.11).
1423#[derive(Debug, Clone, PartialEq, Eq)]
1424#[non_exhaustive]
1425pub struct SubtitleRenditionDecl {
1426 /// `NAME` of the rendition.
1427 pub name: String,
1428 /// `LANGUAGE` (BCP 47); required (Apple HLS Authoring Spec item 4.7).
1429 pub language: String,
1430 /// URI of the subtitle media playlist.
1431 pub uri: String,
1432 /// `DEFAULT` attribute.
1433 pub default: bool,
1434 /// `AUTOSELECT` attribute.
1435 pub autoselect: bool,
1436 /// `FORCED` attribute.
1437 pub forced: bool,
1438 /// Subtitles for deaf or hard-of-hearing viewers.
1439 pub hearing_impaired: bool,
1440}
1441
1442impl SubtitleRenditionDecl {
1443 /// A plain subtitle rendition: `DEFAULT=NO`, `AUTOSELECT=YES`, not forced,
1444 /// not for the hard of hearing.
1445 pub fn new(
1446 name: impl Into<String>,
1447 language: impl Into<String>,
1448 uri: impl Into<String>,
1449 ) -> Self {
1450 SubtitleRenditionDecl {
1451 name: name.into(),
1452 language: language.into(),
1453 uri: uri.into(),
1454 default: false,
1455 autoselect: true,
1456 forced: false,
1457 hearing_impaired: false,
1458 }
1459 }
1460
1461 fn effective_autoselect(&self) -> bool {
1462 self.autoselect || self.forced || self.hearing_impaired
1463 }
1464}
1465
1466/// Validate subtitle rendition declarations (RFC 8216 §4.3.4.1.1; Apple HLS
1467/// Authoring Spec item 4.7).
1468pub fn validate_subtitle_renditions(
1469 decls: &[SubtitleRenditionDecl],
1470) -> Result<(), HlsOriginBuildError> {
1471 use std::collections::BTreeSet;
1472 let mut names = BTreeSet::new();
1473 let mut uris = BTreeSet::new();
1474 let mut defaults = 0usize;
1475 for decl in decls {
1476 if decl.name.is_empty() {
1477 return Err(HlsOriginBuildError::EmptySubtitleRenditionName);
1478 }
1479 if decl.language.is_empty() {
1480 return Err(HlsOriginBuildError::EmptySubtitleRenditionLanguage);
1481 }
1482 if decl.uri.is_empty() {
1483 return Err(HlsOriginBuildError::EmptySubtitleRenditionUri);
1484 }
1485 if decl.default && !decl.effective_autoselect() {
1486 return Err(HlsOriginBuildError::SubtitleRenditionDefaultNotAutoselect);
1487 }
1488 if !names.insert(decl.name.as_str()) {
1489 return Err(HlsOriginBuildError::DuplicateSubtitleRenditionName);
1490 }
1491 if !uris.insert(decl.uri.as_str()) {
1492 return Err(HlsOriginBuildError::DuplicateSubtitleRenditionUri);
1493 }
1494 defaults += usize::from(decl.default);
1495 }
1496 if defaults > 1 {
1497 return Err(HlsOriginBuildError::MultipleDefaultSubtitleRenditions);
1498 }
1499 Ok(())
1500}
1501
1502/// `GROUP-ID` of the `EXT-X-MEDIA TYPE=CLOSED-CAPTIONS` group the master
1503/// playlist advertises for detected captions.
1504pub const CLOSED_CAPTIONS_GROUP_ID: &str = "cc";
1505/// Operator override of how one detected caption stream is advertised
1506/// (RFC 8216bis §4.4.6.1, Apple HLS authoring spec item 4.4). It never makes
1507/// a stream appear: a stream is advertised only once the video is seen to
1508/// carry it. Consumed by [`HlsOriginBuilder::closed_caption_overrides`].
1509#[derive(Debug, Clone, PartialEq, Eq, Default)]
1510#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
1511pub struct ClosedCaptionDecl {
1512 /// The `INSTREAM-ID` this override applies to: `"CC1"`..`"CC4"` or
1513 /// `"SERVICE1"`..`"SERVICE63"`.
1514 pub instream_id: String,
1515 /// `NAME`; empty keeps the default, the `INSTREAM-ID` itself.
1516 pub name: String,
1517 /// `LANGUAGE` (BCP 47); empty omits the attribute. The bitstream carries
1518 /// no language, and Apple's `mediastreamvalidator` flags the placeholder
1519 /// `"und"` as an unrecognised subtag, so none is invented; Apple's
1520 /// authoring rule 4.4 asks for one, which is what this field is for.
1521 pub language: String,
1522}
1523
1524/// Validate closed-caption overrides: well-formed, unique `INSTREAM-ID`s.
1525pub fn validate_closed_caption_overrides(
1526 decls: &[ClosedCaptionDecl],
1527) -> Result<(), HlsOriginBuildError> {
1528 let mut seen: Vec<broadcast_hls::InstreamId> = Vec::new();
1529 for decl in decls {
1530 let id = broadcast_hls::InstreamId::parse_closed_captions(&decl.instream_id)
1531 .map_err(|_| HlsOriginBuildError::InvalidClosedCaptionInstreamId)?;
1532 if seen.contains(&id) {
1533 return Err(HlsOriginBuildError::DuplicateClosedCaptionInstreamId);
1534 }
1535 seen.push(id);
1536 }
1537 Ok(())
1538}
1539
1540/// A sibling rendition's origin + the URI of its media playlist, used to
1541/// render `#EXT-X-RENDITION-REPORT` (RFC 8216bis §4.4.5.4) into this
1542/// origin's media playlist. Stored as a [`Weak`] reference so origins can
1543/// refer to each other without forming a reference cycle.
1544#[derive(Clone)]
1545pub struct RenditionSibling {
1546 /// Weak handle to the sibling origin.
1547 pub origin: Weak<HlsOrigin>,
1548 /// URI of the sibling's media playlist, relative to the same stream root
1549 /// as this origin's own playlist.
1550 pub playlist_uri: String,
1551}
1552
1553/// Fluent builder for [`HlsOrigin`] (issue #873) — replaces the old
1554/// four-positional `HlsOrigin::new` (deleted; this crate is at 0.4.0
1555/// unpublished, so there is no compatibility burden), which could not
1556/// express "classic HLS, no low latency" at all since `part_target_ms` was a
1557/// mandatory positional argument.
1558///
1559/// ```
1560/// # use std::num::NonZeroUsize;
1561/// # use std::sync::Arc;
1562/// # use hls_runtime::server::{Container, HlsOrigin};
1563/// # use media_plane::trunk::{Trunk, TrunkConfig};
1564/// # let nz = |n: usize| NonZeroUsize::new(n).unwrap();
1565/// # let trunk = Arc::new(Trunk::new(TrunkConfig::new(nz(16), nz(4), nz(8), nz(4), nz(16))));
1566/// let classic_ts = HlsOrigin::builder(Arc::clone(&trunk))
1567/// .target_duration_secs(6.0)
1568/// .window_segments(nz(4))
1569/// .container(Container::MpegTs)
1570/// // `.low_latency(..)` omitted entirely -> classic HLS.
1571/// .build()
1572/// .expect("both required fields were set");
1573/// # let _ = classic_ts;
1574/// ```
1575pub struct HlsOriginBuilder {
1576 trunk: Arc<Trunk>,
1577 target_duration_secs: Option<f64>,
1578 window_segments: Option<NonZeroUsize>,
1579 container: Container,
1580 part_target_ms: Option<u32>,
1581 media_sequence_offset: u64,
1582 instance: Option<u64>,
1583 arrival_clock: Option<Arc<dyn Fn() -> i64 + Send + Sync>>,
1584 declared_bandwidth_bps: Option<u64>,
1585 audio_renditions: Vec<AudioRenditionDecl>,
1586 audio_sibling_origins: Vec<Arc<HlsOrigin>>,
1587 subtitle_renditions: Vec<SubtitleRenditionDecl>,
1588 subtitle_sibling_origins: Vec<Arc<HlsOrigin>>,
1589 playlist_uri: Option<String>,
1590 rendition_siblings: Vec<RenditionSibling>,
1591 closed_captions_none: bool,
1592 closed_caption_overrides: Vec<ClosedCaptionDecl>,
1593}
1594
1595impl HlsOriginBuilder {
1596 fn new(trunk: Arc<Trunk>) -> Self {
1597 HlsOriginBuilder {
1598 trunk,
1599 target_duration_secs: None,
1600 window_segments: None,
1601 container: Container::default(),
1602 part_target_ms: None,
1603 media_sequence_offset: 0,
1604 instance: None,
1605 arrival_clock: None,
1606 declared_bandwidth_bps: None,
1607 audio_renditions: Vec::new(),
1608 audio_sibling_origins: Vec::new(),
1609 subtitle_renditions: Vec::new(),
1610 subtitle_sibling_origins: Vec::new(),
1611 playlist_uri: None,
1612 rendition_siblings: Vec::new(),
1613 closed_captions_none: false,
1614 closed_caption_overrides: Vec::new(),
1615 }
1616 }
1617
1618 /// `#EXT-X-TARGETDURATION`'s configured floor (RFC 8216bis §4.4.3.1) —
1619 /// required; [`HlsOriginBuilder::build`] errors if this is never called.
1620 /// The actually-rendered value is raised to the largest real segment
1621 /// duration seen, if that ever exceeds this (see `render_playlist`).
1622 pub fn target_duration_secs(mut self, target_duration_secs: f64) -> Self {
1623 self.target_duration_secs = Some(target_duration_secs);
1624 self
1625 }
1626
1627 /// How many closed segments this origin advertises in a rendered Media
1628 /// Playlist — required; independent of
1629 /// [`media_plane::trunk::TrunkConfig::segment_capacity`] (the `Trunk`'s
1630 /// own retention bound): a caller may legitimately want a shorter
1631 /// advertised window than the `Trunk` retains for other consumers (e.g. a
1632 /// DVR `SegmentEgress` reading the same `Trunk`).
1633 pub fn window_segments(mut self, window_segments: NonZeroUsize) -> Self {
1634 self.window_segments = Some(window_segments);
1635 self
1636 }
1637
1638 /// Which container this origin serves segments/parts as. Defaults to
1639 /// [`Container::Fmp4`] if never called, matching every pre-#873 caller's
1640 /// behaviour. Orthogonal to [`Self::low_latency`] — all four
1641 /// `{Fmp4, MpegTs} x {classic, low-latency}` combinations are valid.
1642 pub fn container(mut self, container: Container) -> Self {
1643 self.container = container;
1644 self
1645 }
1646
1647 /// Opt into LL-HLS: `part_target_ms` becomes `#EXT-X-PART-INF`'s
1648 /// `PART-TARGET` (milliseconds). **Omit this call entirely for classic
1649 /// HLS** — no `#EXT-X-PART`/`#EXT-X-PART-INF`/`#EXT-X-SERVER-CONTROL`/
1650 /// `#EXT-X-PRELOAD-HINT` tags are then rendered, regardless of
1651 /// [`Self::container`]. This is what the old constructor's mandatory
1652 /// `part_target_ms` positional could not express.
1653 pub fn low_latency(mut self, part_target_ms: u32) -> Self {
1654 self.part_target_ms = Some(part_target_ms);
1655 self
1656 }
1657
1658 /// Add `offset` to every Media Sequence Number this origin shows: the
1659 /// playlist's `EXT-X-MEDIA-SEQUENCE`, the numbers in segment/part URIs and
1660 /// the `_HLS_msn` it answers to. Default `0`.
1661 ///
1662 /// A fresh [`Trunk`] restarts its segment numbers at `1`; the offset is
1663 /// the mechanism that keeps the Media Sequence Number a client sees
1664 /// increasing across a replacement origin (RFC 8216bis §6.2.2: it MUST NOT
1665 /// decrease). Pass `previous.next_media_sequence()` — not `- 1`: the number
1666 /// the previous origin's *open* segment was shown with is skipped too, as
1667 /// its parts were already served (see [`HlsOrigin::next_media_sequence`]).
1668 pub fn media_sequence_offset(mut self, offset: u64) -> Self {
1669 self.media_sequence_offset = offset;
1670 self
1671 }
1672
1673 /// Override the per-origin [`HlsOrigin::instance`] token (default: a fresh
1674 /// wall-clock-seeded one per build). For deterministic tests and for an
1675 /// embedder that wants names stable across a rebuild it knows is the
1676 /// *same* media.
1677 pub fn instance(mut self, token: u64) -> Self {
1678 self.instance = Some(token);
1679 self
1680 }
1681
1682 /// Configure an **arrival** wall-clock source for `#EXT-X-PROGRAM-DATE-TIME`
1683 /// (Apple authoring 8.4 MUST — PDT present in live playlists). `clock`
1684 /// returns Unix epoch milliseconds; it is invoked **at most once** over the
1685 /// origin's lifetime (to stamp the first drained segment), never per
1686 /// request, so the caller supplies whatever wall clock it wants (e.g. a
1687 /// closure over `std::time::SystemTime` or a test counter) without the
1688 /// sans-IO engine touching a clock itself. When omitted (the default) no
1689 /// arrival source is used and PDTs come only from an explicit trunk
1690 /// `time_anchor` (B1); when both are absent no PDT is emitted.
1691 ///
1692 /// The arrival source anchors PDTs to the trunk's own continuous segment
1693 /// timeline rather than to a (possibly jumping) source media clock, so a
1694 /// source-clock discontinuity never makes the date↔segment mapping
1695 /// ambiguous (server-requirements.md item 26).
1696 pub fn program_date_time_arrival(mut self, clock: Arc<dyn Fn() -> i64 + Send + Sync>) -> Self {
1697 self.arrival_clock = Some(clock);
1698 self
1699 }
1700
1701 /// Operator-declared peak bitrate (bits/second) for this origin's
1702 /// variant stream, used as a floor for `BANDWIDTH` in the master
1703 /// playlist before any segment with a measurable duration has been
1704 /// published (and as a lower bound afterwards). Validated at
1705 /// [`Self::build`] (`> 0` and `<= u32::MAX`).
1706 pub fn declared_bandwidth_bps(mut self, declared_bandwidth_bps: u64) -> Self {
1707 self.declared_bandwidth_bps = Some(declared_bandwidth_bps);
1708 self
1709 }
1710
1711 /// Declare the audio renditions this origin's master playlist advertises
1712 /// (RFC 8216bis §4.4.6.1). Each entry becomes one `#EXT-X-MEDIA:TYPE=AUDIO`
1713 /// line, emitted before the variants; the video variant's `AUDIO` attribute
1714 /// is set to the entry's `group_id`. Empty by default (no audio
1715 /// rendition lines, byte-identical to the pre-existing output).
1716 pub fn audio_renditions(mut self, audio_renditions: Vec<AudioRenditionDecl>) -> Self {
1717 self.audio_renditions = audio_renditions;
1718 self
1719 }
1720
1721 /// Sibling audio origins (issue #1428 slice 3) whose measured peak/average
1722 /// bandwidth and `CODECS` are folded into this (video) origin's
1723 /// multivariant playlist for split-audio routes. One per declared
1724 /// rendition, in the same order. Each group's `EXT-X-STREAM-INF` carries
1725 /// `BANDWIDTH` = video peak + the group's largest audio peak and
1726 /// `AVERAGE-BANDWIDTH` likewise over averages (Apple authoring items
1727 /// 9.13/9.14); `CODECS` is the video codec plus the group's distinct audio
1728 /// codecs. [`HlsOriginBuilder::declared_bandwidth_bps`] floors the video
1729 /// part only, never the audio part. Empty (the default) keeps the slice 1
1730 /// behaviour: declared-only `CODECS`/`BANDWIDTH`.
1731 pub fn audio_sibling_origins(mut self, origins: Vec<Arc<HlsOrigin>>) -> Self {
1732 self.audio_sibling_origins = origins;
1733 self
1734 }
1735
1736 /// Declare the WebVTT subtitle renditions this origin's master playlist
1737 /// advertises: one `#EXT-X-MEDIA:TYPE=SUBTITLES` per declaration in group
1738 /// [`SUBTITLES_GROUP_ID`], placed after the audio renditions, with every
1739 /// variant's `SUBTITLES` attribute set and `wvtt` added to its `CODECS`
1740 /// (Apple HLS Authoring Spec item 5.10) when `CODECS` is known. Empty by
1741 /// default: the master is byte-identical to the pre-existing output.
1742 pub fn subtitle_renditions(mut self, subtitle_renditions: Vec<SubtitleRenditionDecl>) -> Self {
1743 self.subtitle_renditions = subtitle_renditions;
1744 self
1745 }
1746
1747 /// The [`Container::WebVtt`] origins serving the declared subtitle
1748 /// renditions, one per declaration in the same order. Checked at
1749 /// [`Self::build`]: each must be a `WebVtt` origin whose
1750 /// `EXT-X-TARGETDURATION` equals this origin's (Apple HLS Authoring Spec
1751 /// item 5.6 — live subtitle playlists must match the other media).
1752 /// Subtitle bandwidth is not folded into `BANDWIDTH`.
1753 pub fn subtitle_sibling_origins(mut self, origins: Vec<Arc<HlsOrigin>>) -> Self {
1754 self.subtitle_sibling_origins = origins;
1755 self
1756 }
1757
1758 /// URI of this origin's own media playlist, relative to the stream root
1759 /// (e.g. `"media.m3u8"` or `"audio/a0.m3u8"`). Used to compute the
1760 /// relative `URI` of [`RenditionSibling`] entries when rendering
1761 /// `#EXT-X-RENDITION-REPORT` (RFC 8216bis §4.4.5.4). Defaults to
1762 /// `"media.m3u8"` when unset.
1763 pub fn playlist_uri(mut self, playlist_uri: impl Into<String>) -> Self {
1764 self.playlist_uri = Some(playlist_uri.into());
1765 self
1766 }
1767
1768 /// Sibling renditions whose live edge is reported by this origin's media
1769 /// playlist (RFC 8216bis §4.4.5.4). Each entry is a weak handle plus the
1770 /// sibling's playlist URI. Empty by default.
1771 pub fn rendition_siblings(mut self, siblings: Vec<RenditionSibling>) -> Self {
1772 self.rendition_siblings = siblings;
1773 self
1774 }
1775
1776 /// When `true`, emit `CLOSED-CAPTIONS=NONE` on the video variant
1777 /// (RFC 8216bis §4.4.6.2). Default `false` leaves the attribute omitted,
1778 /// matching the pre-existing output.
1779 pub fn closed_captions_none(mut self, closed_captions_none: bool) -> Self {
1780 self.closed_captions_none = closed_captions_none;
1781 self
1782 }
1783
1784 /// Override `NAME`/`LANGUAGE` of detected caption streams (see
1785 /// [`ClosedCaptionDecl`]). Empty by default.
1786 pub fn closed_caption_overrides(mut self, overrides: Vec<ClosedCaptionDecl>) -> Self {
1787 self.closed_caption_overrides = overrides;
1788 self
1789 }
1790
1791 /// Build the [`HlsOrigin`], subscribing its one [`SegmentCursor`]
1792 /// immediately (so the window starts empty but never misses a segment
1793 /// published from this point on).
1794 ///
1795 /// Errors, never silently defaults, if [`Self::target_duration_secs`] or
1796 /// [`Self::window_segments`] was never called, for a zero
1797 /// [`Self::low_latency`] part target, or for an offset that cannot hold
1798 /// the `u32` sequence range.
1799 ///
1800 /// The advertised window is never shorter than three segments, whatever
1801 /// [`Self::window_segments`] says (RFC 8216bis §6.2.2: a live Playlist
1802 /// MUST NOT be shortened below three Target Durations). Part retention is
1803 /// not checked here: the part ring's capacity belongs to the [`Trunk`]
1804 /// and is not observable from it; the playlist only advertises the parts
1805 /// of a closed segment while all of them are still resident.
1806 pub fn build(self) -> Result<HlsOrigin, HlsOriginBuildError> {
1807 let target_duration_secs = self
1808 .target_duration_secs
1809 .ok_or(HlsOriginBuildError::MissingTargetDurationSecs)?;
1810 if !target_duration_secs.is_finite() || target_duration_secs <= 0.0 {
1811 return Err(HlsOriginBuildError::InvalidTargetDuration);
1812 }
1813 let window_segments = self
1814 .window_segments
1815 .ok_or(HlsOriginBuildError::MissingWindowSegments)?;
1816 if self.part_target_ms == Some(0) {
1817 return Err(HlsOriginBuildError::ZeroPartTarget);
1818 }
1819 if self
1820 .media_sequence_offset
1821 .checked_add(u64::from(u32::MAX))
1822 .is_none()
1823 {
1824 return Err(HlsOriginBuildError::MediaSequenceOffsetTooLarge);
1825 }
1826 if let Some(declared) = self.declared_bandwidth_bps
1827 && (declared == 0 || declared > u64::from(u32::MAX))
1828 {
1829 return Err(HlsOriginBuildError::InvalidDeclaredBandwidth);
1830 }
1831 if self.audio_sibling_origins.is_empty() {
1832 validate_audio_renditions(&self.audio_renditions)?;
1833 } else {
1834 validate_audio_renditions_relaxed(&self.audio_renditions)?;
1835 }
1836 validate_closed_caption_overrides(&self.closed_caption_overrides)?;
1837 validate_subtitle_renditions(&self.subtitle_renditions)?;
1838 if !self.subtitle_sibling_origins.is_empty() {
1839 if self.subtitle_sibling_origins.len() != self.subtitle_renditions.len() {
1840 return Err(HlsOriginBuildError::SubtitleSiblingCountMismatch);
1841 }
1842 let target = target_duration_secs.ceil();
1843 for sibling in &self.subtitle_sibling_origins {
1844 if sibling.container != Container::WebVtt {
1845 return Err(HlsOriginBuildError::SubtitleSiblingNotWebVtt);
1846 }
1847 if sibling.target_duration_secs.ceil() != target {
1848 return Err(HlsOriginBuildError::SubtitleSiblingTargetDurationMismatch);
1849 }
1850 }
1851 }
1852 let cursor = self.trunk.subscribe_segments();
1853 Ok(HlsOrigin {
1854 trunk: self.trunk,
1855 cursor: Mutex::new(cursor),
1856 drained_through: AtomicU64::new(UNDRAINED),
1857 drain_gate: Mutex::new(()),
1858 window: Mutex::new(Window::new(window_segments)),
1859 init: Mutex::new(InitStore::default()),
1860 codecs: Mutex::new(None),
1861 video_codecs: Mutex::new(None),
1862 video_range: Mutex::new(None),
1863 supplemental_codecs: Mutex::new(None),
1864 resolution: Mutex::new(None),
1865 frame_rate: Mutex::new(None),
1866 video_track_id: Mutex::new(None),
1867 target_duration_secs,
1868 container: self.container,
1869 part_target_ms: self.part_target_ms,
1870 part_target_override_ms: AtomicU32::new(0),
1871 media_sequence_offset: self.media_sequence_offset,
1872 instance: self.instance.unwrap_or_else(next_instance_token),
1873 arrival_clock: self.arrival_clock,
1874 arrival_origin: Mutex::new(None),
1875 declared_bandwidth_bps: self.declared_bandwidth_bps,
1876 audio_renditions: self.audio_renditions,
1877 audio_sibling_origins: self.audio_sibling_origins,
1878 subtitle_renditions: Mutex::new(self.subtitle_renditions),
1879 playlist_uri: self
1880 .playlist_uri
1881 .unwrap_or_else(|| "media.m3u8".to_string()),
1882 rendition_siblings: Mutex::new(self.rendition_siblings),
1883 track_specs: Mutex::new(Vec::new()),
1884 closed_captions_none: self.closed_captions_none,
1885 closed_caption_overrides: self.closed_caption_overrides,
1886 detected_closed_captions: Mutex::new(None),
1887 declared_closed_caption_languages: Mutex::new(Vec::new()),
1888 frozen_target_duration_secs: Mutex::new(None),
1889 target_duration_violations: AtomicU64::new(0),
1890 })
1891 }
1892}
1893
1894/// The LL-HLS origin [`ServedEgress`]: renders playlists and resolves
1895/// blocking-reload/part-availability requests for one stream, backed by a
1896/// shared [`Trunk`]. See this module's own doc for exactly what comes
1897/// straight from the `Trunk` and what needs the small synced `Window`.
1898pub struct HlsOrigin {
1899 trunk: Arc<Trunk>,
1900 /// This origin's **one** [`SegmentCursor`] — see [`Trunk::subscribe_segments`]'s
1901 /// own docs (and this crate's `media_plane::egress` module doc) for why a
1902 /// `ServedEgress` must never take one per request/peer.
1903 cursor: Mutex<SegmentCursor>,
1904 /// `last closed segment + 1` (`0` = none) at the end of the last full
1905 /// drain; [`Self::drain`] returns at once while the `Trunk` still shows
1906 /// the same head. [`UNDRAINED`] before the first drain.
1907 drained_through: AtomicU64,
1908 /// Serialises whole drains, so a concurrent caller waits for the window to
1909 /// be populated instead of rendering from a half-drained one.
1910 drain_gate: Mutex<()>,
1911 // Lock order (never the reverse): `drain_gate` -> `cursor` -> `window` -> `arrival_origin`
1912 // -> `init`; `codecs` is taken alone. No `Trunk` call is made while
1913 // `window` is held except inside `drain` (the cursor poll).
1914 window: Mutex<Window>,
1915 /// The fMP4 init segment — see this module's doc for why this, alone, is
1916 /// not answerable by any `Trunk` ring.
1917 init: Mutex<InitStore>,
1918 /// The `CODECS` value (RFC 6381, comma-joined) for the master playlist,
1919 /// known once track specs have been supplied; `None` = unknown, so the
1920 /// attribute is omitted rather than invented.
1921 codecs: Mutex<Option<String>>,
1922 /// `CODECS` value for the I-frame variant: the video-only codec string
1923 /// (the first video track by track_id), or `None` when unknown.
1924 video_codecs: Mutex<Option<String>>,
1925 /// `VIDEO-RANGE` for the master playlist, derived from the first video
1926 /// track (in track-id order) whose display carries a colour description;
1927 /// `None` = unknown / SDR, so the attribute is omitted rather than
1928 /// invented (RFC 8216bis §4.4.6.2).
1929 video_range: Mutex<Option<VideoRange>>,
1930 /// `SUPPLEMENTAL-CODECS` for the master playlist variants: the Dolby
1931 /// Vision codec and brand of a compatible-base-layer track.
1932 supplemental_codecs: Mutex<Option<String>>,
1933 resolution: Mutex<Option<(u32, u32)>>,
1934 /// `FRAME-RATE` for the master playlist, derived as the maximum across
1935 /// all video tracks. Emitted only when an H.264 track's SPS carries VUI
1936 /// `timing_info`; HEVC and codecs without VUI timing omit it.
1937 frame_rate: Mutex<Option<FrameRate>>,
1938 /// The first video track (in track-id order) supplied via
1939 /// [`Self::set_track_specs`], used to locate the video traf when
1940 /// computing an I-frame playlist's byte ranges. `None` until track specs
1941 /// are known (or when there is no video track).
1942 video_track_id: Mutex<Option<TrackId>>,
1943 target_duration_secs: f64,
1944 container: Container,
1945 /// `Some(part_target_ms)` enables LL-HLS; `None` renders classic HLS —
1946 /// no `#EXT-X-PART`/`#EXT-X-PART-INF`/`#EXT-X-SERVER-CONTROL`/
1947 /// `#EXT-X-PRELOAD-HINT` at all, orthogonal to [`Self::container`] (issue
1948 /// #873).
1949 part_target_ms: Option<u32>,
1950 /// Raises this origin's own `PART-TARGET` above `part_target_ms` (`0` =
1951 /// no override); see [`Self::set_part_target_ms`].
1952 part_target_override_ms: AtomicU32,
1953 /// Added to every shown Media Sequence Number (see
1954 /// [`HlsOriginBuilder::media_sequence_offset`]); validated at build so
1955 /// `offset + u32::MAX` fits a `u64`.
1956 media_sequence_offset: u64,
1957 /// See [`Self::instance`].
1958 instance: u64,
1959 /// Optional arrival wall-clock source (see
1960 /// [`HlsOriginBuilder::program_date_time_arrival`]). `None` => no arrival
1961 /// source.
1962 arrival_clock: Option<Arc<dyn Fn() -> i64 + Send + Sync>>,
1963 /// Recorded origin: `(segment timeline start_ns, wall epoch ms)`. Set
1964 /// once, the first time a segment drains into the window while an arrival
1965 /// clock is configured; never reset for the life of the origin.
1966 /// Lock order: `window` -> `arrival_origin` (arrival_origin is never
1967 /// taken before window).
1968 arrival_origin: Mutex<Option<(u64, i64)>>,
1969 /// Operator-declared peak bitrate (see
1970 /// [`HlsOriginBuilder::declared_bandwidth_bps`]) — `None` means none.
1971 declared_bandwidth_bps: Option<u64>,
1972 /// Declared audio renditions (issue #1428 slice 1).
1973 audio_renditions: Vec<AudioRenditionDecl>,
1974 /// Sibling audio origins (issue #1428 slice 3).
1975 audio_sibling_origins: Vec<Arc<HlsOrigin>>,
1976 /// Declared WebVTT subtitle renditions.
1977 subtitle_renditions: Mutex<Vec<SubtitleRenditionDecl>>,
1978 /// URI of this origin's own media playlist.
1979 playlist_uri: String,
1980 /// Sibling renditions whose live edge this origin reports.
1981 rendition_siblings: Mutex<Vec<RenditionSibling>>,
1982 /// Stored track specs (issue #1428 slice 3), for audio metadata derivation.
1983 track_specs: Mutex<Vec<transmux::TrackSpec>>,
1984 /// Emit `CLOSED-CAPTIONS=NONE` (issue #1428 slice 1).
1985 closed_captions_none: bool,
1986 closed_caption_overrides: Vec<ClosedCaptionDecl>,
1987 /// Caption streams the video has been seen to carry, in advertising
1988 /// order. Taken alone (never while `window` or `codecs` is held).
1989 detected_closed_captions: Mutex<Option<Vec<broadcast_hls::InstreamId>>>,
1990 /// BCP 47 `LANGUAGE` the stream itself declares per caption stream; a
1991 /// [`ClosedCaptionDecl`] language wins over it.
1992 declared_closed_caption_languages: Mutex<Vec<(broadcast_hls::InstreamId, String)>>,
1993 /// Frozen `EXT-X-TARGETDURATION` (seconds, pre-rounding) captured the
1994 /// first time a playlist is rendered with at least one segment in the
1995 /// window: RFC 8216 §4.3.3.1 forbids the tag changing once published,
1996 /// so later (possibly longer) segments never move it. `None` until then.
1997 frozen_target_duration_secs: Mutex<Option<f64>>,
1998 /// Number of segments whose (rounded) duration exceeded the frozen
1999 /// `EXT-X-TARGETDURATION` after first publication — a conformance
2000 /// degradation counter (RFC 8216 §4.3.3.1). Incremented once per such
2001 /// segment at drain time.
2002 target_duration_violations: AtomicU64,
2003}
2004
2005impl HlsOrigin {
2006 /// Start building an [`HlsOrigin`] over `trunk` — see [`HlsOriginBuilder`]
2007 /// for the required fields (`target_duration_secs`/`window_segments`),
2008 /// the container choice, and how to opt into LL-HLS.
2009 pub fn builder(trunk: Arc<Trunk>) -> HlsOriginBuilder {
2010 HlsOriginBuilder::new(trunk)
2011 }
2012
2013 /// Store the init segment bytes — see this module's doc for why an init
2014 /// segment is not something any `Trunk` ring holds.
2015 ///
2016 /// **Documented no-op under [`Container::MpegTs`]**: this origin's
2017 /// `MpegTs` grammar has no init resource and never emits `#EXT-X-MAP`
2018 /// by default (see [`Container`]'s own doc), so bytes stored here are
2019 /// never advertised or served in that mode. The method stays callable
2020 /// regardless of container so a caller sharing one code path across
2021 /// both (e.g. a segmenter that always calls `set_init` once available)
2022 /// does not need to branch on which container it configured.
2023 ///
2024 /// Under [`Container::Fmp4`] the init segment is also parsed for its
2025 /// tracks, from which the master playlist's `CODECS` is derived (see
2026 /// [`Self::master_playlist`]); bytes that do not parse (or a track whose
2027 /// codec string cannot be built) make `CODECS` unknown (omitted, and any
2028 /// value from an earlier init is dropped), never guessed at.
2029 pub fn set_init(&self, bytes: impl Into<Bytes>) {
2030 let bytes = bytes.into();
2031 if self.container == Container::Fmp4 {
2032 let derived = transmux::Fmp4Demux::new()
2033 .unpackage(bytes.as_ref())
2034 .ok()
2035 .map(|media| {
2036 media
2037 .tracks
2038 .into_iter()
2039 .map(|t| t.spec)
2040 .collect::<Vec<transmux::TrackSpec>>()
2041 })
2042 .map(|specs| self.with_known_es_info(specs))
2043 .and_then(|specs| self.set_track_specs(&specs).ok());
2044 if derived.is_none() {
2045 // A new init that cannot be read must not leave the previous
2046 // init's codecs (or VIDEO-RANGE) advertised.
2047 *locked(&self.codecs) = None;
2048 *locked(&self.video_codecs) = None;
2049 *locked(&self.video_range) = None;
2050 *locked(&self.supplemental_codecs) = None;
2051 *locked(&self.resolution) = None;
2052 *locked(&self.frame_rate) = None;
2053 *locked(&self.video_track_id) = None;
2054 *locked(&self.track_specs) = Vec::new();
2055 }
2056 }
2057 // The first segment not yet closed is the first one cut against
2058 // these bytes: call this before publishing it.
2059 let next_seq = self
2060 .trunk
2061 .last_closed_segment()
2062 .unwrap_or(SegmentSeq::new(0))
2063 .saturating_next();
2064 locked(&self.init).set(bytes, next_seq);
2065 }
2066
2067 /// Tell the origin which tracks it serves, so the master playlist can
2068 /// carry `CODECS` (RFC 8216 §4.3.4.2: every `EXT-X-STREAM-INF` SHOULD) —
2069 /// derived from each track's codec config with the RFC 6381 builders
2070 /// `transmux` uses for DASH. Opaque data/subtitle tracks have no codec
2071 /// string and are left out. [`Self::set_init`] does this itself for
2072 /// fMP4; call this for a [`Container::MpegTs`] origin, whose segments
2073 /// carry no init segment to read the codecs from.
2074 ///
2075 /// # Errors
2076 /// The `transmux` error for a track whose codec string cannot be built
2077 /// (for example a malformed AAC `esds`); `CODECS` is then left as it was.
2078 pub fn set_track_specs(&self, specs: &[transmux::TrackSpec]) -> Result<(), transmux::Error> {
2079 let mut ordered: Vec<&transmux::TrackSpec> = specs.iter().collect();
2080 ordered.sort_by_key(|spec| spec.track_id);
2081 let stored_track_specs: Vec<transmux::TrackSpec> =
2082 ordered.iter().map(|s| (*s).clone()).collect();
2083 *locked(&self.video_track_id) = ordered
2084 .iter()
2085 .find(|spec| is_video_codec(&spec.config))
2086 .map(|spec| spec.track_id);
2087 let mut codecs: Vec<String> = Vec::new();
2088 // Video-only codec string for the I-frame variant: the first video
2089 // track by track_id (RFC 8216bis §4.4.6.3 / Apple 9.3).
2090 let mut video_codecs: Option<String> = None;
2091 let mut video_codecs_decided = false;
2092 // VIDEO-RANGE is derived from the first video track (in track-id
2093 // order) whose display carries a colour description (RFC 8216bis
2094 // §4.4.6.2). `None` when no such track exists OR its transfer code
2095 // has no defined mapping (attribute omitted).
2096 let mut video_range: Option<VideoRange> = None;
2097 let mut video_range_decided = false;
2098 let mut resolution: Option<(u32, u32)> = None;
2099 let mut resolution_decided = false;
2100 let mut frame_rate: Option<FrameRate> = None;
2101 let mut supplemental_codecs: Option<String> = None;
2102 for spec in ordered {
2103 let dolby_vision = if !video_codecs_decided && is_video_codec(&spec.config) {
2104 dolby_vision_hls(spec)
2105 } else {
2106 None
2107 };
2108 match transmux::rfc6381_codec_string(&spec.config) {
2109 Ok(codec) => {
2110 codecs.push(dolby_vision.as_ref().map_or(codec, |dv| dv.codecs.clone()))
2111 }
2112 Err(transmux::Error::UnsupportedCodec { .. }) => {}
2113 Err(other) => return Err(other),
2114 }
2115 if !video_codecs_decided && is_video_codec(&spec.config) {
2116 match transmux::rfc6381_codec_string(&spec.config) {
2117 Ok(codec) => {
2118 video_codecs =
2119 Some(dolby_vision.as_ref().map_or(codec, |dv| dv.codecs.clone()))
2120 }
2121 Err(transmux::Error::UnsupportedCodec { .. }) => {}
2122 Err(other) => return Err(other),
2123 }
2124 video_codecs_decided = true;
2125 if let Some(dv) = dolby_vision {
2126 supplemental_codecs = dv.supplemental_codecs;
2127 video_range = Some(dv.video_range);
2128 video_range_decided = true;
2129 } else {
2130 supplemental_codecs = hdr10plus_supplemental_codecs(spec);
2131 }
2132 }
2133 if !video_range_decided
2134 && let Some(display) = &spec.display
2135 && let Some(signal) = &display.video_signal
2136 && let Some(colour) = &signal.colour
2137 {
2138 video_range = VideoRange::from_transfer_characteristics(colour.transfer.code());
2139 video_range_decided = true;
2140 }
2141 if !resolution_decided
2142 && let Some((w, h)) = match &spec.config {
2143 transmux::CodecConfig::Avc { width, height, .. } => Some((*width, *height)),
2144 transmux::CodecConfig::Hevc { width, height, .. } => Some((*width, *height)),
2145 _ => None,
2146 }
2147 {
2148 resolution = Some(transmux::presentation_size(spec.display.as_ref(), w, h));
2149 resolution_decided = true;
2150 }
2151 // FRAME-RATE is the maximum across all video tracks (RFC 8216bis
2152 // §4.4.6.2). Source is H.264 SPS VUI timing only; HEVC and other
2153 // codecs yield None and never contribute.
2154 if let Some(mh) = transmux::nominal_frame_rate_millihertz(&spec.config) {
2155 let candidate = FrameRate::from_millihertz(mh);
2156 frame_rate = Some(match frame_rate {
2157 Some(existing) if existing.millihertz() >= mh => existing,
2158 _ => candidate,
2159 });
2160 }
2161 }
2162 *locked(&self.codecs) = (!codecs.is_empty()).then(|| codecs.join(","));
2163 *locked(&self.video_codecs) = video_codecs;
2164 *locked(&self.video_range) = video_range;
2165 *locked(&self.supplemental_codecs) = supplemental_codecs;
2166 *locked(&self.resolution) = resolution;
2167 *locked(&self.frame_rate) = frame_rate;
2168 *locked(&self.track_specs) = stored_track_specs;
2169 Ok(())
2170 }
2171
2172 /// An init segment carries no PMT `ES_info` and no HDR10+ marker (the
2173 /// dynamic metadata lives in the samples); keep what an earlier
2174 /// [`Self::set_track_specs`] recorded for the same track so the language
2175 /// and role of a split audio rendition and the HDR10+ flag survive the init
2176 /// re-derivation.
2177 fn with_known_es_info(&self, mut specs: Vec<transmux::TrackSpec>) -> Vec<transmux::TrackSpec> {
2178 let known = locked(&self.track_specs);
2179 for spec in &mut specs {
2180 if let Some(prev) = known.iter().find(|k| k.track_id == spec.track_id) {
2181 spec.hdr10plus |= prev.hdr10plus;
2182 }
2183 if !spec.es_info_descriptors.is_empty() {
2184 continue;
2185 }
2186 if let Some(prev) = known.iter().find(|k| k.track_id == spec.track_id) {
2187 spec.source_pid = prev.source_pid;
2188 spec.es_info_descriptors = prev.es_info_descriptors.clone();
2189 }
2190 }
2191 specs
2192 }
2193
2194 /// One `Window` lock, one consistent snapshot: peak bits/sec, total bytes
2195 /// and total duration (nanoseconds).
2196 fn window_snapshot(&self) -> (u64, u64, u128) {
2197 let w = locked(&self.window);
2198 (w.peak_bandwidth_bps, w.total_bytes, w.total_duration_nanos)
2199 }
2200
2201 /// Record the caption streams the video is now known to carry. The next
2202 /// master playlist advertises exactly these, one
2203 /// `EXT-X-MEDIA TYPE=CLOSED-CAPTIONS` per stream in group
2204 /// [`CLOSED_CAPTIONS_GROUP_ID`], and every variant references the group.
2205 /// An empty list means the video was scanned and carries none: the master
2206 /// then states `CLOSED-CAPTIONS=NONE` on every variant (RFC 8216bis
2207 /// section 4.4.6.2). Until this is called the master claims nothing either
2208 /// way, unless [`HlsOriginBuilder::closed_captions_none`] is set.
2209 pub fn set_detected_closed_captions(&self, detected: Vec<broadcast_hls::InstreamId>) {
2210 *locked(&self.detected_closed_captions) = Some(detected);
2211 }
2212
2213 /// Record the BCP 47 `LANGUAGE` the stream declares for each caption
2214 /// stream (the ATSC `caption_service_descriptor` in the PMT). A rendition
2215 /// takes it unless a [`ClosedCaptionDecl`] override names a language.
2216 pub fn set_declared_closed_caption_languages(
2217 &self,
2218 languages: Vec<(broadcast_hls::InstreamId, String)>,
2219 ) {
2220 *locked(&self.declared_closed_caption_languages) = languages;
2221 }
2222
2223 /// Replace the declared WebVTT subtitle renditions after construction, for
2224 /// a host that learns them only once the stream is seen (for example from
2225 /// a teletext descriptor). The next master playlist advertises exactly
2226 /// these in group [`SUBTITLES_GROUP_ID`]; an empty list removes the group.
2227 ///
2228 /// # Errors
2229 /// The same [`HlsOriginBuildError`] variants as
2230 /// [`HlsOriginBuilder::subtitle_renditions`] at `build()`; the previous
2231 /// declarations are kept on error.
2232 pub fn set_subtitle_renditions(
2233 &self,
2234 decls: Vec<SubtitleRenditionDecl>,
2235 ) -> Result<(), HlsOriginBuildError> {
2236 validate_subtitle_renditions(&decls)?;
2237 *locked(&self.subtitle_renditions) = decls;
2238 Ok(())
2239 }
2240
2241 /// The master playlist for this origin, pointing at `media_playlist_name`:
2242 /// one `#EXT-X-STREAM-INF` whose `BANDWIDTH` is the highest segment
2243 /// bitrate this origin has published (RFC 8216 §4.3.4.2: an upper bound
2244 /// of the peak segment bitrate, measured as segment bytes over its
2245 /// duration, rounded up) or the operator-declared bandwidth, whichever is
2246 /// larger. Returns [`HlsMasterError::BandwidthUnknown`] until a segment
2247 /// with a measurable duration has closed and no declared bandwidth was
2248 /// configured.
2249 ///
2250 /// # Errors
2251 /// [`HlsMasterError::BandwidthUnknown`] when no measurable segment has
2252 /// been published and no declared bandwidth was configured;
2253 /// [`HlsMasterError::BandwidthOverflow`] when the measured `BANDWIDTH`
2254 /// does not fit the `u32` of a master playlist variant;
2255 /// [`HlsMasterError::Render`] when the playlist cannot be rendered.
2256 pub fn master_playlist(&self, media_playlist_name: &str) -> Result<String, HlsMasterError> {
2257 self.master_playlist_inner(media_playlist_name, None)
2258 }
2259
2260 /// Like [`Self::master_playlist`], but additionally advertises an
2261 /// `#EXT-X-I-FRAME-STREAM-INF` variant pointing at
2262 /// `iframe_playlist_name` when an I-frame playlist is currently
2263 /// available for this origin (fMP4, a video track, and at least one
2264 /// usable entry with no mid-stream gap). The plain variant is unchanged.
2265 ///
2266 /// # Errors
2267 /// As for [`Self::master_playlist`].
2268 pub fn master_playlist_with_iframes(
2269 &self,
2270 media_playlist_name: &str,
2271 iframe_playlist_name: &str,
2272 ) -> Result<String, HlsMasterError> {
2273 self.master_playlist_inner(media_playlist_name, Some(iframe_playlist_name))
2274 }
2275
2276 fn master_playlist_inner(
2277 &self,
2278 media_playlist_name: &str,
2279 iframe_playlist_name: Option<&str>,
2280 ) -> Result<String, HlsMasterError> {
2281 self.drain();
2282 let (video_peak, video_bytes, video_duration) = {
2283 let window = locked(&self.window);
2284 (
2285 window.peak_bandwidth_bps,
2286 window.total_bytes,
2287 window.total_duration_nanos,
2288 )
2289 };
2290 let declared = self.declared_bandwidth_bps.unwrap_or(0);
2291 let video_part_peak = video_peak.max(declared);
2292 if video_part_peak == 0 {
2293 return Err(HlsMasterError::BandwidthUnknown);
2294 }
2295 let video_avg = average_bandwidth_bps(video_bytes, video_duration);
2296 let has_siblings = !self.audio_sibling_origins.is_empty();
2297 let primary_codecs = locked(&self.codecs).clone();
2298 let mut extra_attrs: Vec<(String, broadcast_hls::AttrValue)> = Vec::new();
2299 if let Some(range) = *locked(&self.video_range)
2300 && matches!(range, VideoRange::Hlg | VideoRange::Pq)
2301 {
2302 extra_attrs.push(range.attr());
2303 }
2304 if let Some(supplemental) = locked(&self.supplemental_codecs).clone() {
2305 extra_attrs.push((
2306 String::from(SUPPLEMENTAL_CODECS_ATTR),
2307 broadcast_hls::AttrValue::for_attr(SUPPLEMENTAL_CODECS_ATTR, supplemental)
2308 .map_err(HlsMasterError::Render)?,
2309 ));
2310 }
2311
2312 for decl in &self.audio_renditions {
2313 if decl.uri == media_playlist_name {
2314 return Err(HlsMasterError::AudioRenditionUriEqualsMediaPlaylist);
2315 }
2316 }
2317
2318 let media: Vec<MediaRendition> = self
2319 .audio_renditions
2320 .iter()
2321 .enumerate()
2322 .map(|(i, decl)| {
2323 let sibling = self.audio_sibling_origins.get(i);
2324 let audio_spec = sibling.and_then(|s| {
2325 locked(&s.track_specs)
2326 .iter()
2327 .find(|t| !is_video_codec(&t.config))
2328 .cloned()
2329 });
2330 let meta = audio_spec.as_ref().map(|t| t.audio_metadata());
2331 let role = meta
2332 .as_ref()
2333 .map(|m| m.role)
2334 .unwrap_or(transmux::AudioRole::Unknown);
2335 let derived_iso = meta
2336 .as_ref()
2337 .and_then(|m| m.languages.first().map(|l| l.language.clone()))
2338 .unwrap_or_else(|| "und".to_string());
2339 // LANGUAGE must be a BCP 47 tag: canonicalise the ISO 639-2
2340 // code from the PMT. NAME keeps the source code as before.
2341 let derived_lang =
2342 bcp47_language(&derived_iso).unwrap_or_else(|| "und".to_string());
2343 let language = if decl.language.is_empty() {
2344 derived_lang.clone()
2345 } else {
2346 decl.language.clone()
2347 };
2348 let name_source = if decl.language.is_empty() {
2349 derived_iso.clone()
2350 } else {
2351 decl.language.clone()
2352 };
2353 let name = if decl.name.is_empty() {
2354 if role == transmux::AudioRole::AudioDescription {
2355 format!("{} (AD)", name_source)
2356 } else {
2357 name_source.clone()
2358 }
2359 } else {
2360 decl.name.clone()
2361 };
2362 let characteristics = if decl.characteristics.is_empty() {
2363 if role == transmux::AudioRole::AudioDescription {
2364 vec!["public.accessibility.describes-video".to_string()]
2365 } else {
2366 Vec::new()
2367 }
2368 } else {
2369 decl.characteristics.clone()
2370 };
2371 let channels = decl.channels.clone().or_else(|| {
2372 audio_spec
2373 .as_ref()
2374 .and_then(|t| channel_count_string(&t.config))
2375 });
2376 let mut m = MediaRendition::default();
2377 m.media_type = MediaType::Audio;
2378 m.uri = Some(decl.uri.clone());
2379 m.group_id = decl.group_id.clone();
2380 m.language = Some(language);
2381 m.name = name;
2382 m.default = decl.default;
2383 m.autoselect = decl.autoselect;
2384 m.channels = channels;
2385 m.characteristics = characteristics;
2386 m
2387 })
2388 .collect();
2389 let mut media = dedupe_rendition_names(media);
2390 media.extend(self.subtitle_media_renditions());
2391 let subtitles_group =
2392 (!locked(&self.subtitle_renditions).is_empty()).then(|| SUBTITLES_GROUP_ID.to_string());
2393 let with_wvtt = |codecs: Option<String>| match (&subtitles_group, codecs) {
2394 (Some(_), Some(c)) => Some(format!("{c},{WEBVTT_CODEC}")),
2395 (_, c) => c,
2396 };
2397 let detected_captions = locked(&self.detected_closed_captions).clone();
2398 let scanned_without_captions = detected_captions.as_ref().is_some_and(Vec::is_empty);
2399 let caption_media =
2400 self.closed_caption_renditions(detected_captions.as_deref().unwrap_or_default());
2401 let captions_advertised = !caption_media.is_empty();
2402 media.extend(caption_media);
2403
2404 let audio_groups: Vec<String> = {
2405 let mut g: Vec<String> = Vec::new();
2406 for decl in &self.audio_renditions {
2407 if !g.contains(&decl.group_id) {
2408 g.push(decl.group_id.clone());
2409 }
2410 }
2411 g
2412 };
2413
2414 let closed_captions = if captions_advertised {
2415 Some(ClosedCaptionsRef::Group(
2416 CLOSED_CAPTIONS_GROUP_ID.to_string(),
2417 ))
2418 } else if self.closed_captions_none || scanned_without_captions {
2419 Some(ClosedCaptionsRef::None)
2420 } else {
2421 None
2422 };
2423
2424 let mut variants: Vec<Variant> = Vec::new();
2425 if audio_groups.is_empty() {
2426 let bw = video_part_peak;
2427 let bw = u32::try_from(bw)
2428 .map_err(|_| HlsMasterError::BandwidthOverflow { bandwidth: bw })?;
2429 let avg = video_avg.map(|a| a.min(bw));
2430 variants.push(Variant {
2431 bandwidth: bw,
2432 average_bandwidth: avg,
2433 codecs: with_wvtt(primary_codecs.clone()),
2434 uri: media_playlist_name.to_string(),
2435 audio: None,
2436 video: None,
2437 subtitles: subtitles_group.clone(),
2438 closed_captions,
2439 extra_attrs: extra_attrs.clone(),
2440 resolution: *locked(&self.resolution),
2441 frame_rate: self.effective_frame_rate(),
2442 });
2443 } else {
2444 for group in &audio_groups {
2445 // Audio peak/average max over this group's sibling origins.
2446 let mut audio_peak_max = 0u64;
2447 let mut audio_avg_max = 0u32;
2448 let mut audio_codecs: Vec<String> = Vec::new();
2449 for (i, decl) in self.audio_renditions.iter().enumerate() {
2450 if decl.group_id != *group {
2451 continue;
2452 }
2453 if let Some(sib) = self.audio_sibling_origins.get(i) {
2454 sib.drain();
2455 let (peak, tb, td) = sib.window_snapshot();
2456 audio_peak_max = audio_peak_max.max(peak);
2457 if let Some(avg) = average_bandwidth_bps(tb, td) {
2458 audio_avg_max = audio_avg_max.max(avg);
2459 }
2460 if let Some(codec) = locked(&sib.codecs).clone()
2461 && !audio_codecs.contains(&codec)
2462 {
2463 audio_codecs.push(codec);
2464 }
2465 }
2466 }
2467 let group_peak = video_part_peak.saturating_add(audio_peak_max);
2468 let group_bw =
2469 u32::try_from(group_peak).map_err(|_| HlsMasterError::BandwidthOverflow {
2470 bandwidth: group_peak,
2471 })?;
2472 let group_avg_candidate = video_avg.unwrap_or(0) as u64 + u64::from(audio_avg_max);
2473 let group_avg = if group_avg_candidate == 0 {
2474 None
2475 } else {
2476 Some(
2477 u32::try_from(group_avg_candidate.min(u64::from(group_bw)))
2478 .unwrap_or(group_bw),
2479 )
2480 };
2481 let group_codecs = if has_siblings {
2482 let mut all: Vec<String> = primary_codecs.iter().cloned().collect();
2483 all.extend(audio_codecs);
2484 if all.is_empty() {
2485 None
2486 } else {
2487 Some(all.join(","))
2488 }
2489 } else {
2490 primary_codecs.clone()
2491 };
2492 variants.push(Variant {
2493 bandwidth: group_bw,
2494 average_bandwidth: group_avg,
2495 codecs: with_wvtt(group_codecs),
2496 uri: media_playlist_name.to_string(),
2497 audio: Some(group.clone()),
2498 video: None,
2499 subtitles: subtitles_group.clone(),
2500 closed_captions: closed_captions.clone(),
2501 extra_attrs: extra_attrs.clone(),
2502 resolution: *locked(&self.resolution),
2503 frame_rate: self.effective_frame_rate(),
2504 });
2505 }
2506 }
2507
2508 // I-frame variant, only when the I-frame playlist is available.
2509 let mut iframe_variants: Vec<broadcast_hls::IFrameVariant> = Vec::new();
2510 if let Some(iframe_name) = iframe_playlist_name
2511 && self.iframe_playlist_available()
2512 {
2513 let (iframe_measured, iframe_total_bytes, iframe_total_duration_nanos) = {
2514 let window = locked(&self.window);
2515 (
2516 window.iframe_peak_bandwidth_bps,
2517 window.iframe_total_bytes,
2518 window.iframe_total_duration_nanos,
2519 )
2520 };
2521 if iframe_measured != 0 {
2522 let iframe_bandwidth = u32::try_from(iframe_measured).map_err(|_| {
2523 HlsMasterError::BandwidthOverflow {
2524 bandwidth: iframe_measured,
2525 }
2526 })?;
2527 let iframe_average =
2528 average_bandwidth_bps(iframe_total_bytes, iframe_total_duration_nanos)
2529 .map(|avg| avg.min(iframe_bandwidth));
2530 let mut iframe_extra: Vec<(String, broadcast_hls::AttrValue)> = Vec::new();
2531 if let Some(range) = *locked(&self.video_range)
2532 && matches!(range, VideoRange::Hlg | VideoRange::Pq)
2533 {
2534 iframe_extra.push(range.attr());
2535 }
2536 iframe_variants.push(broadcast_hls::IFrameVariant {
2537 bandwidth: iframe_bandwidth,
2538 average_bandwidth: iframe_average,
2539 codecs: locked(&self.video_codecs).clone(),
2540 resolution: *locked(&self.resolution),
2541 uri: iframe_name.to_string(),
2542 extra_attrs: iframe_extra,
2543 });
2544 }
2545 }
2546
2547 let master = MasterPlaylist {
2548 media,
2549 variants,
2550 iframe_variants,
2551 ..Default::default()
2552 };
2553 master.to_m3u8().map_err(HlsMasterError::Render)
2554 }
2555
2556 fn subtitle_media_renditions(&self) -> Vec<MediaRendition> {
2557 locked(&self.subtitle_renditions)
2558 .iter()
2559 .map(|decl| {
2560 let mut m = MediaRendition::default();
2561 m.media_type = MediaType::Subtitles;
2562 m.uri = Some(decl.uri.clone());
2563 m.group_id = SUBTITLES_GROUP_ID.to_string();
2564 m.language = Some(decl.language.clone());
2565 m.name = decl.name.clone();
2566 m.default = decl.default;
2567 m.autoselect = decl.effective_autoselect();
2568 m.forced = decl.forced;
2569 if decl.hearing_impaired {
2570 m.characteristics = HEARING_IMPAIRED_CHARACTERISTICS
2571 .iter()
2572 .map(|c| (*c).to_string())
2573 .collect();
2574 }
2575 m
2576 })
2577 .collect()
2578 }
2579
2580 fn closed_caption_renditions(
2581 &self,
2582 detected: &[broadcast_hls::InstreamId],
2583 ) -> Vec<MediaRendition> {
2584 let stream_languages = locked(&self.declared_closed_caption_languages).clone();
2585 let renditions = detected
2586 .iter()
2587 .map(|id| {
2588 let rendered = id.render();
2589 let declared = self.closed_caption_overrides.iter().find(|d| {
2590 broadcast_hls::InstreamId::parse_closed_captions(&d.instream_id).as_ref()
2591 == Ok(id)
2592 });
2593 let pick = |value: Option<&String>, default: &str| match value {
2594 Some(v) if !v.is_empty() => v.clone(),
2595 _ => default.to_string(),
2596 };
2597 let mut m = MediaRendition::default();
2598 m.media_type = MediaType::ClosedCaptions;
2599 m.group_id = CLOSED_CAPTIONS_GROUP_ID.to_string();
2600 m.name = pick(declared.map(|d| &d.name), &rendered);
2601 m.language = declared
2602 .map(|d| d.language.clone())
2603 .filter(|l| !l.is_empty())
2604 .or_else(|| {
2605 stream_languages
2606 .iter()
2607 .find(|(declared_id, _)| declared_id == id)
2608 .map(|(_, language)| language.clone())
2609 });
2610 m.autoselect = false;
2611 m.instream_id = Some(id.clone());
2612 m
2613 })
2614 .collect();
2615 dedupe_rendition_names(renditions)
2616 }
2617
2618 fn iframe_playlist_available(&self) -> bool {
2619 if self.container != Container::Fmp4 {
2620 return false;
2621 }
2622 if locked(&self.video_track_id).is_none() {
2623 return false;
2624 }
2625 let window = self.window_view();
2626 self.iframe_entries(&window).is_some()
2627 }
2628
2629 /// Shared D3' rule: collect I-frame entries, allowing leading `None`s but
2630 /// rejecting a `None` after the first entry. Returns `None` when there is
2631 /// no usable entry (or a gap occurred).
2632 fn iframe_entries(&self, window: &WindowView) -> Option<Vec<(WindowSegment, u64)>> {
2633 let mut entries: Vec<(WindowSegment, u64)> = Vec::new();
2634 let mut seen_entry = false;
2635 for s in window.segments.iter() {
2636 match s.iframe_len {
2637 Some(len) => {
2638 seen_entry = true;
2639 entries.push((s.clone(), len));
2640 }
2641 None => {
2642 if seen_entry {
2643 return None;
2644 }
2645 }
2646 }
2647 }
2648 if entries.is_empty() {
2649 None
2650 } else {
2651 Some(entries)
2652 }
2653 }
2654
2655 /// This origin's instance token: a number that differs for every origin
2656 /// built, in this process and across restarts (see
2657 /// [`HlsOriginBuilder::instance`]), and is part of **every resource name
2658 /// served `immutable`** — `init-{track}-{instance}-{generation}.mp4`,
2659 /// `seg-{track}-{instance}-{msn}.{ext}`,
2660 /// `part-{track}-{instance}-{msn}.{idx}.{ext}`. A name therefore maps to
2661 /// one origin's bytes for ever: a reconnect (new `Trunk`, numbers
2662 /// restarting, possibly a different SPS/ASC), a process restart with no
2663 /// DVR, a replacement origin reusing the open segment's number — none can
2664 /// put new bytes under a name a cache may already hold. Playlists name
2665 /// these URIs, so players are unaffected. A request carrying another
2666 /// instance's token is `NotFound`. The unnamed legacy forms
2667 /// (`seg-{track}-{msn}`, `part-{track}-{msn}.{idx}`, used by DASH/Smooth
2668 /// templates that cannot carry the token) still resolve but are
2669 /// [`CachePolicy::NoCache`]; the bare `init-{track}.mp4` is the current
2670 /// init, also [`CachePolicy::NoCache`].
2671 pub fn instance(&self) -> u64 {
2672 self.instance
2673 }
2674
2675 /// The wire name of this origin's I-frame playlist for `track_id`.
2676 pub fn iframe_playlist_name(&self, track_id: TrackId) -> String {
2677 format!("iframe-{}.m3u8", track_id)
2678 }
2679
2680 /// The wire name of init generation `generation` for `track_id`.
2681 pub fn init_name(&self, track_id: TrackId, generation: Generation) -> String {
2682 format!("init-{track_id}-{}-{generation}.mp4", self.instance)
2683 }
2684
2685 fn segment_name(&self, track_id: TrackId, seq: SegmentSeq, ext: &str) -> String {
2686 format!(
2687 "seg-{track_id}-{}-{}.{ext}",
2688 self.instance,
2689 self.public_msn(seq)
2690 )
2691 }
2692
2693 /// The wire name of the I-frame (trick-play) resource for `track_id`'s
2694 /// segment `seq` (issue #1422): `iframe-seg-{track}-{instance}-{seq}.{ext}`,
2695 /// the same grammar as `segment_name` but under the
2696 /// [`IFRAME_SEGMENT_PREFIX`] prefix, so the bytes it resolves to are
2697 /// always the stored segment bytes and never the `emsg`-spliced ones.
2698 pub fn iframe_segment_name(&self, track_id: TrackId, seq: SegmentSeq, ext: &str) -> String {
2699 format!(
2700 "iframe-seg-{track_id}-{}-{}.{ext}",
2701 self.instance,
2702 self.public_msn(seq)
2703 )
2704 }
2705
2706 fn part_name(&self, track_id: TrackId, seq: SegmentSeq, idx: PartIndex, ext: &str) -> String {
2707 format!(
2708 "part-{track_id}-{}-{}.{idx}.{ext}",
2709 self.instance,
2710 self.public_msn(seq)
2711 )
2712 }
2713
2714 /// Playlist-relative reference to a resource served at the stream root:
2715 /// a playlist in a subdirectory (`audio/a0.m3u8`) must climb back to it
2716 /// (RFC 8216bis §4.1, URIs resolve against the playlist's URI).
2717 fn root_relative(&self, name: String) -> String {
2718 let depth = self.playlist_uri.rfind('/').map_or(0, |i| {
2719 self.playlist_uri[..i]
2720 .split('/')
2721 .filter(|s| !s.is_empty())
2722 .count()
2723 });
2724 format!("{}{name}", "../".repeat(depth))
2725 }
2726
2727 fn init_uri(&self, track_id: TrackId, generation: Generation) -> String {
2728 self.root_relative(self.init_name(track_id, generation))
2729 }
2730
2731 fn segment_uri(&self, track_id: TrackId, seq: SegmentSeq, ext: &str) -> String {
2732 self.root_relative(self.segment_name(track_id, seq, ext))
2733 }
2734
2735 fn iframe_segment_uri(&self, track_id: TrackId, seq: SegmentSeq, ext: &str) -> String {
2736 self.root_relative(self.iframe_segment_name(track_id, seq, ext))
2737 }
2738
2739 fn part_uri(&self, track_id: TrackId, seq: SegmentSeq, idx: PartIndex, ext: &str) -> String {
2740 self.root_relative(self.part_name(track_id, seq, idx, ext))
2741 }
2742
2743 /// The offset added to every Media Sequence Number this origin shows (see
2744 /// [`HlsOriginBuilder::media_sequence_offset`]): the number in a segment's
2745 /// URI/playlist is this plus its origin sequence number
2746 /// ([`ClosedSegment::sequence_number`]).
2747 pub fn media_sequence_offset(&self) -> u64 {
2748 self.media_sequence_offset
2749 }
2750
2751 /// The Media Sequence Number the next closed segment will be shown with
2752 /// (last closed segment + 1, plus the configured offset).
2753 ///
2754 /// A fresh [`Trunk`] numbers its segments from `1` again, so an origin
2755 /// replacing this one over a new `Trunk` (a reconnect) must pass
2756 /// `next_media_sequence()` to
2757 /// [`HlsOriginBuilder::media_sequence_offset`]: the new origin's first
2758 /// segment (sequence number one) is then shown one above this value — the
2759 /// number this origin's *open* segment was shown with is skipped, since
2760 /// its parts were already served — and the Media Sequence Number clients
2761 /// see keeps increasing (RFC 8216bis §6.2.2).
2762 pub fn next_media_sequence(&self) -> u64 {
2763 self.public_msn(
2764 self.trunk
2765 .last_closed_segment()
2766 .unwrap_or(SegmentSeq::new(0))
2767 .saturating_next(),
2768 )
2769 }
2770
2771 /// HTTP Priority signalling for `request` (RFC 8216bis §6.2.6): the
2772 /// urgency the adapter should attach to the response. Playlists get
2773 /// urgency 1, parts and segments urgency 2.
2774 pub fn http_priority(&self, request: &HlsRequest) -> HttpPriority {
2775 match request {
2776 HlsRequest::Playlist { .. } | HlsRequest::IFramePlaylist { .. } => {
2777 HttpPriority::PLAYLIST
2778 }
2779 HlsRequest::Resource { name } => {
2780 let _ = parse_part(name, self.container);
2781 HttpPriority::SEGMENT
2782 }
2783 }
2784 }
2785
2786 /// The Media Sequence Number shown for origin sequence number `seq`.
2787 fn public_msn(&self, seq: SegmentSeq) -> u64 {
2788 // `media_sequence_offset + u32::MAX` fits a `u64` (checked in `build`).
2789 self.media_sequence_offset
2790 .saturating_add(u64::from(seq.value()))
2791 }
2792
2793 /// The origin sequence number a shown Media Sequence Number refers to, or
2794 /// `None` when it lies outside the `u32` range this origin numbers in.
2795 fn internal_seq(&self, public: u64) -> Option<SegmentSeq> {
2796 u32::try_from(public.checked_sub(self.media_sequence_offset)?)
2797 .ok()
2798 .map(SegmentSeq::new)
2799 }
2800
2801 /// How many parts past the live edge a request may name (RFC 8216bis
2802 /// §6.2.5.2's Advance Part Limit): three over the part target in seconds
2803 /// when that is under one second, else three. Zero when this origin is
2804 /// not low-latency.
2805 fn advance_part_limit(&self) -> u64 {
2806 match self.effective_part_target_ms() {
2807 None | Some(0) => 0,
2808 Some(ms) if u64::from(ms) < MILLIS_PER_SEC => {
2809 (ADVANCE_PART_LIMIT_NUMERATOR * MILLIS_PER_SEC).div_ceil(u64::from(ms))
2810 }
2811 Some(_) => ADVANCE_PART_LIMIT_NUMERATOR,
2812 }
2813 }
2814
2815 /// Number of segments whose (rounded) duration exceeded the frozen
2816 /// `EXT-X-TARGETDURATION` after first publication — a conformance
2817 /// degradation count (RFC 8216 §4.3.3.1).
2818 pub fn target_duration_violations(&self) -> u64 {
2819 self.target_duration_violations.load(Ordering::Relaxed)
2820 }
2821
2822 /// The `EXT-X-TARGETDURATION` this origin's Media Playlists carry (or will
2823 /// carry): the frozen value once published, otherwise the largest of its
2824 /// own, its longest segment so far and its sibling renditions'. Unlike a
2825 /// render it never freezes the value. RFC 8216bis section 6.2.5.2 measures
2826 /// a blocking reload's patience in these.
2827 pub fn published_target_duration(&self) -> Duration {
2828 let frozen = *locked(&self.frozen_target_duration_secs);
2829 let secs = frozen.unwrap_or_else(|| {
2830 let sibling = locked(&self.rendition_siblings)
2831 .iter()
2832 .filter_map(|s| s.origin.upgrade())
2833 .map(|o| o.target_duration_candidate_secs())
2834 .fold(0.0, f64::max);
2835 self.target_duration_candidate_secs().max(sibling)
2836 });
2837 Duration::from_secs(secs.ceil().clamp(0.0, f64::from(u32::MAX)) as u64)
2838 }
2839
2840 /// Raise this origin's `PART-TARGET` to `ms` when that exceeds the
2841 /// configured part target. An audio rendition whose frames do not divide
2842 /// the part target needs a Part Target Duration of a whole number of
2843 /// frames, because RFC 8216bis §4.4.4.9 requires every Partial Segment to
2844 /// be no longer than it. Has no effect on a classic-HLS origin.
2845 pub fn set_part_target_ms(&self, ms: u32) {
2846 self.part_target_override_ms.store(ms, Ordering::Relaxed);
2847 }
2848
2849 fn effective_part_target_ms(&self) -> Option<u32> {
2850 self.part_target_ms
2851 .map(|base| base.max(self.part_target_override_ms.load(Ordering::Relaxed)))
2852 }
2853
2854 /// The largest Part Target Duration across this origin and its sibling
2855 /// renditions: the `PART-HOLD-BACK` every playlist carries is three
2856 /// times it (RFC 8216bis §4.4.3.8, §6.2.4).
2857 fn max_part_target_ms(&self) -> Option<u32> {
2858 let own = self.effective_part_target_ms()?;
2859 let siblings = locked(&self.rendition_siblings).clone();
2860 Some(
2861 siblings
2862 .iter()
2863 .filter_map(|s| s.origin.upgrade())
2864 .filter_map(|o| o.effective_part_target_ms())
2865 .fold(own, u32::max),
2866 )
2867 }
2868
2869 fn part_timing(&self) -> Option<(DecimalSeconds, DecimalSeconds)> {
2870 let part_target = self.effective_part_target_ms()?;
2871 let hold_basis = self.max_part_target_ms().unwrap_or(part_target);
2872 let part_target = DecimalSeconds::new(f64::from(part_target) / 1000.0)
2873 .expect("u32 / 1000.0 is finite, >= 0");
2874 let part_hold_back =
2875 DecimalSeconds::new(f64::from(hold_basis) / 1000.0 * PART_HOLD_BACK_MULTIPLIER)
2876 .expect("finite, >= 0 times a positive constant is finite, >= 0");
2877 Some((part_target, part_hold_back))
2878 }
2879
2880 /// Replace the sibling renditions whose live edge this origin reports.
2881 pub fn set_rendition_siblings(&self, siblings: Vec<RenditionSibling>) {
2882 *locked(&self.rendition_siblings) = siblings;
2883 }
2884
2885 /// Snapshot of this origin's current live edge as `(LAST-MSN, LAST-PART)`,
2886 /// suitable for an `#EXT-X-RENDITION-REPORT` (RFC 8216bis §4.4.5.4).
2887 /// Drains this origin first and queries only this origin's own `Trunk`,
2888 /// so callers never hold another origin's lock.
2889 pub fn current_live_edge(&self) -> (u64, Option<u64>) {
2890 self.drain();
2891 let (seq, open_parts) = self.live_edge();
2892 if !open_parts.is_empty() {
2893 let msn = self.public_msn(seq);
2894 let last_part = open_parts.last().map(|p| u64::from(p.part_index.value()));
2895 return (msn, last_part);
2896 }
2897 // Live edge is a closed segment: report its last part (if any) so
2898 // the sibling's partial segments are accounted for (RFC 8216bis
2899 // §4.4.5.4; Apple's validator requires LAST-PART whenever the
2900 // rendition carries partial segments).
2901 let last_closed = self
2902 .trunk
2903 .last_closed_segment()
2904 .unwrap_or(SegmentSeq::new(0));
2905 let msn = self.public_msn(last_closed);
2906 let closed_parts = self.trunk.parts_in_segment(last_closed);
2907 let last_part = closed_parts.last().map(|p| u64::from(p.part_index.value()));
2908 (msn, last_part)
2909 }
2910
2911 /// The fMP4 init segment bytes, if set.
2912 pub fn init_bytes(&self) -> Option<Bytes> {
2913 locked(&self.init).current.clone()
2914 }
2915
2916 /// Drain this origin's [`SegmentCursor`] into `Window` — called at the
2917 /// top of every [`ServedEgress::resolve`] so a render always reflects
2918 /// whatever has published since the last call. Non-blocking, bounded by
2919 /// however many segments actually published since the last drain.
2920 ///
2921 /// A [`SegmentCursorItem::Lagged`] report (this origin's `window_segments`/
2922 /// polling cadence fell behind the `Trunk`'s own
2923 /// `segment_capacity` eviction) is accepted, not treated as an error:
2924 /// exactly like every other lossy cursor in this workspace, the honest
2925 /// response is to resume from the next segment, not to fabricate the
2926 /// lost entries' duration/discontinuity data.
2927 fn drain(&self) {
2928 // Nothing published since the last drain: skip both locks. The head
2929 // is read *before* draining and stored after, so a segment published
2930 // in between just makes the next call drain once more (audit r09-O4).
2931 let head = self
2932 .trunk
2933 .last_closed_segment()
2934 .map_or(0, |n| u64::from(n.value()).saturating_add(1));
2935 if self.drained_through.load(Ordering::Acquire) == head {
2936 return;
2937 }
2938 let _gate = locked(&self.drain_gate);
2939 if self.drained_through.load(Ordering::Acquire) == head {
2940 return;
2941 }
2942 // Poll the cursor first, so the heavy I-frame byte scan below runs
2943 // without holding the window lock (and without holding the cursor
2944 // lock while the init store is consulted).
2945 let mut cursor = locked(&self.cursor);
2946 let mut polled = Vec::new();
2947 while let Some(item) = cursor.poll() {
2948 polled.push(item);
2949 }
2950 drop(cursor);
2951
2952 enum Prepared {
2953 Segment(SegmentEntry, Generation, Option<u64>, Option<u32>),
2954 Gap,
2955 }
2956 let video_track_id = *locked(&self.video_track_id);
2957 let signalled_rate = locked(&self.frame_rate).is_some();
2958 let video_timescale = locked(&self.track_specs)
2959 .iter()
2960 .find(|spec| is_video_codec(&spec.config))
2961 .map(|spec| spec.timescale);
2962 let mut prepared = Vec::with_capacity(polled.len());
2963 for item in polled {
2964 match item {
2965 SegmentCursorItem::Segment(entry) => {
2966 let iframe_len = video_track_id
2967 .and_then(|track_id| compute_iframe_len(&entry.bytes, track_id));
2968 let video_rate = if signalled_rate {
2969 None
2970 } else {
2971 match self.container {
2972 Container::MpegTs => ts_video_rate(&entry.bytes),
2973 _ => video_track_id.zip(video_timescale).and_then(
2974 |(track_id, timescale)| {
2975 fmp4_video_rate(&entry.bytes, track_id, timescale)
2976 },
2977 ),
2978 }
2979 };
2980 let init_gen = locked(&self.init).generation_for(entry.sequence_number);
2981 prepared.push(Prepared::Segment(entry, init_gen, iframe_len, video_rate));
2982 }
2983 // `Lagged`/`Gap`/anything newer: segments were lost, so the
2984 // next one does not follow the window's contents.
2985 _ => prepared.push(Prepared::Gap),
2986 }
2987 }
2988
2989 let mut window = locked(&self.window);
2990 for item in prepared {
2991 match item {
2992 Prepared::Segment(entry, init_gen, iframe_len, video_rate) => {
2993 // Record the arrival origin exactly once, the first time a
2994 // segment drains into the window while an arrival clock is
2995 // configured. "this segment ended about now, so it started
2996 // duration ago" (issue #1423 B2a).
2997 if self.arrival_clock.is_some() {
2998 let mut origin = locked(&self.arrival_origin);
2999 if origin.is_none()
3000 && let Some(clock) = &self.arrival_clock
3001 {
3002 let now_ms = clock();
3003 let dur_nanos = entry.duration.as_nanos();
3004 let dur_ms = (dur_nanos + 500_000) / 1_000_000;
3005 let dur_ms = i64::try_from(dur_ms).unwrap_or(i64::MAX);
3006 let epoch_ms = now_ms.saturating_sub(dur_ms);
3007 *origin = Some((entry.timeline_position.as_nanos(), epoch_ms));
3008 }
3009 }
3010 let seg_duration = entry.duration;
3011 window.push(entry, init_gen, iframe_len, video_rate);
3012 if let Some(frozen) = *locked(&self.frozen_target_duration_secs)
3013 && seg_duration.as_secs_f64().round() > frozen.round()
3014 {
3015 self.target_duration_violations
3016 .fetch_add(1, Ordering::Relaxed);
3017 }
3018 }
3019 Prepared::Gap => window.note_gap(),
3020 }
3021 }
3022 self.drained_through.store(head, Ordering::Release);
3023 }
3024
3025 /// `FRAME-RATE` of the video variants: the SPS VUI timing rate when the
3026 /// stream signals one, else the highest per-segment mean rate measured over
3027 /// the segments currently in the playlist window (fMP4 `trun` durations, or MPEG-TS DTS deltas). For
3028 /// a constant-rate stream that mean is the rate; for a variable-rate one it
3029 /// is the maximum, as the attribute requires, and a single long gap lowers
3030 /// only its own segment. Apple HLS
3031 /// authoring spec item 9.15 requires it on every `EXT-X-STREAM-INF` with
3032 /// video, and RFC 8216bis section 4.4.6.2 defines it as the maximum frame
3033 /// rate of the variant's video.
3034 fn effective_frame_rate(&self) -> Option<FrameRate> {
3035 if let Some(signalled) = *locked(&self.frame_rate) {
3036 return Some(signalled);
3037 }
3038 let millihertz = locked(&self.window).max_video_millihertz()?;
3039 Some(FrameRate::from_millihertz(millihertz))
3040 }
3041
3042 /// What `render_playlist` needs from the window, copied out so the window
3043 /// lock is not held while the `Trunk` is queried. Segment bytes are
3044 /// `Bytes` (reference-counted), so the copy shares them.
3045 fn window_view(&self) -> WindowView {
3046 let w = locked(&self.window);
3047 WindowView {
3048 segments: w.segments.clone(),
3049 max_segment_duration_secs: w.max_segment_duration_secs,
3050 discontinuity_sequence: w.discontinuity_sequence,
3051 }
3052 }
3053
3054 /// Private PDT helper shared by the media, open-segment and I-frame
3055 /// renderers so the three can never disagree (issue #1423 B2a, Apple
3056 /// 9.x/8.15). `seq` is the segment's sequence number (needed for the
3057 /// explicit-anchor path); `start_ns` is the segment's position on the
3058 /// trunk's continuous timeline (nanoseconds).
3059 fn segment_pdt(
3060 &self,
3061 seq: SegmentSeq,
3062 start_ns: u64,
3063 ) -> Option<broadcast_hls::ProgramDateTime> {
3064 // (1) Explicit trunk time anchor wins.
3065 if let Some(anchor) = self.trunk.time_anchor()
3066 && let Some(media_start) = self.trunk.segment_start_time(seq)
3067 && let Ok(ts) = jiff::Timestamp::from_millisecond(anchor.media_to_epoch_ms(media_start))
3068 {
3069 return Some(broadcast_hls::ProgramDateTime::from_timestamp(ts));
3070 }
3071 // (2) Arrival origin.
3072 let origin = *locked(&self.arrival_origin);
3073 if let Some((origin_start_ns, origin_epoch_ms)) = origin {
3074 let delta_ns = i128::from(start_ns) - i128::from(origin_start_ns);
3075 let delta_ms = delta_ns / 1_000_000;
3076 let epoch_ms = i128::from(origin_epoch_ms) + delta_ms;
3077 let epoch_ms: i64 = epoch_ms.try_into().ok()?;
3078 if let Ok(ts) = jiff::Timestamp::from_millisecond(epoch_ms) {
3079 return Some(broadcast_hls::ProgramDateTime::from_timestamp(ts));
3080 }
3081 }
3082 // (3) None.
3083 None
3084 }
3085
3086 /// A snapshot of this origin's currently-advertised closed segments
3087 /// (drains the cursor first, same as `render_playlist`) — ascending by
3088 /// sequence number.
3089 ///
3090 /// Exists for a caller that needs to merge this origin's live window
3091 /// with a *different* source of segments over the same numbering
3092 /// (multimux's DVR archive, issue #900: catch-up serving must present
3093 /// one continuous playlist spanning the archive and the still-live
3094 /// tail that hasn't been archived yet). Reuses the one cursor `drain`
3095 /// already maintains rather than making the caller open a second
3096 /// cursor on the same `Trunk` just to learn the same window
3097 /// `render_playlist` itself renders — `media_plane`'s own module doc:
3098 /// writer cost is O(N) in cursor count, so a cursor is per distinct
3099 /// consumer, never per peer, and never duplicated for data another
3100 /// cursor already tracks.
3101 pub fn closed_segments(&self) -> Vec<ClosedSegment> {
3102 self.drain();
3103 locked(&self.window)
3104 .segments
3105 .iter()
3106 .map(|s| ClosedSegment {
3107 sequence_number: s.sequence_number,
3108 start_ns: s.start_ns,
3109 duration_secs: s.duration_secs,
3110 discontinuous: s.discontinuous,
3111 init_generation: s.init_gen,
3112 })
3113 .collect()
3114 }
3115
3116 /// `(in-progress-or-last-active segment sequence number, its currently
3117 /// resident live parts)` — the `Trunk`-only replacement for the deleted
3118 /// `MediaStore::latest_progress`.
3119 ///
3120 /// Derivation: the only segment that can possibly have live, not-yet-
3121 /// closed parts is the one immediately after
3122 /// [`Trunk::last_closed_segment`] (a segmenter never opens segment N+2's
3123 /// parts before N+1 closes) — so probing exactly that one candidate via
3124 /// [`Trunk::parts_in_segment`] is exact, not a heuristic. If that probe
3125 /// is empty (nothing has started for the next segment yet — e.g. the
3126 /// instant after a close, before its successor's first part lands), the
3127 /// answer falls back to `last_closed_segment` itself, with an empty part
3128 /// list — exactly the degenerate state `MediaStore::latest_progress`
3129 /// also returned right after `add_segment` cleared `live_parts`.
3130 fn live_edge(&self) -> (SegmentSeq, Vec<PartEntry>) {
3131 let last_closed = self
3132 .trunk
3133 .last_closed_segment()
3134 .unwrap_or(SegmentSeq::new(0));
3135 let candidate = last_closed.saturating_next();
3136 let parts = self.trunk.parts_in_segment(candidate);
3137 if parts.is_empty() {
3138 (last_closed, Vec::new())
3139 } else {
3140 (candidate, parts)
3141 }
3142 }
3143
3144 /// The `EXT-X-PART` entries of a closed segment, taken from the parts the
3145 /// `Trunk` still holds. Empty unless *all* of them are resident, as a
3146 /// run `0..n`: a part ring that already evicted some of this segment's
3147 /// parts cannot back the playlist's promise that they stay downloadable
3148 /// (RFC 8216bis §6.2.2), and a run with a hole would number the parts
3149 /// wrongly.
3150 fn resident_part_specs(&self, track_id: TrackId, seq: SegmentSeq, ext: &str) -> Vec<PartSpec> {
3151 // Keyed by index: a stale entry reusing a number is overwritten by
3152 // the newer one (publish order).
3153 let by_index: BTreeMap<PartIndex, PartEntry> = self
3154 .trunk
3155 .parts_in_segment(seq)
3156 .into_iter()
3157 .map(|p| (p.part_index, p))
3158 .collect();
3159 let contiguous = by_index
3160 .keys()
3161 .copied()
3162 .zip(0u32..)
3163 .all(|(index, expected)| index.value() == expected);
3164 if by_index.is_empty() || !contiguous {
3165 return Vec::new();
3166 }
3167 by_index
3168 .values()
3169 .map(|p| PartSpec {
3170 uri: self.part_uri(track_id, p.segment_number, p.part_index, ext),
3171 // `Duration::as_secs_f64`: finite and >= 0 (issue #1140).
3172 duration: DecimalSeconds::new(p.duration.as_secs_f64())
3173 .expect("Duration::as_secs_f64 is finite, >= 0"),
3174 independent: p.independent,
3175 ..Default::default()
3176 })
3177 .collect()
3178 }
3179
3180 /// This origin's own `EXT-X-TARGETDURATION` before sibling renditions are
3181 /// considered: the frozen value once published. Every Media Playlist of a
3182 /// Multivariant Playlist carries the same `EXT-X-SERVER-CONTROL`
3183 /// attributes (RFC 8216bis §6.2.4), and `HOLD-BACK` and `CAN-SKIP-UNTIL`
3184 /// are multiples of the Target Duration, so renditions share the largest.
3185 fn target_duration_candidate_secs(&self) -> f64 {
3186 let own = self
3187 .target_duration_secs
3188 .max(locked(&self.window).max_segment_duration_secs);
3189 (*locked(&self.frozen_target_duration_secs)).unwrap_or(own)
3190 }
3191
3192 /// Resolve the `EXT-X-TARGETDURATION` to publish for `window`, freezing
3193 /// it at the first render that carries at least one segment (RFC 8216
3194 /// §4.3.3.1): once published it never changes, even as later (possibly
3195 /// longer) segments ratchet `max_segment_duration_secs`.
3196 fn published_target_duration_secs(&self, window: &WindowView) -> f64 {
3197 let siblings = locked(&self.rendition_siblings).clone();
3198 let sibling_target = siblings
3199 .iter()
3200 .filter_map(|s| s.origin.upgrade())
3201 .map(|o| o.target_duration_candidate_secs())
3202 .fold(0.0, f64::max);
3203 let candidate = self
3204 .target_duration_secs
3205 .max(window.max_segment_duration_secs)
3206 .max(sibling_target);
3207 let mut frozen = locked(&self.frozen_target_duration_secs);
3208 if let Some(existing) = *frozen {
3209 existing
3210 } else if window.segments.is_empty() {
3211 candidate
3212 } else {
3213 *frozen = Some(candidate);
3214 candidate
3215 }
3216 }
3217
3218 /// Render the LL-HLS media playlist for `track_id` from this origin's
3219 /// current `Window` (closed segments) and the `Trunk`'s live edge (open
3220 /// segment's parts + preload hint).
3221 ///
3222 /// RFC 8216bis §4.4.4.9: an in-progress (not yet closed) segment MUST NOT
3223 /// be advertised with an `#EXTINF`/URI pair — that segment has no
3224 /// fetchable resource yet — it may only appear as trailing `#EXT-X-PART`
3225 /// lines. `broadcast_hls::MediaPlaylist::open_segment` is exactly this
3226 /// representation: its parts render as trailing `#EXT-X-PART` lines with
3227 /// no `#EXTINF`/URI, so the in-progress segment's parts and the
3228 /// `#EXT-X-PRELOAD-HINT` for the next, not-yet-available part are both
3229 /// rendered by `to_m3u8()` itself — this method only supplies the URI
3230 /// scheme (`part-<track>-<seq>.<idx>.m4s`) and the part metadata.
3231 /// Returns the rendered playlist, or a human-readable reason the
3232 /// playlist could not be rendered (a field carrying a character
3233 /// forbidden in an HLS quoted-string/URI — audit BH-W7, issue
3234 /// #1111; unreachable for this origin, which only ever supplies
3235 /// its own generated URIs).
3236 fn render_playlist(
3237 &self,
3238 track_id: TrackId,
3239 skip: Option<SkipDirective>,
3240 ) -> core::result::Result<String, String> {
3241 self.drain();
3242 // Snapshot the window and release its lock before any `Trunk` call:
3243 // `live_edge`, the part lookups and the event lookups below take the
3244 // `Trunk`'s own lock, which must not nest inside this origin's
3245 // (audit r09-O4).
3246 let window = self.window_view();
3247 let (open_seq, open_parts) = self.live_edge();
3248 // Only render an open segment/preload-hint once the live edge is
3249 // genuinely a not-yet-closed segment with at least one live part —
3250 // never re-render an already-closed segment's lingering parts (the
3251 // `Trunk`'s live-part log deliberately does not evict them on close;
3252 // see `trunk`'s own module doc) as if they were still open.
3253 // Classic HLS (no `.low_latency(...)` call, issue #873) never
3254 // advertises an in-progress segment at all — RFC 8216bis §4.4.4.9's
3255 // trailing-`#EXT-X-PART`-only representation is itself an LL-HLS
3256 // directive, so it is gated on low latency being enabled, not merely
3257 // on the Trunk happening to have live parts.
3258 let low_latency_enabled = self.part_target_ms.is_some();
3259 let has_open_parts = low_latency_enabled && !open_parts.is_empty();
3260 let ext = self.container.segment_extension();
3261
3262 let media_sequence = window
3263 .segments
3264 .front()
3265 .map(|s| self.public_msn(s.sequence_number))
3266 .or_else(|| has_open_parts.then(|| self.public_msn(open_seq)))
3267 .unwrap_or_else(|| self.public_msn(SegmentSeq::new(1)));
3268 // RFC 8216bis §4.4.3.1 (MUST): every Media Segment's EXTINF duration,
3269 // rounded to the nearest integer, MUST be <= TARGETDURATION. The
3270 // segmenter cuts on the next keyframe *after* the configured target,
3271 // so a real segment routinely exceeds it — advertising the
3272 // configured target alone can under-declare. Use whichever is
3273 // larger. RFC 8216 §4.3.3.1: `EXT-X-TARGETDURATION` is frozen once
3274 // published, so it is rounded up to leave headroom for a later,
3275 // slightly longer segment.
3276 let target_duration = self.published_target_duration_secs(&window);
3277 // `target_duration_secs` is validated finite and positive at build
3278 // and segment durations come from `Duration`, so this is finite;
3279 // the clamp makes the float->int conversion exact-or-saturating.
3280 let target_duration = target_duration.ceil().clamp(0.0, f64::from(u32::MAX)) as u32;
3281 // RFC 8216bis §6.2.5.1: the Skip Boundary is at least six times the
3282 // Target Duration (advertised as CAN-SKIP-UNTIL). A delta update skips
3283 // the Media Segments further from the end of the last segment than
3284 // this boundary.
3285 let skip_boundary = if low_latency_enabled {
3286 Some(f64::from(target_duration) * SKIP_BOUNDARY_MULTIPLIER)
3287 } else {
3288 None
3289 };
3290 let playlist_has_endlist = false;
3291 let skip_accepted = skip.is_some() && skip_boundary.is_some() && !playlist_has_endlist;
3292 // RFC 8216bis §6.2.5.1: count the leading segments whose distance
3293 // (from their start to the end of the last segment) exceeds the Skip
3294 // Boundary; never skip every segment in the window.
3295 let omitted = if skip_accepted {
3296 let durations: Vec<f64> = window.segments.iter().map(|s| s.duration_secs).collect();
3297 let total: f64 = durations.iter().sum();
3298 let mut prefix = 0.0_f64;
3299 let mut count = 0usize;
3300 for d in &durations {
3301 let distance = total - prefix;
3302 if distance > skip_boundary.unwrap_or(f64::INFINITY) {
3303 count += 1;
3304 prefix += d;
3305 } else {
3306 break;
3307 }
3308 }
3309 count.min(durations.len().saturating_sub(1))
3310 } else {
3311 0
3312 };
3313 // RFC 8216bis §6.2.2: EXT-X-PART tags stay in the Playlist until they
3314 // are more than three Target Durations from its end — so a closed
3315 // segment keeps its parts while the media after it (later closed
3316 // segments plus the open segment's parts) is within that distance.
3317 let retention_secs = PART_RETENTION_TARGET_DURATIONS * f64::from(target_duration);
3318 let mut behind_secs: f64 = if has_open_parts {
3319 open_parts.iter().map(|p| p.duration.as_secs_f64()).sum()
3320 } else {
3321 0.0
3322 };
3323 let mut closed_parts: Vec<Vec<PartSpec>> = Vec::with_capacity(window.segments.len());
3324 for s in window.segments.iter().rev() {
3325 let keep = low_latency_enabled && behind_secs <= retention_secs;
3326 closed_parts.push(if keep {
3327 self.resident_part_specs(track_id, s.sequence_number, ext)
3328 } else {
3329 Vec::new()
3330 });
3331 behind_secs += s.duration_secs;
3332 }
3333 closed_parts.reverse();
3334 let any_closed_parts = closed_parts.iter().any(|parts| !parts.is_empty());
3335 let fmp4 = self.container == Container::Fmp4;
3336 let map_for = |generation: Generation| {
3337 fmp4.then(|| MapTag {
3338 uri: self.init_uri(track_id, generation),
3339 byte_range: None,
3340 extra_attrs: Vec::new(),
3341 })
3342 };
3343 // PDT per segment: explicit trunk time anchor (B1) wins, else the
3344 // arrival wall-clock source (B2a), else none. One private helper
3345 // keeps the media, open-segment and I-frame renderers in agreement.
3346 let segments: Vec<MediaSegment> = window
3347 .segments
3348 .iter()
3349 .zip(closed_parts)
3350 .map(|(s, parts)| {
3351 let program_date_time = self.segment_pdt(s.sequence_number, s.start_ns);
3352 MediaSegment {
3353 map: map_for(s.init_gen),
3354 uri: self.segment_uri(track_id, s.sequence_number, ext),
3355 // `duration_secs` is always `Duration::as_secs_f64()`
3356 // (issue #1140): a `std::time::Duration` can never be
3357 // negative or non-finite.
3358 duration: DecimalSeconds::new(s.duration_secs)
3359 .expect("Duration::as_secs_f64 is finite, >= 0"),
3360 discontinuous: s.discontinuous,
3361 parts,
3362 program_date_time,
3363 // A segment published with no media bytes is a gap placeholder
3364 // (issue #1428 slice 2b): it occupies its MSN in the
3365 // playlist without carrying media, so the client skips it.
3366 gap: s.bytes.is_empty(),
3367 ..Default::default()
3368 }
3369 })
3370 .collect::<Vec<_>>();
3371 // Playlist Delta Update (RFC 8216bis §6.2.5.1): drop the segments
3372 // older than the Skip Boundary; MEDIA-SEQUENCE stays at the original
3373 // first segment's number and EXT-X-SKIP records how many were omitted.
3374 let segments: Vec<MediaSegment> = if omitted > 0 {
3375 segments.into_iter().skip(omitted).collect()
3376 } else {
3377 segments
3378 };
3379 // The open segment is cut against the init generation that covers its
3380 // sequence number (`#EXT-X-MAP` carries forward, so this repeats the
3381 // last closed segment's unless the init changed in between).
3382 let open_gen = locked(&self.init).generation_for(open_seq);
3383 let open_segment = has_open_parts.then(|| {
3384 let mut open = OpenSegment::new(
3385 open_parts
3386 .iter()
3387 .map(|p| PartSpec {
3388 uri: self.part_uri(track_id, p.segment_number, p.part_index, ext),
3389 // `p.duration` is a `std::time::Duration`: always
3390 // finite and non-negative (issue #1140).
3391 duration: DecimalSeconds::new(p.duration.as_secs_f64())
3392 .expect("Duration::as_secs_f64 is finite, >= 0"),
3393 independent: p.independent,
3394 ..Default::default()
3395 })
3396 .collect(),
3397 );
3398 if let Some(map) = map_for(open_gen) {
3399 open = open.with_map(map);
3400 }
3401 // Open (in-progress) segment PDT. The arrival path uses the last
3402 // closed segment's start + its duration (B2a); with no closed
3403 // segment yet it is omitted. The explicit-anchor path is handled
3404 // inside `segment_pdt` via `segment_start_time(open_seq)`.
3405 let open_start_ns = window.segments.back().map(|s| {
3406 s.start_ns.saturating_add(
3407 u64::try_from((s.duration_secs * 1_000_000_000.0).round() as i64).unwrap_or(0),
3408 )
3409 });
3410 open.program_date_time = match open_start_ns {
3411 Some(sn) => self.segment_pdt(open_seq, sn),
3412 None => {
3413 // No closed segment: explicit anchor only.
3414 if self.trunk.time_anchor().is_some() {
3415 self.segment_pdt(open_seq, 0)
3416 } else {
3417 None
3418 }
3419 }
3420 };
3421 open
3422 });
3423 // RFC 8216bis Appendix B.1: "A Playlist that contains an EXT-X-PART
3424 // tag but no EXT-X-ENDLIST tag will also contain an
3425 // EXT-X-PRELOAD-HINT tag that specifies the next Partial Segment that
3426 // is expected to be added to the Playlist." Right after a segment
3427 // closes and before its successor has a part, the playlist still
3428 // carries the closed segments' EXT-X-PART tags, so the hint names
3429 // part 0 of the next Media Sequence Number (section 4.4.5.3).
3430 let next_part_hint = if has_open_parts {
3431 let next_idx = open_parts
3432 .iter()
3433 .map(|p| p.part_index)
3434 .max()
3435 .map(PartIndex::saturating_next)
3436 .unwrap_or(PartIndex::new(0));
3437 Some(self.part_uri(track_id, open_seq, next_idx, ext))
3438 } else if low_latency_enabled && any_closed_parts {
3439 Some(self.part_uri(track_id, open_seq.saturating_next(), PartIndex::new(0), ext))
3440 } else {
3441 None
3442 };
3443 // `#EXT-X-MAP`: unconditional under Fmp4 (RFC 8216bis §3.1.2 MUST);
3444 // omitted under MpegTs by default (§3.1.1's PAT/PMT-or-MAP
3445 // disjunction — see `Container`'s own doc for why this is a default,
3446 // not a hard restriction). Then any SCTE-35 cues published to the
3447 // trunk's event ring for this window render as `#EXT-X-DATERANGE`
3448 // tag lines (issue #965): they carry a wall-clock `START-DATE` only
3449 // once the trunk's `time_anchor` has been set, so events are
3450 // silently skipped while the anchor is absent (their `to_daterange`
3451 // fails the same way it does for a non-SCTE-35 source).
3452 // The map rides on the segments (each one names the init generation
3453 // it was cut against, so a changed init is announced together with its
3454 // `EXT-X-DISCONTINUITY`); only a playlist with no segment at all
3455 // carries it as a bare tag.
3456 // Date-range tags, collected per segment so a Playlist Delta
3457 // Update can drop the ones belonging to skipped segments (and, for
3458 // `_HLS_skip=v2`, report their ids in RECENTLY-REMOVED-DATERANGES).
3459 let mut segment_dateranges: Vec<Vec<(String, String)>> =
3460 vec![Vec::new(); window.segments.len()];
3461 let mut recently_removed_daterange_ids: Vec<String> = Vec::new();
3462 if let Some(anchor) = self.trunk.time_anchor() {
3463 let timeline = timed_metadata::Timeline::with_anchor(anchor);
3464 for (idx, seg) in window.segments.iter().enumerate() {
3465 for entry in self.trunk.events_in_segment(seg.sequence_number) {
3466 if let Ok(dr) = timeline.to_daterange(&entry.event)
3467 && let Ok(tag) = dr.to_tag_line()
3468 {
3469 if idx < omitted && skip_accepted && skip == Some(SkipDirective::V2) {
3470 recently_removed_daterange_ids.push(dr.id.clone());
3471 }
3472 segment_dateranges[idx].push((tag, dr.id));
3473 }
3474 }
3475 }
3476 }
3477 let mut extra_tags = Vec::new();
3478 if fmp4 && segments.is_empty() && open_segment.is_none() {
3479 let generation = locked(&self.init).generation.max(Generation::new(1));
3480 extra_tags.push(format!(
3481 "#EXT-X-MAP:URI=\"{}\"",
3482 self.init_uri(track_id, generation)
3483 ));
3484 }
3485 for (idx, tags) in segment_dateranges.iter().enumerate() {
3486 // RFC 8216bis §6.2.5.1: only `_HLS_skip=v2` additionally omits
3487 // `EXT-X-DATERANGE` tags; with `YES` every daterange stays.
3488 if idx < omitted && skip == Some(SkipDirective::V2) {
3489 continue;
3490 }
3491 for (tag, _) in tags {
3492 extra_tags.push(tag.clone());
3493 }
3494 }
3495 let low_latency = self.part_timing().map(|(part_target, part_hold_back)| {
3496 LowLatencyConfig {
3497 part_target: Some(part_target),
3498 part_hold_back: Some(part_hold_back),
3499 // RFC 8216bis §4.4.3.8: HOLD-BACK MUST be >= 3x the Target
3500 // Duration; Apple's validator also requires it in LL mode.
3501 hold_back: DecimalSeconds::new(f64::from(target_duration) * HOLD_BACK_MULTIPLIER)
3502 .ok(),
3503 preload_hint_part: next_part_hint,
3504 can_skip_until: skip_boundary.and_then(|b| DecimalSeconds::new(b).ok()),
3505 can_skip_dateranges: true,
3506 can_block_reload: true,
3507 ..Default::default()
3508 }
3509 });
3510 let skip_info = if skip_accepted {
3511 Some(broadcast_hls::SkipInfo {
3512 skipped_segments: omitted as u64,
3513 recently_removed_daterange_ids,
3514 extra_attrs: Vec::new(),
3515 })
3516 } else {
3517 None
3518 };
3519 // #EXT-X-RENDITION-REPORT for sibling renditions (RFC 8216bis
3520 // §4.4.5.4): one per OTHER rendition, snapshotted under each
3521 // sibling's own lock at render time.
3522 let rendition_reports = if self.part_target_ms.is_some() {
3523 let siblings = locked(&self.rendition_siblings).clone();
3524 let mut reports = Vec::with_capacity(siblings.len());
3525 for sibling in siblings {
3526 let Some(origin) = sibling.origin.upgrade() else {
3527 continue;
3528 };
3529 if sibling.playlist_uri == self.playlist_uri {
3530 continue;
3531 }
3532 let uri = relative_playlist_uri(&self.playlist_uri, &sibling.playlist_uri);
3533 let (last_msn, last_part) = origin.current_live_edge();
3534 reports.push(broadcast_hls::RenditionReport {
3535 uri,
3536 last_msn,
3537 last_part,
3538 extra_attrs: Vec::new(),
3539 });
3540 }
3541 reports
3542 } else {
3543 Vec::new()
3544 };
3545
3546 let playlist = MediaPlaylist {
3547 // No explicit floor: `broadcast_hls::MediaPlaylist::to_m3u8`
3548 // computes `EXT-X-VERSION` from the content actually emitted
3549 // (RFC 8216bis §8) rather than this origin choosing a value
3550 // ahead of time (issue #871) — none of the LL-HLS directives
3551 // below (`EXT-X-PART`/`EXT-X-PART-INF`/`EXT-X-PRELOAD-HINT`/
3552 // `EXT-X-SERVER-CONTROL`) carry any version requirement at all.
3553 target_duration,
3554 media_sequence,
3555 discontinuity_sequence: window.discontinuity_sequence,
3556 always_emit_discontinuity_sequence: true,
3557 segments,
3558 open_segment,
3559 endlist: false,
3560 extra_tags,
3561 low_latency,
3562 iframes_only: false,
3563 skip: skip_info,
3564 rendition_reports,
3565 ..Default::default()
3566 };
3567 playlist.to_m3u8().map_err(|error| error.to_string())
3568 }
3569
3570 fn resolve_playlist(
3571 &self,
3572 track_id: TrackId,
3573 query: BlockingQuery,
3574 now: Timestamp,
3575 await_policy: AwaitPolicy,
3576 ) -> EgressResponse<HlsBody> {
3577 if query.hls_part.is_some() && query.hls_msn.is_none() {
3578 return EgressResponse::BadRequest {
3579 reason: "_HLS_part without _HLS_msn is meaningless",
3580 };
3581 }
3582 if let Some(msn) = query.hls_msn {
3583 let (in_progress_seg, live_parts) = self.live_edge();
3584 let in_progress = self.public_msn(in_progress_seg);
3585 if msn > in_progress.saturating_add(ABUSE_MSN_FUTURE_BOUND) {
3586 return EgressResponse::BadRequest {
3587 reason: "_HLS_msn unreasonably far beyond the live edge",
3588 };
3589 }
3590 // RFC 8216bis §6.2.5.2 (SHOULD): an `_HLS_part` that exceeds the
3591 // last Partial Segment in the Playlist by the Advance Part Limit
3592 // is a Bad Request at once, not a request to park. Measured
3593 // against the live-edge segment's parts (none exist yet for a
3594 // later `_HLS_msn`).
3595 if let Some(part) = query.hls_part
3596 && msn >= in_progress
3597 {
3598 let existing = if msn == in_progress {
3599 u64::try_from(live_parts.len()).unwrap_or(u64::MAX)
3600 } else {
3601 0
3602 };
3603 let limit = existing
3604 .saturating_sub(1)
3605 .saturating_add(self.advance_part_limit());
3606 if u64::from(part) > limit {
3607 return EgressResponse::BadRequest {
3608 reason: "_HLS_part beyond the Advance Part Limit",
3609 };
3610 }
3611 }
3612 let satisfied = match query.hls_part {
3613 Some(part) => {
3614 in_progress > msn
3615 || (in_progress == msn
3616 && u64::try_from(live_parts.len()).unwrap_or(u64::MAX)
3617 > u64::from(part))
3618 }
3619 None => {
3620 self.public_msn(
3621 self.trunk
3622 .last_closed_segment()
3623 .unwrap_or(SegmentSeq::new(0)),
3624 ) >= msn
3625 }
3626 };
3627 if !satisfied {
3628 return EgressResponse::pending(await_policy, now, now);
3629 }
3630 }
3631 match self.render_playlist(track_id, query.hls_skip) {
3632 Ok(playlist) => EgressResponse::Ready {
3633 body: HlsBody::Playlist(playlist),
3634 cache: CachePolicy::NoCache,
3635 },
3636 // A field the renderer would have to quote carries a character
3637 // forbidden there (audit BH-W7, issue #1111) — unreachable for
3638 // this origin's own generated URIs, but never emitted mangled.
3639 // Answered as a plain miss, never a debug panic: a hostile or
3640 // malformed data-driven field must not take the origin down.
3641 Err(_) => EgressResponse::NotFound,
3642 }
3643 }
3644
3645 fn resolve_iframe_playlist(
3646 &self,
3647 track_id: TrackId,
3648 query: BlockingQuery,
3649 media_playlist_name: String,
3650 now: Timestamp,
3651 await_policy: AwaitPolicy,
3652 ) -> EgressResponse<HlsBody> {
3653 // RFC 8216bis §6.2.5.2: `_HLS_part` without `_HLS_msn` MUST be a Bad
3654 // Request. With `_HLS_msn` it is legitimate on a playlist without
3655 // Partial Segments (Apple's validator sends `_HLS_msn=M&_HLS_part=N`
3656 // to the I-frame playlist): a Part Index past the parent's final part
3657 // is treated as Part 0 of the next parent, so it waits on the MSN only.
3658 if query.hls_part.is_some() && query.hls_msn.is_none() {
3659 return EgressResponse::BadRequest {
3660 reason: "_HLS_part without _HLS_msn is meaningless",
3661 };
3662 }
3663 if let Some(msn) = query.hls_msn {
3664 let last_closed = self
3665 .trunk
3666 .last_closed_segment()
3667 .unwrap_or(SegmentSeq::new(0));
3668 if msn
3669 > self
3670 .public_msn(last_closed.saturating_next())
3671 .saturating_add(ABUSE_MSN_FUTURE_BOUND)
3672 {
3673 return EgressResponse::BadRequest {
3674 reason: "_HLS_msn unreasonably far beyond the live edge",
3675 };
3676 }
3677 let satisfied = self.public_msn(last_closed) >= msn;
3678 if !satisfied {
3679 return EgressResponse::pending(await_policy, now, now);
3680 }
3681 }
3682 match self.render_iframe_playlist(track_id, &media_playlist_name) {
3683 Ok(playlist) => EgressResponse::Ready {
3684 body: HlsBody::Playlist(playlist),
3685 cache: CachePolicy::NoCache,
3686 },
3687 Err(_) => EgressResponse::NotFound,
3688 }
3689 }
3690
3691 /// Render the I-frame trick-play playlist for `track_id`.
3692 fn render_iframe_playlist(
3693 &self,
3694 track_id: TrackId,
3695 media_playlist_name: &str,
3696 ) -> core::result::Result<String, String> {
3697 self.drain();
3698 if self.container != Container::Fmp4 {
3699 return Err("not fmp4".to_string());
3700 }
3701 if locked(&self.video_track_id).is_none() {
3702 return Err("no video track".to_string());
3703 }
3704 let window = self.window_view();
3705 // D3': collect entries; a `None` is allowed only before the first entry.
3706 let Some(entries) = self.iframe_entries(&window) else {
3707 return Err("no iframe entry".to_string());
3708 };
3709
3710 // RFC 8216 §4.3.3.1: `EXT-X-TARGETDURATION` is frozen once published.
3711 let target_duration = self.published_target_duration_secs(&window);
3712 let target_duration = target_duration.ceil().clamp(0.0, f64::from(u32::MAX)) as u32;
3713
3714 let fmp4 = self.container == Container::Fmp4;
3715 let map_for = |generation: Generation| {
3716 fmp4.then(|| MapTag {
3717 uri: self.init_uri(track_id, generation),
3718 byte_range: None,
3719 extra_attrs: Vec::new(),
3720 })
3721 };
3722
3723 let first_entry_seq = entries[0].0.sequence_number;
3724 let media_sequence = self.public_msn(first_entry_seq);
3725 let segments: Vec<MediaSegment> = entries
3726 .iter()
3727 .map(|(s, len)| MediaSegment {
3728 map: map_for(s.init_gen),
3729 uri: self.iframe_segment_uri(
3730 track_id,
3731 s.sequence_number,
3732 self.container.segment_extension(),
3733 ),
3734 duration: DecimalSeconds::new(s.duration_secs).unwrap_or_default(),
3735 discontinuous: s.discontinuous,
3736 byte_range: Some(ByteRange {
3737 length: *len,
3738 offset: Some(0),
3739 }),
3740 parts: Vec::new(),
3741 program_date_time: self.segment_pdt(s.sequence_number, s.start_ns),
3742 ..Default::default()
3743 })
3744 .collect();
3745
3746 let low_latency = if let Some((part_target, part_hold_back)) = self.part_timing() {
3747 Some(broadcast_hls::LowLatencyConfig {
3748 part_target: Some(part_target),
3749 part_hold_back: Some(part_hold_back),
3750 hold_back: DecimalSeconds::new(f64::from(target_duration) * HOLD_BACK_MULTIPLIER)
3751 .ok(),
3752 can_skip_until: DecimalSeconds::new(
3753 f64::from(target_duration) * SKIP_BOUNDARY_MULTIPLIER,
3754 )
3755 .ok(),
3756 can_skip_dateranges: true,
3757 can_block_reload: true,
3758 ..Default::default()
3759 })
3760 } else {
3761 None
3762 };
3763
3764 // RFC 8216bis Appendix B.1: "Each Media Playlist MUST contain one
3765 // EXT-X-RENDITION-REPORT tag for each Media Playlist (Rendition) in the
3766 // Multivariant Playlist, except for the Media Playlist to which the
3767 // EXT-X-RENDITION-REPORT tag is being added, and Playlists that
3768 // contain the EXT-X-I-FRAMES-ONLY tag." This I-frame playlist is a
3769 // Media Playlist, so it reports the primary media playlist, with the
3770 // media playlist's own live edge (section 4.4.5.4: LAST-MSN is the
3771 // last Partial Segment's Media Sequence Number and LAST-PART is
3772 // "REQUIRED if the Rendition contains a Partial Segment").
3773 // The reverse, a media playlist reporting this I-frame playlist, is
3774 // only a SHOULD, and Apple's validator rejects it as "Unknown
3775 // rendition report" (issue #1595).
3776 let rendition_reports = if self.part_target_ms.is_some() {
3777 let (last_msn, last_part) = self.current_live_edge();
3778 let mut reports = vec![broadcast_hls::RenditionReport {
3779 uri: media_playlist_name.to_string(),
3780 last_msn,
3781 last_part,
3782 extra_attrs: Vec::new(),
3783 }];
3784 // Split-audio routes: report each audio sibling rendition as well
3785 // (RFC 8216bis §4.4.5.4), keeping the existing iframe->media
3786 // report above unchanged.
3787 let siblings = locked(&self.rendition_siblings).clone();
3788 for sibling in siblings {
3789 let Some(origin) = sibling.origin.upgrade() else {
3790 continue;
3791 };
3792 if sibling.playlist_uri == self.playlist_uri {
3793 continue;
3794 }
3795 let uri = relative_playlist_uri(
3796 &self.iframe_playlist_name(track_id),
3797 &sibling.playlist_uri,
3798 );
3799 let (last_msn, last_part) = origin.current_live_edge();
3800 reports.push(broadcast_hls::RenditionReport {
3801 uri,
3802 last_msn,
3803 last_part,
3804 extra_attrs: Vec::new(),
3805 });
3806 }
3807 reports
3808 } else {
3809 Vec::new()
3810 };
3811
3812 let playlist = MediaPlaylist {
3813 target_duration,
3814 media_sequence,
3815 discontinuity_sequence: window.discontinuity_sequence,
3816 always_emit_discontinuity_sequence: true,
3817 segments,
3818 open_segment: None,
3819 endlist: false,
3820 extra_tags: Vec::new(),
3821 low_latency,
3822 rendition_reports,
3823 iframes_only: true,
3824 ..Default::default()
3825 };
3826 playlist.to_m3u8().map_err(|error| error.to_string())
3827 }
3828
3829 /// A part request is the preload-hinted Partial Segment a client fetches
3830 /// ahead of time (RFC 8216bis §6.2.2, §6.3.1). If the origin promised it
3831 /// via `#EXT-X-PRELOAD-HINT` but hasn't produced it yet,
3832 /// [`EgressResponse::Await`] — the caller should hold the request open
3833 /// (not 404 immediately, which spams errors and defeats low latency).
3834 /// [`EgressResponse::NotFound`] is returned **promptly** (without the
3835 /// caller needing to wait out its own [`AwaitPolicy`]) once the part can
3836 /// no longer appear: its segment has closed (now only addressable as a
3837 /// whole segment via `seg-…`) — a legitimate 404 the client answers by
3838 /// fetching the next segment/part.
3839 fn resolve_resource(
3840 &self,
3841 name: &str,
3842 now: Timestamp,
3843 await_policy: AwaitPolicy,
3844 ) -> EgressResponse<HlsBody> {
3845 if let Some((epoch, public_seq, idx)) = parse_part(name, self.container) {
3846 // Another instance's token never resolves here; the legacy
3847 // (token-less) form resolves but is not `immutable`.
3848 if epoch.is_some_and(|e| e != self.instance) {
3849 return EgressResponse::NotFound;
3850 }
3851 let Some(seq) = self.internal_seq(public_seq) else {
3852 return EgressResponse::NotFound;
3853 };
3854 if let Some(bytes) = self.trunk.part_bytes(seq, idx) {
3855 return EgressResponse::Ready {
3856 body: HlsBody::Resource(bytes),
3857 cache: if epoch.is_some() {
3858 CachePolicy::Immutable
3859 } else {
3860 CachePolicy::NoCache
3861 },
3862 };
3863 }
3864 // A classic (non-LL) origin has no parts at all.
3865 if self.part_target_ms.is_none() {
3866 return EgressResponse::NotFound;
3867 }
3868 // The requested part's segment has already closed (whether or
3869 // not this origin's own `Window` still retains its bytes) -> it
3870 // will never be produced. `Trunk::last_closed_segment` answers
3871 // this exactly, with no dependence on `Window`'s retention.
3872 let never_will = self.trunk.last_closed_segment().is_some_and(|c| c >= seq);
3873 if never_will {
3874 return EgressResponse::NotFound;
3875 }
3876 // Only a part a client could have been hinted at is worth holding
3877 // a request open for (RFC 8216bis §6.2.6): the hinted next part,
3878 // in the open segment or the one after it. Anything else is not
3879 // found at once rather than parked for the whole blocking
3880 // timeout.
3881 let last_closed = self
3882 .trunk
3883 .last_closed_segment()
3884 .unwrap_or(SegmentSeq::new(0));
3885 if last_closed
3886 .checked_add(FUTURE_PART_SEGMENT_BOUND)
3887 .is_some_and(|limit| seq > limit)
3888 {
3889 return EgressResponse::NotFound;
3890 }
3891 // Exactly one part is ever hinted: the one after the last part
3892 // the ring holds for that segment (index 0 when it holds none).
3893 // A lower index that is not resident was evicted (or never
3894 // existed) and will not come back; a higher one was never hinted.
3895 let next_hinted = self
3896 .trunk
3897 .parts_in_segment(seq)
3898 .iter()
3899 .map(|p| p.part_index)
3900 .max()
3901 .map_or(PartIndex::new(0), PartIndex::saturating_next);
3902 if idx != next_hinted {
3903 return EgressResponse::NotFound;
3904 }
3905 return EgressResponse::pending(await_policy, now, now);
3906 }
3907 self.drain();
3908 // `Immutable` only where the name can never map to other bytes: the
3909 // instance-named init/segment forms (see `Self::instance`). The
3910 // token-less forms and the bare `init-{track}.mp4` can.
3911 let (bytes, cache) = match parse_immediate(name, self.container) {
3912 Some(ImmediateResource::CurrentInit) => (self.init_bytes(), CachePolicy::NoCache),
3913 Some(ImmediateResource::VersionedInit { epoch, generation })
3914 if epoch == self.instance =>
3915 {
3916 (
3917 locked(&self.init).bytes_of(generation),
3918 CachePolicy::Immutable,
3919 )
3920 }
3921 Some(ImmediateResource::Segment { epoch, seq })
3922 if epoch.is_none_or(|e| e == self.instance) =>
3923 {
3924 (
3925 self.internal_seq(seq)
3926 .and_then(|s| locked(&self.window).bytes_of(s)),
3927 if epoch.is_some() {
3928 CachePolicy::Immutable
3929 } else {
3930 CachePolicy::NoCache
3931 },
3932 )
3933 }
3934 _ => (None, CachePolicy::NoCache),
3935 };
3936 match bytes {
3937 Some(bytes) => EgressResponse::Ready {
3938 body: HlsBody::Resource(bytes),
3939 cache,
3940 },
3941 None => EgressResponse::NotFound,
3942 }
3943 }
3944}
3945
3946impl ServedEgress for HlsOrigin {
3947 type Request = HlsRequest;
3948 type Body = HlsBody;
3949
3950 fn resolve(
3951 &self,
3952 request: HlsRequest,
3953 now: Timestamp,
3954 await_policy: AwaitPolicy,
3955 ) -> EgressResponse<HlsBody> {
3956 match request {
3957 HlsRequest::Playlist { track_id, query } => {
3958 self.resolve_playlist(track_id, query, now, await_policy)
3959 }
3960 HlsRequest::IFramePlaylist {
3961 track_id,
3962 query,
3963 media_playlist_name,
3964 } => self.resolve_iframe_playlist(
3965 track_id,
3966 query,
3967 media_playlist_name,
3968 now,
3969 await_policy,
3970 ),
3971 HlsRequest::Resource { name } => self.resolve_resource(&name, now, await_policy),
3972 }
3973 }
3974}
3975
3976/// Canonicalise an ISO 639-2 language code into a BCP 47 (RFC 5646) primary
3977/// language subtag, as required by RFC 8216bis `EXT-X-MEDIA` `LANGUAGE`.
3978///
3979/// HLS `LANGUAGE` must be an RFC 5646 (BCP 47) language tag; in BCP 47 the
3980/// primary subtag is the ISO 639-1 two-letter code when one exists, otherwise
3981/// the ISO 639-2/T (== ISO 639-3) three-letter code. ISO 639-2/B
3982/// bibliographic codes (`fre`, `ger`, `dut`, `chi`, ...) are not valid BCP 47
3983/// subtags and are mapped to their terminology (T) equivalents first.
3984///
3985/// `qaa`..`qtz` (private use), `und`, `mul`, `mis` and `zxx` pass through
3986/// unchanged. Unknown or malformed codes return `None`.
3987pub fn bcp47_language(iso639_2: &str) -> Option<String> {
3988 let code = iso639_2.trim().to_ascii_lowercase();
3989 if code.len() != 3 || !code.bytes().all(|b| b.is_ascii_lowercase()) {
3990 return None;
3991 }
3992 // Private-use range and the reserved singleton codes pass through verbatim.
3993 if code == "und" || code == "mul" || code == "mis" || code == "zxx" {
3994 return Some(code);
3995 }
3996 if code.starts_with('q') {
3997 let bytes = code.as_bytes();
3998 if (b'a'..=b't').contains(&bytes[1]) && bytes[2].is_ascii_lowercase() {
3999 return Some(code);
4000 }
4001 }
4002 let lang = iso_639::Language::from_str(&code)?;
4003 if let Some(a2) = lang.alpha2() {
4004 return Some(a2.to_string());
4005 }
4006 if let Some(t) = lang.alpha3_t() {
4007 return Some(t.to_string());
4008 }
4009 lang.alpha3_b().map(|b| b.to_string())
4010}
4011
4012/// Relative URI from one playlist to another (RFC 3986 relative reference),
4013/// computed over slash-separated path segments. The carrying playlist's own
4014/// filename is dropped before differencing, so two playlists in the same
4015/// directory resolve to a bare filename.
4016fn relative_playlist_uri(from: &str, to: &str) -> String {
4017 let from_dir: Vec<&str> = match from.rfind('/') {
4018 Some(i) => from[..i].split('/').filter(|s| !s.is_empty()).collect(),
4019 None => Vec::new(),
4020 };
4021 let to_segs: Vec<&str> = to.split('/').filter(|s| !s.is_empty()).collect();
4022 let mut common = 0usize;
4023 while common < from_dir.len() && common < to_segs.len() && from_dir[common] == to_segs[common] {
4024 common += 1;
4025 }
4026 let mut parts: Vec<&str> = Vec::new();
4027 parts.extend(std::iter::repeat_n(
4028 "..",
4029 from_dir.len().saturating_sub(common),
4030 ));
4031 parts.extend(&to_segs[common..]);
4032 if parts.is_empty() {
4033 // Same path: emit the target's last segment (or empty).
4034 to_segs.last().copied().unwrap_or("").to_string()
4035 } else {
4036 parts.join("/")
4037 }
4038}
4039
4040#[cfg(test)]
4041mod tests {
4042 #[test]
4043 fn a_segment_with_fewer_than_two_samples_has_no_measurable_frame_rate() {
4044 assert_eq!(super::segment_frame_rate_millihertz(1, 6000, 90_000), None);
4045 assert_eq!(
4046 super::segment_frame_rate_millihertz(2, 6000, 90_000),
4047 Some(30_000)
4048 );
4049 assert_eq!(super::segment_frame_rate_millihertz(2, 0, 90_000), None);
4050 }
4051
4052 #[test]
4053 fn a_segment_with_broken_timing_does_not_yield_an_implausible_frame_rate() {
4054 assert_eq!(super::segment_frame_rate_millihertz(2, 1, 90_000), None);
4055 assert_eq!(
4056 super::segment_frame_rate_millihertz(300, 90_000, 90_000),
4057 Some(300_000)
4058 );
4059 assert_eq!(
4060 super::segment_frame_rate_millihertz(301, 90_000, 90_000),
4061 None
4062 );
4063 }
4064
4065 use super::*;
4066 use media_plane::trunk::{EventAnchor, TrunkConfig};
4067 use std::time::{Duration, Instant};
4068 use timed_metadata::{MediaTime, TimeAnchor};
4069 use transmux::SegmentMeta;
4070
4071 fn nz(n: usize) -> NonZeroUsize {
4072 NonZeroUsize::new(n).expect("test capacity must be non-zero")
4073 }
4074
4075 /// A fresh `Trunk` sized generously for these tests, plus the one
4076 /// `HlsOrigin` under test — `Fmp4` + low-latency, matching every
4077 /// pre-#873 test in this module (regression guard).
4078 fn make_origin() -> (Arc<Trunk>, HlsOrigin, media_plane::trunk::SegmentWriter) {
4079 let trunk = Trunk::new(TrunkConfig::new(nz(64), nz(8), nz(8), nz(8), nz(64)));
4080 let writer = trunk.segment_writer().expect("first segment writer");
4081 let origin = HlsOrigin::builder(Arc::clone(&trunk))
4082 .target_duration_secs(4.0)
4083 .window_segments(nz(4))
4084 .low_latency(500)
4085 .instance(7)
4086 .build()
4087 .expect("both required fields set");
4088 origin.set_init(vec![0xAAu8; 8]);
4089 (trunk, origin, writer)
4090 }
4091
4092 /// #1159: the resource-naming functions take distinct id types, and every
4093 /// argument lands in its own slot of the wire name. All six numbers differ
4094 /// (track 3, instance 7, generation 4, sequence 5, part 2, offset 100), so
4095 /// a swapped or misrouted argument changes the exact string.
4096 #[test]
4097 fn resource_names_place_every_distinct_id_in_its_own_slot() {
4098 let (_trunk, origin, _writer) = make_origin();
4099 let track = TrackId::new(3);
4100 assert_eq!(origin.instance(), 7);
4101 assert_eq!(
4102 origin.init_name(track, Generation::new(4)),
4103 "init-3-7-4.mp4"
4104 );
4105 assert_eq!(
4106 origin.segment_name(track, SegmentSeq::new(5), "m4s"),
4107 "seg-3-7-5.m4s"
4108 );
4109 assert_eq!(
4110 origin.part_name(track, SegmentSeq::new(5), PartIndex::new(2), "m4s"),
4111 "part-3-7-5.2.m4s"
4112 );
4113
4114 let trunk = Trunk::new(TrunkConfig::new(nz(64), nz(8), nz(8), nz(8), nz(64)));
4115 let offset = HlsOrigin::builder(trunk)
4116 .target_duration_secs(4.0)
4117 .window_segments(nz(4))
4118 .instance(7)
4119 .media_sequence_offset(100)
4120 .build()
4121 .expect("both required fields set");
4122 assert_eq!(
4123 offset.segment_name(track, SegmentSeq::new(5), "m4s"),
4124 "seg-3-7-105.m4s",
4125 "the Media Sequence Number offset applies to the sequence slot only"
4126 );
4127 assert_eq!(
4128 offset.part_name(track, SegmentSeq::new(5), PartIndex::new(2), "m4s"),
4129 "part-3-7-105.2.m4s"
4130 );
4131 assert_eq!(
4132 offset.init_name(track, Generation::new(4)),
4133 "init-3-7-4.mp4",
4134 "the offset never touches the generation"
4135 );
4136 }
4137
4138 fn seg(
4139 writer: &media_plane::trunk::SegmentWriter,
4140 seq: u32,
4141 duration_secs: f64,
4142 discontinuous: bool,
4143 ) {
4144 writer
4145 .publish_segment(SegmentEntry::new(
4146 Bytes::from(vec![seq as u8; 8]),
4147 SegmentSeq::new(seq),
4148 Duration::from_secs_f64(duration_secs),
4149 Timestamp::from_nanos(0),
4150 SegmentMeta { discontinuous },
4151 ))
4152 .unwrap();
4153 }
4154
4155 fn part(writer: &media_plane::trunk::SegmentWriter, seg_no: u32, idx: u32, independent: bool) {
4156 writer.publish_part(PartEntry::new(
4157 Bytes::from(vec![idx as u8; 4]),
4158 SegmentSeq::new(seg_no),
4159 PartIndex::new(idx),
4160 Duration::from_millis(500),
4161 independent,
4162 ));
4163 }
4164
4165 fn resolve_now(origin: &HlsOrigin, request: HlsRequest) -> EgressResponse<HlsBody> {
4166 origin.resolve(
4167 request,
4168 Timestamp::from_nanos(0),
4169 AwaitPolicy::new(Timestamp::from_nanos(0)),
4170 )
4171 }
4172
4173 // --- EXT-X-DATERANGE rendering from the trunk event ring (#965) -----
4174
4175 /// MUTATION VERIFIED: dropping the `events_in_segment` loop (or the
4176 /// `time_anchor()` gate) from `render_playlist` makes all three
4177 /// assertions below fail — the playlist no longer carries the
4178 /// `#EXT-X-DATERANGE` line, its `SCTE35-OUT=` attribute, or its
4179 /// `ID="2002"`. `if let Ok(dr)` (rather than `unwrap`) is the deliberate
4180 /// skip, not asserted here, because `time_anchor` is set in this test so
4181 /// no event legitimately fails conversion.
4182 #[test]
4183 fn trunk_events_render_as_ext_x_daterange() {
4184 // splice_insert, event_id=2002, pts=2_160_000 (~24s at 90kHz) — the
4185 // same fixture timed-metadata's own timeline tests use.
4186 let hex = "FC302100000000000000FFF01005000007D27FEF7F7E0020F580C0000000000088B9661D";
4187 let splice_bytes: Vec<u8> = (0..hex.len())
4188 .step_by(2)
4189 .map(|i| u8::from_str_radix(&hex[i..i + 2], 16).unwrap())
4190 .collect();
4191
4192 let (trunk, origin, writer) = make_origin();
4193
4194 // Give the event log a wall-clock anchor so drafts can carry a
4195 // START-DATE, and a media-clock one so the event resolves immediately.
4196 writer.set_time_anchor(TimeAnchor {
4197 pts_90k: 0,
4198 utc_epoch_ms: 1_000_000_000_000,
4199 });
4200
4201 // Parse the splice into a TimedEvent and publish it to the event ring,
4202 // Media-anchored at its own PTS (2_160_000, unwrapped on first call).
4203 let mut timeline = timed_metadata::Timeline::new();
4204 let ev = timeline.push_scte35(&splice_bytes).unwrap();
4205 let trunk_writer = trunk.writer().expect("first trunk writer");
4206 trunk_writer.publish_event(
4207 ev.clone(),
4208 EventAnchor::Media(ev.at.unwrap_or(MediaTime(0))),
4209 );
4210
4211 // Segment 1 spans [0, 4_000_000) — includes the cue's PTS.
4212 writer.note_segment_start(SegmentSeq::new(1), MediaTime(0));
4213 writer.note_segment_start(SegmentSeq::new(2), MediaTime(4_000_000));
4214 seg(&writer, 1, 4.0, false);
4215
4216 let playlist = origin
4217 .render_playlist(DEFAULT_TRACK_ID, None)
4218 .expect("render");
4219 assert!(
4220 playlist.contains("#EXT-X-DATERANGE"),
4221 "playlist should contain EXT-X-DATERANGE, got: {playlist}"
4222 );
4223 assert!(
4224 playlist.contains("SCTE35-OUT="),
4225 "daterange should carry SCTE35-OUT attribute: {playlist}"
4226 );
4227 assert!(
4228 playlist.contains("ID=\"2002\""),
4229 "daterange ID should be the event_id: {playlist}"
4230 );
4231 }
4232
4233 /// MUTATION VERIFIED: replacing the `if let Ok(dr)` guard in
4234 /// `render_playlist` with a bare `timeline.to_daterange(&entry.event)`
4235 /// (i.e. unwrapping the `Err`) makes this test fail — an Emsg-sourced
4236 /// event that `to_daterange` rejects ("event is not SCTE-35-sourced")
4237 /// would panic `render_playlist` instead of being skipped. This test
4238 /// pins that skip: with a time anchor set and the event inside the
4239 /// segment span, no `#EXT-X-DATERANGE` line may appear (and rendering
4240 /// must not panic).
4241 #[test]
4242 fn non_scte35_events_are_silently_skipped() {
4243 use timed_metadata::event::{EventKind, SourcePayload};
4244 use timed_metadata::{MediaDuration, TimedEvent};
4245
4246 let (trunk, origin, writer) = make_origin();
4247 writer.set_time_anchor(TimeAnchor {
4248 pts_90k: 0,
4249 utc_epoch_ms: 1_000_000_000_000,
4250 });
4251
4252 // An Emsg-sourced event, Media-anchored inside segment 1's span — the
4253 // kinds of event that legitimately cannot become a DATERANGE.
4254 let emsg_ev = TimedEvent {
4255 id: Some(7),
4256 kind: EventKind::Unspecified,
4257 at: Some(MediaTime(2_000_000)),
4258 duration: Some(MediaDuration(2_000_000)),
4259 source: SourcePayload::Emsg {
4260 scheme_id_uri: "urn:example".to_string(),
4261 value: "1".to_string(),
4262 raw: vec![0x01, 0x02],
4263 },
4264 };
4265 let trunk_writer = trunk.writer().expect("first trunk writer");
4266 trunk_writer.publish_event(emsg_ev, EventAnchor::Media(MediaTime(2_000_000)));
4267
4268 writer.note_segment_start(SegmentSeq::new(1), MediaTime(0));
4269 writer.note_segment_start(SegmentSeq::new(2), MediaTime(4_000_000));
4270 seg(&writer, 1, 4.0, false);
4271
4272 let playlist = origin
4273 .render_playlist(DEFAULT_TRACK_ID, None)
4274 .expect("render");
4275 assert!(
4276 !playlist.contains("#EXT-X-DATERANGE"),
4277 "no DATERANGE expected for a non-SCTE-35 event: {playlist}"
4278 );
4279 }
4280
4281 // --- master playlist (unaffected by the Trunk migration) -------------
4282
4283 #[test]
4284 fn master_playlist_has_stream_inf() {
4285 let m = master_playlist_m3u8("media.m3u8", 5_000_000);
4286 assert!(m.contains("#EXTM3U"));
4287 assert!(m.contains("#EXT-X-STREAM-INF"));
4288 assert!(m.contains("media.m3u8"));
4289 }
4290
4291 #[test]
4292 fn master_playlist_points_at_configured_playlist_name() {
4293 let m = master_playlist_m3u8("index.m3u8", 5_000_000);
4294 assert!(m.contains("index.m3u8"));
4295 assert!(!m.contains("media.m3u8"));
4296 }
4297
4298 // --- 1. playlist rendered from a populated Trunk matches the expected
4299 // shape ---------------------------------------------------------
4300
4301 /// MUTATION VERIFIED: changing `render_playlist`'s
4302 /// `low_latency: Some(...)` to `None` makes this test's
4303 /// `assert!(m.contains("#EXT-X-PART-INF"))` (and every other
4304 /// LL-HLS-tag assertion) fail — `to_m3u8()` omits the entire
4305 /// low-latency header block when `low_latency` is `None`, so none of
4306 /// `#EXT-X-PART-INF`/`#EXT-X-SERVER-CONTROL`/`#EXT-X-PART` appear in the
4307 /// rendered body. Recompiled and re-run to confirm the failure, then
4308 /// reverted.
4309 #[test]
4310 fn playlist_rendered_from_populated_trunk_matches_expected_shape() {
4311 let (_trunk, origin, writer) = make_origin();
4312 seg(&writer, 1, 4.0, false);
4313 part(&writer, 2, 0, true);
4314 part(&writer, 2, 1, false);
4315
4316 let body = match resolve_now(
4317 &origin,
4318 HlsRequest::Playlist {
4319 track_id: DEFAULT_TRACK_ID,
4320 query: BlockingQuery::default(),
4321 },
4322 ) {
4323 EgressResponse::Ready {
4324 body: HlsBody::Playlist(m),
4325 cache,
4326 } => {
4327 assert_eq!(cache, CachePolicy::NoCache);
4328 m
4329 }
4330 other => panic!("expected Ready(Playlist), got {other:?}"),
4331 };
4332
4333 // RFC 8216bis §8 (issue #871): this playlist's true minimum is 6
4334 // (EXT-X-MAP without EXT-X-I-FRAMES-ONLY) — none of the LL-HLS
4335 // directives it also carries require any version at all. The old
4336 // hardcoded `EXT-X-VERSION:9` over-declared and would have locked
4337 // out every client on protocol version 6, 7, or 8.
4338 assert!(body.contains("#EXT-X-VERSION:6"), "body: {body}");
4339 assert!(!body.contains("#EXT-X-VERSION:9"), "body: {body}");
4340 assert!(body.contains("#EXT-X-TARGETDURATION:4"), "body: {body}");
4341 assert!(
4342 body.contains("#EXT-X-SERVER-CONTROL:CAN-BLOCK-RELOAD=YES,PART-HOLD-BACK=1.5"),
4343 "body: {body}"
4344 );
4345 assert!(
4346 body.contains("#EXT-X-PART-INF:PART-TARGET=0.5"),
4347 "body: {body}"
4348 );
4349 assert!(
4350 body.contains("#EXT-X-MAP:URI=\"init-1-7-1.mp4\""),
4351 "body: {body}"
4352 );
4353 assert!(body.contains("seg-1-7-1.m4s"), "body: {body}");
4354 assert!(
4355 body.contains("#EXT-X-PART:DURATION=0.5") && body.contains("INDEPENDENT=YES"),
4356 "body: {body}"
4357 );
4358 assert!(body.contains("#EXT-X-PRELOAD-HINT"), "body: {body}");
4359 assert!(
4360 body.contains("part-1-7-2.2.m4s"),
4361 "preload hint for the next part: {body}"
4362 );
4363 }
4364
4365 /// Issue #1577. RFC 8216bis Appendix B.1: a Playlist with an EXT-X-PART
4366 /// tag and no EXT-X-ENDLIST names the next Partial Segment in an
4367 /// EXT-X-PRELOAD-HINT. Right after segment 1 closes, with no part of
4368 /// segment 2 yet, the hint is part 0 of segment 2, and the server holds a
4369 /// request for that resource open (section 6.2.6) rather than 404ing it.
4370 #[test]
4371 fn preload_hint_present_between_segments() {
4372 let (_trunk, origin, writer) = make_origin();
4373 part(&writer, 1, 0, true);
4374 part(&writer, 1, 1, false);
4375 seg(&writer, 1, 4.0, false);
4376
4377 let body = render_body(&origin);
4378 assert!(body.contains("#EXT-X-PART:"), "body: {body}");
4379 assert!(
4380 body.contains("#EXT-X-PRELOAD-HINT:TYPE=PART,URI=\"part-1-7-2.0.m4s\""),
4381 "hint names part 0 of the next segment: {body}"
4382 );
4383
4384 let first = origin.resolve(
4385 HlsRequest::Resource {
4386 name: "part-1-7-2.0.m4s".to_string(),
4387 },
4388 Timestamp::from_nanos(0),
4389 AwaitPolicy::new(Timestamp::from_nanos(5_000_000_000)),
4390 );
4391 assert!(
4392 matches!(first, EgressResponse::Await { .. }),
4393 "the hinted resource blocks, got {first:?}"
4394 );
4395 part(&writer, 2, 0, true);
4396 let served = resolve_now(
4397 &origin,
4398 HlsRequest::Resource {
4399 name: "part-1-7-2.0.m4s".to_string(),
4400 },
4401 );
4402 assert!(
4403 matches!(served, EgressResponse::Ready { .. }),
4404 "served once produced, got {served:?}"
4405 );
4406 }
4407
4408 // --- 2. a preload-hinted part BLOCKS until produced, then serves -----
4409
4410 /// MUTATION VERIFIED: changing `resolve_resource`'s `never_will` check
4411 /// (whether the requested part's segment has already closed, via
4412 /// `last_closed_segment`) to always `true` ("never will produce this
4413 /// part") makes this test's first assertion fail: the not-yet-produced
4414 /// part resolves `NotFound` immediately instead of `Await`, so
4415 /// `assert!(matches!(first, EgressResponse::Await { .. }))` sees
4416 /// `NotFound` and fails. Recompiled and re-run to confirm the failure,
4417 /// then reverted. This is the RFC 8216bis section 6.2.2 behaviour that
4418 /// shipped as multimux 0.2.1's bug fix — regressing it would break the
4419 /// live camera route.
4420 #[test]
4421 fn preload_hinted_part_blocks_until_produced_then_serves() {
4422 let (trunk, origin, writer) = make_origin();
4423 let origin = Arc::new(origin);
4424
4425 // Not produced yet: must Await, not NotFound.
4426 let deadline = Timestamp::from_nanos(5_000_000_000);
4427 let policy = AwaitPolicy::new(deadline);
4428 let first = origin.resolve(
4429 HlsRequest::Resource {
4430 name: "part-1-7-1.0.m4s".to_string(),
4431 },
4432 Timestamp::from_nanos(0),
4433 policy,
4434 );
4435 assert!(
4436 matches!(first, EgressResponse::Await { .. }),
4437 "expected Await before the part exists, got {first:?}"
4438 );
4439
4440 // Register a real Trunk::listen() wake-up and block a worker thread
4441 // on it -- the actual mechanism a real adapter (Step 5) uses, not a
4442 // poll loop -- to prove the part genuinely blocks rather than
4443 // merely returning Await once and never resolving.
4444 let listener = trunk.listen().expect("listener slot available");
4445 let woken = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false));
4446 let woken2 = std::sync::Arc::clone(&woken);
4447 // HANG GUARD (issue #807): deliberately generous, same reasoning as
4448 // `media-plane/src/trunk.rs`'s own `Trunk::listen()` wake tests --
4449 // the claim is "wakes rather than parking forever", not "wakes
4450 // within N seconds"; the publish happens on another thread, so a
4451 // tight bound would measure the machine's scheduler, not this code.
4452 let waiter = std::thread::spawn(move || {
4453 let ok = listener.wait_deadline(Instant::now() + Duration::from_secs(60));
4454 woken2.store(ok, std::sync::atomic::Ordering::SeqCst);
4455 });
4456
4457 // Produce the part the request was waiting on.
4458 part(&writer, 1, 0, true);
4459
4460 waiter.join().expect("waiter thread must not panic");
4461 assert!(
4462 woken.load(std::sync::atomic::Ordering::SeqCst),
4463 "Trunk::listen() must wake once publish_part lands"
4464 );
4465
4466 // Re-resolving now must serve it -- not 404.
4467 match origin.resolve(
4468 HlsRequest::Resource {
4469 name: "part-1-7-1.0.m4s".to_string(),
4470 },
4471 Timestamp::from_nanos(1),
4472 policy,
4473 ) {
4474 EgressResponse::Ready {
4475 body: HlsBody::Resource(bytes),
4476 cache,
4477 } => {
4478 assert_eq!(bytes, Bytes::from(vec![0u8; 4]));
4479 assert_eq!(cache, CachePolicy::Immutable);
4480 }
4481 other => panic!("expected Ready once produced, got {other:?}"),
4482 }
4483 }
4484
4485 /// MUTATION VERIFIED: removing `EgressResponse::pending`'s expiry check
4486 /// (i.e. always returning `Await`) would make a client wait forever for
4487 /// a part that will never exist -- this test proves the OTHER half of
4488 /// the bound: once `now` reaches the caller's own `AwaitPolicy::deadline`,
4489 /// resolve must stop Awaiting. Changing the deadline comparison in
4490 /// `resolve_resource`'s `EgressResponse::pending(await_policy, now, now)`
4491 /// call to ignore `now` (always pass `Timestamp::from_nanos(0)`) makes
4492 /// this test's final assertion fail: `resolve` at `now == deadline`
4493 /// keeps returning `Await` instead of `NotFound`. Recompiled and re-run
4494 /// to confirm the failure, then reverted.
4495 #[test]
4496 fn awaiting_part_is_bounded_by_await_policy_deadline() {
4497 let (_trunk, origin, _writer) = make_origin();
4498 let deadline = Timestamp::from_nanos(1_000_000_000);
4499 let policy = AwaitPolicy::new(deadline);
4500
4501 let still_waiting = origin.resolve(
4502 HlsRequest::Resource {
4503 name: "part-1-7-1.0.m4s".to_string(),
4504 },
4505 Timestamp::from_nanos(999_999_999),
4506 policy,
4507 );
4508 assert!(matches!(still_waiting, EgressResponse::Await { .. }));
4509
4510 let expired = origin.resolve(
4511 HlsRequest::Resource {
4512 name: "part-1-7-1.0.m4s".to_string(),
4513 },
4514 deadline,
4515 policy,
4516 );
4517 assert!(
4518 matches!(expired, EgressResponse::NotFound),
4519 "expected NotFound once the deadline passed, got {expired:?}"
4520 );
4521 }
4522
4523 // --- 3. a just-closed segment's final part still serves ---------------
4524
4525 /// MUTATION VERIFIED: this behaviour depends entirely on
4526 /// `media_plane::trunk::SegmentWriter::publish_segment` (`media-plane/src/trunk.rs`) never
4527 /// touching the live-part log. Simulating the old `MediaStore` bug here
4528 /// by having `resolve_resource` check `last_closed_segment() >= seq`
4529 /// ("this segment already closed -> NotFound") **before** checking
4530 /// `Trunk::part_bytes` (i.e. swapping the two checks' order) makes this
4531 /// test's first assertion fail: the just-closed segment's final part
4532 /// resolves `NotFound` instead of `Ready` (`panicked at ...: the
4533 /// just-closed segment's final part must still serve, got NotFound`),
4534 /// because the eager closed-check now shadows the still-valid
4535 /// `part_bytes` hit. Recompiled and re-run to confirm the failure, then
4536 /// reverted. This is the RFC 8216bis boundary behaviour that shipped as
4537 /// multimux 0.2.2's bug fix — regressing it would break the live camera
4538 /// route (its own `#EXT-X-PRELOAD-HINT` part races exactly this
4539 /// boundary every segment).
4540 #[test]
4541 fn just_closed_segment_final_part_still_serves() {
4542 let (_trunk, origin, writer) = make_origin();
4543 part(&writer, 1, 0, true);
4544 part(&writer, 1, 1, false); // segment 1's final part
4545 seg(&writer, 1, 4.0, false); // close segment 1
4546
4547 match resolve_now(
4548 &origin,
4549 HlsRequest::Resource {
4550 name: "part-1-7-1.1.m4s".to_string(),
4551 },
4552 ) {
4553 EgressResponse::Ready {
4554 body: HlsBody::Resource(bytes),
4555 ..
4556 } => assert_eq!(bytes, Bytes::from(vec![1u8; 4])),
4557 other => panic!("the just-closed segment's final part must still serve, got {other:?}"),
4558 }
4559
4560 // A genuinely-nonexistent part of the closed segment is NotFound.
4561 assert_eq!(
4562 resolve_now(
4563 &origin,
4564 HlsRequest::Resource {
4565 name: "part-1-7-1.9.m4s".to_string(),
4566 }
4567 ),
4568 EgressResponse::NotFound
4569 );
4570
4571 // The playlist must not resurrect the closed segment's parts as
4572 // "open" -- it is rendered whole.
4573 let body = match resolve_now(
4574 &origin,
4575 HlsRequest::Playlist {
4576 track_id: DEFAULT_TRACK_ID,
4577 query: BlockingQuery::default(),
4578 },
4579 ) {
4580 EgressResponse::Ready {
4581 body: HlsBody::Playlist(m),
4582 ..
4583 } => m,
4584 other => panic!("expected Ready(Playlist), got {other:?}"),
4585 };
4586 assert!(
4587 body.contains("seg-1-7-1.m4s"),
4588 "closed segment rendered whole: {body}"
4589 );
4590 // RFC 8216bis §6.2.2: the closed segment keeps its EXT-X-PART tags
4591 // (inside the segment, before its EXTINF) for three Target
4592 // Durations. They are not an *open* segment, and every part line
4593 // precedes the segment's own EXTINF/URI. The playlist carries parts,
4594 // so Appendix B.1 requires a preload hint: part 0 of the next
4595 // segment (#1577).
4596 assert_eq!(
4597 body.lines().skip(5).collect::<Vec<_>>(),
4598 vec![
4599 "#EXT-X-SERVER-CONTROL:CAN-BLOCK-RELOAD=YES,PART-HOLD-BACK=1.5,HOLD-BACK=12,CAN-SKIP-UNTIL=24,CAN-SKIP-DATERANGES=YES",
4600 "#EXT-X-PART-INF:PART-TARGET=0.5",
4601 "#EXT-X-MAP:URI=\"init-1-7-1.mp4\"",
4602 "#EXT-X-PART:DURATION=0.5,URI=\"part-1-7-1.0.m4s\",INDEPENDENT=YES",
4603 "#EXT-X-PART:DURATION=0.5,URI=\"part-1-7-1.1.m4s\"",
4604 "#EXTINF:4,",
4605 "seg-1-7-1.m4s",
4606 "#EXT-X-PRELOAD-HINT:TYPE=PART,URI=\"part-1-7-2.0.m4s\"",
4607 ],
4608 "{body}"
4609 );
4610 }
4611
4612 // --- 4. MEDIA-SEQUENCE / DISCONTINUITY-SEQUENCE advance as the window
4613 // rolls -----------------------------------------------------
4614
4615 /// MUTATION VERIFIED: changing `Window::push`'s eviction guard from
4616 /// `if evicted.discontinuous` to `if false` (never counting an evicted
4617 /// discontinuity) makes this test's
4618 /// `assert!(body.contains("#EXT-X-DISCONTINUITY-SEQUENCE:1"))` fail --
4619 /// the tag is omitted entirely (the renderer only emits it when
4620 /// `discontinuity_sequence > 0`), because the counter never advances
4621 /// past `0`. Recompiled and re-run to confirm the failure, then
4622 /// reverted.
4623 #[test]
4624 fn media_sequence_and_discontinuity_sequence_advance_as_window_rolls() {
4625 let (_trunk, origin, writer) = make_origin(); // window_segments = 4
4626
4627 seg(&writer, 1, 4.0, false);
4628 seg(&writer, 2, 4.0, true); // discontinuous
4629 seg(&writer, 3, 4.0, false);
4630 seg(&writer, 4, 4.0, false);
4631
4632 // Window (capacity 4) holds exactly 1..=4 -- MEDIA-SEQUENCE=1, and
4633 // segment 2's own #EXT-X-DISCONTINUITY renders in-window (no
4634 // DISCONTINUITY-SEQUENCE yet, nothing has rolled off).
4635 let body = match resolve_now(
4636 &origin,
4637 HlsRequest::Playlist {
4638 track_id: DEFAULT_TRACK_ID,
4639 query: BlockingQuery::default(),
4640 },
4641 ) {
4642 EgressResponse::Ready {
4643 body: HlsBody::Playlist(m),
4644 ..
4645 } => m,
4646 other => panic!("expected Ready(Playlist), got {other:?}"),
4647 };
4648 assert!(body.contains("#EXT-X-MEDIA-SEQUENCE:1"), "body: {body}");
4649 assert!(
4650 body.contains("#EXT-X-DISCONTINUITY-SEQUENCE:0"),
4651 "fresh live playlist always carries DISCONTINUITY-SEQUENCE:0 (Apple 8.17): {body}"
4652 );
4653 assert!(body.contains("#EXT-X-DISCONTINUITY\n"), "body: {body}");
4654
4655 // Roll the window: segment 5 evicts segment 1 (not discontinuous;
4656 // DISCONTINUITY-SEQUENCE stays 0), segment 6 evicts segment 2
4657 // (discontinuous -- DISCONTINUITY-SEQUENCE becomes 1).
4658 seg(&writer, 5, 4.0, false);
4659 let body = match resolve_now(
4660 &origin,
4661 HlsRequest::Playlist {
4662 track_id: DEFAULT_TRACK_ID,
4663 query: BlockingQuery::default(),
4664 },
4665 ) {
4666 EgressResponse::Ready {
4667 body: HlsBody::Playlist(m),
4668 ..
4669 } => m,
4670 other => panic!("expected Ready(Playlist), got {other:?}"),
4671 };
4672 assert!(body.contains("#EXT-X-MEDIA-SEQUENCE:2"), "body: {body}");
4673 assert!(
4674 body.contains("#EXT-X-DISCONTINUITY-SEQUENCE:0"),
4675 "evicted segment 1 was not discontinuous: {body}"
4676 );
4677
4678 seg(&writer, 6, 4.0, false);
4679 let body = match resolve_now(
4680 &origin,
4681 HlsRequest::Playlist {
4682 track_id: DEFAULT_TRACK_ID,
4683 query: BlockingQuery::default(),
4684 },
4685 ) {
4686 EgressResponse::Ready {
4687 body: HlsBody::Playlist(m),
4688 ..
4689 } => m,
4690 other => panic!("expected Ready(Playlist), got {other:?}"),
4691 };
4692 assert!(body.contains("#EXT-X-MEDIA-SEQUENCE:3"), "body: {body}");
4693 assert!(
4694 body.contains("#EXT-X-DISCONTINUITY-SEQUENCE:1"),
4695 "segment 2 (discontinuous) has now rolled off the window: {body}"
4696 );
4697 }
4698
4699 // --- misc: target-duration MUST, abuse bound, bad request -------------
4700
4701 #[test]
4702 fn published_target_duration_matches_the_rendered_playlist() {
4703 let (_trunk, origin, writer) = make_origin(); // configured target 4.0
4704 assert_eq!(origin.published_target_duration(), Duration::from_secs(4));
4705 seg(&writer, 1, 7.5, false);
4706 let _ = resolve_now(
4707 &origin,
4708 HlsRequest::Playlist {
4709 track_id: DEFAULT_TRACK_ID,
4710 query: BlockingQuery::default(),
4711 },
4712 );
4713 assert_eq!(origin.published_target_duration(), Duration::from_secs(8));
4714 seg(&writer, 2, 20.0, false);
4715 assert_eq!(origin.published_target_duration(), Duration::from_secs(8));
4716 }
4717
4718 #[test]
4719 fn target_duration_is_max_of_configured_and_actual_segment_duration() {
4720 let (_trunk, origin, writer) = make_origin(); // configured target 4.0
4721 seg(&writer, 1, 7.5, false);
4722 let body = match resolve_now(
4723 &origin,
4724 HlsRequest::Playlist {
4725 track_id: DEFAULT_TRACK_ID,
4726 query: BlockingQuery::default(),
4727 },
4728 ) {
4729 EgressResponse::Ready {
4730 body: HlsBody::Playlist(m),
4731 ..
4732 } => m,
4733 other => panic!("expected Ready(Playlist), got {other:?}"),
4734 };
4735 assert!(
4736 body.contains("#EXT-X-TARGETDURATION:8"),
4737 "TARGETDURATION must be round(7.5)=8, not the configured target: {body}"
4738 );
4739 }
4740
4741 #[test]
4742 fn far_future_msn_rejected() {
4743 let (_trunk, origin, writer) = make_origin();
4744 seg(&writer, 1, 4.0, false);
4745 let outcome = resolve_now(
4746 &origin,
4747 HlsRequest::Playlist {
4748 track_id: DEFAULT_TRACK_ID,
4749 query: BlockingQuery {
4750 hls_msn: Some(1002),
4751 hls_part: None,
4752 hls_skip: None,
4753 hls_primary: None,
4754 },
4755 },
4756 );
4757 assert!(matches!(outcome, EgressResponse::BadRequest { .. }));
4758 }
4759
4760 /// RFC 8216bis §6.2.5.2: `_HLS_msn` at `last_closed + 2` is the
4761 /// spec's bound — it MUST be accepted. With one segment closed (seq=1),
4762 /// the last closed is 1, so `_HLS_msn=3` (last_closed + 2) is accepted
4763 /// and the request blocks (returns Pending, not BadRequest).
4764 #[test]
4765 fn msn_at_spec_bound_is_accepted() {
4766 let (_trunk, origin, writer) = make_origin();
4767 seg(&writer, 1, 4.0, false);
4768 let outcome = resolve_now(
4769 &origin,
4770 HlsRequest::Playlist {
4771 track_id: DEFAULT_TRACK_ID,
4772 query: BlockingQuery {
4773 hls_msn: Some(3),
4774 hls_part: None,
4775 hls_skip: None,
4776 hls_primary: None,
4777 },
4778 },
4779 );
4780 assert!(
4781 !matches!(outcome, EgressResponse::BadRequest { .. }),
4782 "msn at spec bound (last_closed+2) must be accepted, not rejected"
4783 );
4784 }
4785
4786 /// RFC 8216bis §6.2.5.2: `_HLS_msn` one beyond the spec's +2 SHOULD
4787 /// boundary is rejected. With one segment closed (seq=1, last_closed=1),
4788 /// the live edge in_progress_seg is 1 (not 2, because no parts exist to
4789 /// advance it), so `_HLS_msn=4` (1 + 2 + 1) is rejected.
4790 #[test]
4791 fn msn_one_beyond_spec_bound_is_rejected() {
4792 let (_trunk, origin, writer) = make_origin();
4793 seg(&writer, 1, 4.0, false);
4794 // With segments up to 1 closed, in_progress_seg is 1.
4795 // _HLS_msn > 1 + 2 == 3 → rejected. So msn=4 is rejected.
4796 let outcome = resolve_now(
4797 &origin,
4798 HlsRequest::Playlist {
4799 track_id: DEFAULT_TRACK_ID,
4800 query: BlockingQuery {
4801 hls_msn: Some(4),
4802 hls_part: None,
4803 hls_skip: None,
4804 hls_primary: None,
4805 },
4806 },
4807 );
4808 assert!(
4809 matches!(outcome, EgressResponse::BadRequest { .. }),
4810 "msn at spec bound + 1 (last_closed+3) must be rejected"
4811 );
4812 }
4813
4814 #[test]
4815 fn part_without_msn_rejected() {
4816 let (_trunk, origin, _writer) = make_origin();
4817 let outcome = resolve_now(
4818 &origin,
4819 HlsRequest::Playlist {
4820 track_id: DEFAULT_TRACK_ID,
4821 query: BlockingQuery {
4822 hls_msn: None,
4823 hls_part: Some(0),
4824 hls_skip: None,
4825 hls_primary: None,
4826 },
4827 },
4828 );
4829 assert!(matches!(outcome, EgressResponse::BadRequest { .. }));
4830 }
4831
4832 #[test]
4833 fn resolve_resource_init_present() {
4834 let (_trunk, origin, _writer) = make_origin();
4835 match resolve_now(
4836 &origin,
4837 HlsRequest::Resource {
4838 name: "init-1.mp4".to_string(),
4839 },
4840 ) {
4841 EgressResponse::Ready {
4842 body: HlsBody::Resource(bytes),
4843 cache,
4844 } => {
4845 assert_eq!(bytes, Bytes::from(vec![0xAAu8; 8]));
4846 // The bare name means "the current init" and can change.
4847 assert_eq!(cache, CachePolicy::NoCache);
4848 }
4849 other => panic!("expected Ready, got {other:?}"),
4850 }
4851 }
4852
4853 #[test]
4854 fn resolve_resource_unmatched_filename_not_found() {
4855 let (_trunk, origin, _writer) = make_origin();
4856 assert_eq!(
4857 resolve_now(
4858 &origin,
4859 HlsRequest::Resource {
4860 name: "not-a-thing.txt".to_string(),
4861 }
4862 ),
4863 EgressResponse::NotFound
4864 );
4865 }
4866
4867 // --- issue #873: Container x low-latency matrix -----------------------
4868 //
4869 // Every cell below asserts on rendered playlist text AND on request
4870 // resolution (fetch every URI the rendered text itself advertised, never
4871 // a hard-coded filename) -- "advertised == servable" is the entire
4872 // reason `HlsOrigin` exists, per the issue. `EXT-X-VERSION` is checked
4873 // against `broadcast_hls::MediaPlaylist::computed_version()` re-derived
4874 // from a round-trip parse of the rendered text -- never an independently
4875 // guessed integer literal, which is exactly the bug issue #871 removed.
4876
4877 fn make_origin_with(
4878 container: Container,
4879 low_latency_ms: Option<u32>,
4880 ) -> (Arc<Trunk>, HlsOrigin, media_plane::trunk::SegmentWriter) {
4881 let trunk = Trunk::new(TrunkConfig::new(nz(64), nz(8), nz(8), nz(8), nz(64)));
4882 let writer = trunk.segment_writer().expect("first segment writer");
4883 let mut builder = HlsOrigin::builder(Arc::clone(&trunk))
4884 .target_duration_secs(4.0)
4885 .window_segments(nz(4))
4886 .container(container);
4887 if let Some(ms) = low_latency_ms {
4888 builder = builder.low_latency(ms);
4889 }
4890 let origin = builder
4891 .instance(7)
4892 .build()
4893 .expect("both required fields set");
4894 (trunk, origin, writer)
4895 }
4896
4897 fn render_body(origin: &HlsOrigin) -> String {
4898 match resolve_now(
4899 origin,
4900 HlsRequest::Playlist {
4901 track_id: DEFAULT_TRACK_ID,
4902 query: BlockingQuery::default(),
4903 },
4904 ) {
4905 EgressResponse::Ready {
4906 body: HlsBody::Playlist(m),
4907 ..
4908 } => m,
4909 other => panic!("expected Ready(Playlist), got {other:?}"),
4910 }
4911 }
4912
4913 /// The `#EXT-X-VERSION:<n>` value present in `body`, if any.
4914 fn extract_version_tag(body: &str) -> Option<u8> {
4915 body.lines()
4916 .find_map(|l| l.strip_prefix("#EXT-X-VERSION:")?.parse::<u8>().ok())
4917 }
4918
4919 /// Every segment URI (the line immediately after a `#EXTINF:` line), in
4920 /// playlist order -- exactly what a real client would fetch next.
4921 fn segment_uris(body: &str) -> Vec<String> {
4922 let lines: Vec<&str> = body.lines().collect();
4923 let mut out = Vec::new();
4924 for i in 0..lines.len() {
4925 if lines[i].starts_with("#EXTINF:")
4926 && let Some(next) = lines.get(i + 1)
4927 && !next.starts_with('#')
4928 {
4929 out.push((*next).to_string());
4930 }
4931 }
4932 out
4933 }
4934
4935 /// Every `#EXT-X-PART:` line's `URI="..."` value, in order.
4936 fn part_uris(body: &str) -> Vec<String> {
4937 body.lines()
4938 .filter(|l| l.starts_with("#EXT-X-PART:"))
4939 .filter_map(|l| {
4940 let start = l.find("URI=\"")? + "URI=\"".len();
4941 let rest = &l[start..];
4942 let end = rest.find('"')?;
4943 Some(rest[..end].to_string())
4944 })
4945 .collect()
4946 }
4947
4948 /// Assert the rendered `#EXT-X-VERSION` equals `broadcast_hls`'s own
4949 /// `computed_version()`, re-derived by round-trip parsing the rendered
4950 /// text -- never an integer this test independently guessed. Returns the
4951 /// rendered value so a caller can make a further, non-vacuous claim
4952 /// about it.
4953 fn assert_version_matches_broadcast_hls_derivation(body: &str) -> Option<u8> {
4954 let parsed = MediaPlaylist::parse(body).expect("rendered body must round-trip parse");
4955 let rendered = extract_version_tag(body);
4956 assert_eq!(
4957 rendered,
4958 parsed.computed_version(),
4959 "rendered #EXT-X-VERSION must equal broadcast_hls's own derivation, body: {body}"
4960 );
4961 rendered
4962 }
4963
4964 /// [`assert_version_matches_broadcast_hls_derivation`] **plus a
4965 /// non-vacuity guard**: the playlist must actually trigger at least one
4966 /// RFC 8216bis §8 version rule, so the equality above cannot pass by
4967 /// comparing `None` against `None`.
4968 ///
4969 /// Without this, a cell whose every `#EXTINF` is integral (e.g. a
4970 /// duration of exactly `4.0`, which renders `#EXTINF:4,`) trips no §8
4971 /// row at all, and the derivation could be entirely broken while the
4972 /// test stayed green. Real segmenters cut on keyframes, not whole
4973 /// seconds, so a fractional `#EXTINF` is also the realistic shape --
4974 /// cf. `fixtures/hls/spec/9.1-simple-media-playlist.m3u8` (`#EXTINF:9.009`,
4975 /// `#EXT-X-VERSION:3`).
4976 fn assert_version_present_and_matches_derivation(body: &str) -> u8 {
4977 assert_version_matches_broadcast_hls_derivation(body).unwrap_or_else(|| {
4978 panic!(
4979 "this cell must trigger a real RFC 8216bis §8 version rule -- a \
4980 missing #EXT-X-VERSION makes the derivation check vacuous, body: {body}"
4981 )
4982 })
4983 }
4984
4985 /// MUTATION VERIFIED (issue #873): making `HlsOriginBuilder::container`
4986 /// a no-op (so every origin renders as `Fmp4` regardless of the
4987 /// `Container` passed to it) makes this test's
4988 /// `assert!(!body.contains("#EXT-X-MAP"))` fail -- the mutated build
4989 /// unconditionally emits `#EXT-X-MAP:URI="init-1.mp4"`, and the `.ts`
4990 /// URI assertions fail too (segments render as `seg-1-7-1.m4s` instead of
4991 /// `seg-1-7-1.ts`). Recompiled and re-run to confirm the failure (see the
4992 /// PR description for the pasted `cargo test` output), then reverted.
4993 /// The mutation bites four tests in total -- this one, the low-latency
4994 /// `MpegTs` cell, the integral-`EXTINF` version case, and the
4995 /// cross-container refusal.
4996 #[test]
4997 fn mpegts_classic_no_map_ts_uris_no_ll_tags() {
4998 let (_trunk, origin, writer) = make_origin_with(Container::MpegTs, None);
4999 // Fractional, like every real keyframe-cut segment (and like the
5000 // RFC's own classic examples) -- so RFC 8216bis §8 row 3
5001 // (floating-point EXTINF) genuinely fires and the version assertion
5002 // below is not `None == None`.
5003 seg(&writer, 1, 4.004, false);
5004
5005 let body = render_body(&origin);
5006 assert!(!body.contains("#EXT-X-MAP"), "body: {body}");
5007 assert!(!body.contains("#EXT-X-PART"), "body: {body}");
5008 assert!(!body.contains("#EXT-X-SERVER-CONTROL"), "body: {body}");
5009 assert!(!body.contains("#EXT-X-PRELOAD-HINT"), "body: {body}");
5010 assert!(!body.contains(".m4s"), "body: {body}");
5011 assert!(body.contains("seg-1-7-1.ts"), "body: {body}");
5012 assert_version_present_and_matches_derivation(&body);
5013
5014 // advertised == servable: fetch every URI the rendered text itself
5015 // named, and check its bytes against what was actually published
5016 // (via this crate's own `parse_immediate`, not a hard-coded filename).
5017 let uris = segment_uris(&body);
5018 assert_eq!(uris, vec!["seg-1-7-1.ts".to_string()]);
5019 for uri in uris {
5020 let ImmediateResource::Segment { seq, .. } = parse_immediate(&uri, Container::MpegTs)
5021 .expect("advertised segment URI must parse under MpegTs")
5022 else {
5023 panic!("expected a Segment resource for {uri}");
5024 };
5025 match resolve_now(&origin, HlsRequest::Resource { name: uri.clone() }) {
5026 EgressResponse::Ready {
5027 body: HlsBody::Resource(bytes),
5028 ..
5029 } => assert_eq!(bytes, Bytes::from(vec![seq as u8; 8])),
5030 other => panic!("expected Ready for {uri}, got {other:?}"),
5031 }
5032 }
5033 }
5034
5035 /// MUTATION VERIFIED (issue #873): making `HlsOriginBuilder::container`
5036 /// a no-op makes this test's `assert!(!body.contains("#EXT-X-MAP"))`
5037 /// fail identically to the classic-`MpegTs` test above, and the part
5038 /// URIs render as `part-1-7-2.0.m4s` instead of `part-1-7-2.0.ts`, so
5039 /// `assert!(body.contains("part-1-7-2.0.ts"))` also fails. Recompiled and
5040 /// re-run to confirm, then reverted.
5041 #[test]
5042 fn mpegts_low_latency_part_ts_uris_blocking_part_requests_resolve() {
5043 let (trunk, origin, writer) = make_origin_with(Container::MpegTs, Some(500));
5044 let origin = Arc::new(origin);
5045 // A closed segment with a fractional (keyframe-cut) duration, so
5046 // RFC 8216bis §8 row 3 fires and the version assertion below is not
5047 // `None == None`; segment 2 is then the open one carrying live parts.
5048 seg(&writer, 1, 4.004, false);
5049 part(&writer, 2, 0, true);
5050 part(&writer, 2, 1, false);
5051
5052 let body = render_body(&origin);
5053 assert!(!body.contains("#EXT-X-MAP"), "body: {body}");
5054 assert!(body.contains("#EXT-X-PART-INF"), "body: {body}");
5055 assert!(body.contains("#EXT-X-PART:"), "body: {body}");
5056 assert!(body.contains("seg-1-7-1.ts"), "body: {body}");
5057 assert!(body.contains("part-1-7-2.0.ts"), "body: {body}");
5058 assert!(!body.contains(".m4s"), "body: {body}");
5059 // LL-HLS directives add no §8 version requirement of their own --
5060 // the finding that killed the old hardcoded `EXT-X-VERSION:9` --
5061 // so this must derive to exactly the same value as the classic
5062 // MpegTs cell above.
5063 assert_version_present_and_matches_derivation(&body);
5064
5065 // advertised == servable for every advertised part.
5066 let parts = part_uris(&body);
5067 assert!(!parts.is_empty(), "body: {body}");
5068 for uri in &parts {
5069 let (_epoch, seq, idx) = parse_part(uri, Container::MpegTs)
5070 .unwrap_or_else(|| panic!("advertised part URI {uri} must parse under MpegTs"));
5071 match resolve_now(&origin, HlsRequest::Resource { name: uri.clone() }) {
5072 EgressResponse::Ready {
5073 body: HlsBody::Resource(bytes),
5074 ..
5075 } => {
5076 assert_eq!(bytes, Bytes::from(vec![idx.value() as u8; 4]));
5077 assert_eq!(seq, 2);
5078 }
5079 other => panic!("expected Ready for {uri}, got {other:?}"),
5080 }
5081 }
5082
5083 // A blocking request for a `.ts` part not yet produced must Await,
5084 // then resolve once produced -- the same guarantee the pre-#873
5085 // fMP4-only test proves, now exercised over the `.ts` URI scheme.
5086 let deadline = Timestamp::from_nanos(5_000_000_000);
5087 let policy = AwaitPolicy::new(deadline);
5088 let pending = origin.resolve(
5089 HlsRequest::Resource {
5090 name: "part-1-7-2.2.ts".to_string(),
5091 },
5092 Timestamp::from_nanos(0),
5093 policy,
5094 );
5095 assert!(
5096 matches!(pending, EgressResponse::Await { .. }),
5097 "expected Await before the part exists, got {pending:?}"
5098 );
5099
5100 let listener = trunk.listen().expect("listener slot available");
5101 let woken = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false));
5102 let woken2 = std::sync::Arc::clone(&woken);
5103 let waiter = std::thread::spawn(move || {
5104 let ok = listener.wait_deadline(Instant::now() + Duration::from_secs(60));
5105 woken2.store(ok, std::sync::atomic::Ordering::SeqCst);
5106 });
5107 part(&writer, 2, 2, false);
5108 waiter.join().expect("waiter thread must not panic");
5109 assert!(
5110 woken.load(std::sync::atomic::Ordering::SeqCst),
5111 "Trunk::listen() must wake once publish_part lands"
5112 );
5113
5114 match origin.resolve(
5115 HlsRequest::Resource {
5116 name: "part-1-7-2.2.ts".to_string(),
5117 },
5118 Timestamp::from_nanos(1),
5119 policy,
5120 ) {
5121 EgressResponse::Ready {
5122 body: HlsBody::Resource(bytes),
5123 ..
5124 } => assert_eq!(bytes, Bytes::from(vec![2u8; 4])),
5125 other => panic!("expected Ready once produced, got {other:?}"),
5126 }
5127 }
5128
5129 /// RFC 8216bis §8's opening rule: a playlist that triggers no version
5130 /// row at all is version-1 compatible and need not carry the tag. Kept
5131 /// as its own case (rather than as the `MpegTs` cells' only version
5132 /// check, which made those assertions vacuous) because an integral
5133 /// `#EXTINF` is a real, if unusual, shape.
5134 ///
5135 /// MUTATION VERIFIED (issue #873): making `HlsOriginBuilder::container`
5136 /// a no-op makes this fail with `left: Some(6), right: None` -- the
5137 /// mutated build emits `#EXT-X-MAP`, which trips §8 row 6. Recompiled
5138 /// and re-run to confirm, then reverted.
5139 #[test]
5140 fn mpegts_classic_integral_extinf_emits_no_version_tag() {
5141 let (_trunk, origin, writer) = make_origin_with(Container::MpegTs, None);
5142 seg(&writer, 1, 4.0, false);
5143 let body = render_body(&origin);
5144 // A whole number of seconds renders as an integer, so the playlist
5145 // genuinely contains no floating-point EXTINF value -- §8 row 3
5146 // does not fire, and omitting `EXT-X-VERSION` is honest rather than
5147 // a lie to a v1/v2 client. (`broadcast-hls` used to render `4.000`
5148 // here while still reporting no version requirement; fixed in the
5149 // same PR as this test.)
5150 assert!(body.contains("#EXTINF:4,"), "body: {body}");
5151 assert!(!body.contains("4.000"), "body: {body}");
5152 assert_eq!(
5153 assert_version_matches_broadcast_hls_derivation(&body),
5154 None,
5155 "nothing in this playlist triggers an RFC 8216bis §8 row: {body}"
5156 );
5157 }
5158
5159 /// LL-HLS directives carry no RFC 8216bis §8 version requirement of
5160 /// their own -- the finding that killed the old hardcoded
5161 /// `EXT-X-VERSION:9`. Enabling low latency must therefore not change
5162 /// the derived version for otherwise-identical content.
5163 #[test]
5164 fn low_latency_does_not_raise_the_derived_version() {
5165 let (_t1, classic, w1) = make_origin_with(Container::MpegTs, None);
5166 seg(&w1, 1, 4.004, false);
5167 let classic_version = assert_version_present_and_matches_derivation(&render_body(&classic));
5168
5169 let (_t2, low_latency, w2) = make_origin_with(Container::MpegTs, Some(500));
5170 seg(&w2, 1, 4.004, false);
5171 part(&w2, 2, 0, true);
5172 let ll_body = render_body(&low_latency);
5173 assert!(ll_body.contains("#EXT-X-PART:"), "body: {ll_body}");
5174 assert_eq!(
5175 assert_version_present_and_matches_derivation(&ll_body),
5176 classic_version,
5177 "enabling low latency must not raise the derived version"
5178 );
5179 }
5180
5181 #[test]
5182 fn fmp4_classic_map_present_no_ll_tags() {
5183 let (_trunk, origin, writer) = make_origin_with(Container::Fmp4, None);
5184 origin.set_init(vec![0xBBu8; 8]);
5185 // Fractional, so §8 row 3 fires alongside row 6 (EXT-X-MAP without
5186 // EXT-X-I-FRAMES-ONLY). `max(6, 3) = 6`, so the rendered value is
5187 // unchanged -- this removes the ambiguity of an integral EXTINF
5188 // leaving row 3 entirely untested here.
5189 seg(&writer, 1, 4.004, false);
5190
5191 let body = render_body(&origin);
5192 assert!(
5193 body.contains("#EXT-X-MAP:URI=\"init-1-7-1.mp4\""),
5194 "body: {body}"
5195 );
5196 assert!(!body.contains("#EXT-X-PART"), "body: {body}");
5197 assert!(!body.contains("#EXT-X-SERVER-CONTROL"), "body: {body}");
5198 assert!(!body.contains("#EXT-X-PRELOAD-HINT"), "body: {body}");
5199 assert!(body.contains("seg-1-7-1.m4s"), "body: {body}");
5200 assert_version_present_and_matches_derivation(&body);
5201
5202 // advertised == servable, including the init segment the MAP names.
5203 match resolve_now(
5204 &origin,
5205 HlsRequest::Resource {
5206 name: "init-1.mp4".to_string(),
5207 },
5208 ) {
5209 EgressResponse::Ready {
5210 body: HlsBody::Resource(bytes),
5211 ..
5212 } => assert_eq!(bytes, Bytes::from(vec![0xBBu8; 8])),
5213 other => panic!("expected Ready(init), got {other:?}"),
5214 }
5215 for uri in segment_uris(&body) {
5216 let ImmediateResource::Segment { seq, .. } = parse_immediate(&uri, Container::Fmp4)
5217 .expect("advertised segment URI must parse under Fmp4")
5218 else {
5219 panic!("expected a Segment resource for {uri}");
5220 };
5221 match resolve_now(&origin, HlsRequest::Resource { name: uri.clone() }) {
5222 EgressResponse::Ready {
5223 body: HlsBody::Resource(bytes),
5224 ..
5225 } => assert_eq!(bytes, Bytes::from(vec![seq as u8; 8])),
5226 other => panic!("expected Ready for {uri}, got {other:?}"),
5227 }
5228 }
5229 }
5230
5231 /// Regression guard (issue #873, matrix cell 4): the pre-#873 default
5232 /// shape (`Fmp4` + low-latency) must render unchanged.
5233 #[test]
5234 fn fmp4_low_latency_existing_behaviour_preserved() {
5235 let (_trunk, origin, writer) = make_origin_with(Container::Fmp4, Some(500));
5236 origin.set_init(vec![0xAAu8; 8]);
5237 // Fractional for the same reason as the Fmp4-classic cell above.
5238 seg(&writer, 1, 4.004, false);
5239 part(&writer, 2, 0, true);
5240 part(&writer, 2, 1, false);
5241
5242 let body = render_body(&origin);
5243 assert!(
5244 body.contains("#EXT-X-MAP:URI=\"init-1-7-1.mp4\""),
5245 "body: {body}"
5246 );
5247 assert!(body.contains("#EXT-X-PART-INF"), "body: {body}");
5248 assert!(body.contains("#EXT-X-SERVER-CONTROL"), "body: {body}");
5249 assert!(body.contains("#EXT-X-PRELOAD-HINT"), "body: {body}");
5250 assert!(body.contains("seg-1-7-1.m4s"), "body: {body}");
5251 assert!(body.contains("part-1-7-2.0.m4s"), "body: {body}");
5252 assert!(!body.contains(".ts\""), "body: {body}");
5253 assert_version_present_and_matches_derivation(&body);
5254
5255 match resolve_now(
5256 &origin,
5257 HlsRequest::Resource {
5258 name: "init-1.mp4".to_string(),
5259 },
5260 ) {
5261 EgressResponse::Ready {
5262 body: HlsBody::Resource(bytes),
5263 ..
5264 } => assert_eq!(bytes, Bytes::from(vec![0xAAu8; 8])),
5265 other => panic!("expected Ready(init), got {other:?}"),
5266 }
5267 for uri in segment_uris(&body) {
5268 let ImmediateResource::Segment { seq, .. } = parse_immediate(&uri, Container::Fmp4)
5269 .expect("advertised segment URI must parse under Fmp4")
5270 else {
5271 panic!("expected a Segment resource for {uri}");
5272 };
5273 match resolve_now(&origin, HlsRequest::Resource { name: uri.clone() }) {
5274 EgressResponse::Ready {
5275 body: HlsBody::Resource(bytes),
5276 ..
5277 } => assert_eq!(bytes, Bytes::from(vec![seq as u8; 8])),
5278 other => panic!("expected Ready for {uri}, got {other:?}"),
5279 }
5280 }
5281 for uri in part_uris(&body) {
5282 let (_epoch, _seq, idx) = parse_part(&uri, Container::Fmp4)
5283 .unwrap_or_else(|| panic!("advertised part URI {uri} must parse under Fmp4"));
5284 match resolve_now(&origin, HlsRequest::Resource { name: uri.clone() }) {
5285 EgressResponse::Ready {
5286 body: HlsBody::Resource(bytes),
5287 ..
5288 } => assert_eq!(bytes, Bytes::from(vec![idx.value() as u8; 4])),
5289 other => panic!("expected Ready for {uri}, got {other:?}"),
5290 }
5291 }
5292 }
5293
5294 /// MUTATION VERIFIED (issue #873): making `HlsOriginBuilder::container`
5295 /// a no-op makes every origin behave as `Fmp4`, so `init-1.mp4` under a
5296 /// nominally-`MpegTs` origin resolves `Ready` instead of `NotFound` --
5297 /// this test's final assertion fails. Recompiled and re-run to confirm,
5298 /// then reverted.
5299 #[test]
5300 fn cross_container_refusal_mp4_init_under_mpegts_not_found() {
5301 let (_trunk, origin, _writer) = make_origin_with(Container::MpegTs, None);
5302 // Still callable (documented no-op) -- the bytes are simply never
5303 // advertised or served under `MpegTs`.
5304 origin.set_init(vec![0xCCu8; 8]);
5305 assert_eq!(
5306 resolve_now(
5307 &origin,
5308 HlsRequest::Resource {
5309 name: "init-1.mp4".to_string(),
5310 }
5311 ),
5312 EgressResponse::NotFound
5313 );
5314 }
5315
5316 #[test]
5317 fn builder_errors_on_missing_required_fields() {
5318 let trunk = Trunk::new(TrunkConfig::new(nz(64), nz(8), nz(8), nz(8), nz(64)));
5319 match HlsOrigin::builder(Arc::clone(&trunk))
5320 .window_segments(nz(4))
5321 .instance(7)
5322 .build()
5323 {
5324 Err(e) => assert_eq!(e, HlsOriginBuildError::MissingTargetDurationSecs),
5325 Ok(_) => panic!("expected an error: target_duration_secs was never set"),
5326 }
5327 match HlsOrigin::builder(Arc::clone(&trunk))
5328 .target_duration_secs(4.0)
5329 .instance(7)
5330 .build()
5331 {
5332 Err(e) => assert_eq!(e, HlsOriginBuildError::MissingWindowSegments),
5333 Ok(_) => panic!("expected an error: window_segments was never set"),
5334 }
5335 }
5336
5337 #[test]
5338 fn container_label_and_display() {
5339 assert_eq!(Container::Fmp4.name(), "fmp4");
5340 assert_eq!(Container::MpegTs.name(), "mpeg-ts");
5341 assert_eq!(Container::Fmp4.to_string(), "fmp4");
5342 assert_eq!(Container::default(), Container::Fmp4);
5343 }
5344
5345 // --- issue #900: `closed_segments()` — the snapshot multimux's DVR
5346 // catch-up serving merges with its on-disk archive ---
5347
5348 /// MUTATION VERIFIED: changing `Window::push`'s
5349 /// `start_ns: entry.timeline_position.as_nanos()` to a constant `0`
5350 /// makes this test's `assert_eq!(snapshot[1].start_ns, 4_000_000_000)`
5351 /// fail (`left: 0, right: 4000000000`) — `closed_segments()` would then
5352 /// report every segment as starting at time zero, which is exactly the
5353 /// bug that would make a caller's time-based catch-up window (issue
5354 /// #900) unable to tell segments apart. Recompiled and re-run to
5355 /// confirm the failure, then reverted.
5356 #[test]
5357 fn closed_segments_snapshot_matches_published_segments_ascending() {
5358 let (_trunk, origin, writer) = make_origin();
5359 writer
5360 .publish_segment(SegmentEntry::new(
5361 Bytes::from(vec![1u8; 8]),
5362 SegmentSeq::new(1),
5363 Duration::from_secs_f64(4.0),
5364 Timestamp::from_nanos(0),
5365 SegmentMeta {
5366 discontinuous: false,
5367 },
5368 ))
5369 .unwrap();
5370 writer
5371 .publish_segment(SegmentEntry::new(
5372 Bytes::from(vec![2u8; 8]),
5373 SegmentSeq::new(2),
5374 Duration::from_secs_f64(4.0),
5375 Timestamp::from_nanos(4_000_000_000),
5376 SegmentMeta {
5377 discontinuous: true,
5378 },
5379 ))
5380 .unwrap();
5381
5382 let snapshot = origin.closed_segments();
5383 assert_eq!(snapshot.len(), 2);
5384 assert_eq!(snapshot[0].sequence_number, SegmentSeq::new(1));
5385 assert_eq!(snapshot[0].start_ns, 0);
5386 assert!(!snapshot[0].discontinuous);
5387 assert_eq!(snapshot[1].sequence_number, SegmentSeq::new(2));
5388 assert_eq!(snapshot[1].start_ns, 4_000_000_000);
5389 assert!(snapshot[1].discontinuous);
5390 }
5391
5392 /// A segment evicted from the window (beyond `window_segments`
5393 /// capacity) no longer appears in the snapshot — `closed_segments()`
5394 /// reports the *advertised* window, the same one `render_playlist`
5395 /// renders, not every segment ever published.
5396 #[test]
5397 fn closed_segments_reflects_window_eviction() {
5398 let (_trunk, origin, writer) = make_origin(); // window_segments = 4
5399 for seq in 1..=5 {
5400 seg(&writer, seq, 4.0, false);
5401 }
5402 let snapshot = origin.closed_segments();
5403 let seqs: Vec<u32> = snapshot.iter().map(|s| s.sequence_number.value()).collect();
5404 assert_eq!(
5405 seqs,
5406 vec![2, 3, 4, 5],
5407 "oldest segment 1 must have rolled off"
5408 );
5409 }
5410
5411 fn window_entry(seq: u32, tag: u8, discontinuous: bool) -> SegmentEntry {
5412 SegmentEntry::new(
5413 Bytes::from(vec![tag; 4]),
5414 SegmentSeq::new(seq),
5415 Duration::from_secs(4),
5416 Timestamp::from_nanos(0),
5417 SegmentMeta { discontinuous },
5418 )
5419 }
5420
5421 fn window_seqs(w: &Window) -> Vec<(u32, bool)> {
5422 w.segments
5423 .iter()
5424 .map(|s| (s.sequence_number.value(), s.discontinuous))
5425 .collect()
5426 }
5427
5428 // r09-W11: the Trunk's `publish_segment` refuses a decreasing number, but
5429 // the window must not depend on that: a repeat or decrease clears it
5430 // instead of holding two entries under one number (the old `bytes_of`
5431 // served the stale one).
5432 #[test]
5433 fn window_restarts_on_a_repeated_or_decreasing_sequence_number() {
5434 let mut w = Window::new(nz(6));
5435 w.push(window_entry(5, 0xA5, false), Generation::new(1), None, None);
5436 w.push(window_entry(6, 0xA6, true), Generation::new(1), None, None);
5437 assert_eq!(window_seqs(&w), vec![(5, false), (6, true)]);
5438
5439 // Decrease: 3 after 6.
5440 w.push(window_entry(3, 0xB3, false), Generation::new(1), None, None);
5441 assert_eq!(window_seqs(&w), vec![(3, true)]);
5442 // The discontinuous 6 rolled off the front.
5443 assert_eq!(w.discontinuity_sequence, 1);
5444 assert_eq!(
5445 w.bytes_of(SegmentSeq::new(3)),
5446 Some(Bytes::from(vec![0xB3; 4]))
5447 );
5448 assert_eq!(w.bytes_of(SegmentSeq::new(5)), None);
5449 assert_eq!(w.bytes_of(SegmentSeq::new(6)), None);
5450
5451 // Repeat: 3 again with different bytes replaces, never duplicates.
5452 w.push(window_entry(3, 0xC3, false), Generation::new(1), None, None);
5453 assert_eq!(window_seqs(&w), vec![(3, true)]);
5454 assert_eq!(
5455 w.bytes_of(SegmentSeq::new(3)),
5456 Some(Bytes::from(vec![0xC3; 4]))
5457 );
5458 }
5459
5460 // Hostile edge: the largest sequence number has no successor; a repeat of
5461 // it must restart the window, not overflow `last + 1`.
5462 #[test]
5463 fn window_handles_u32_max_sequence_numbers() {
5464 let mut w = Window::new(nz(4));
5465 w.push(
5466 window_entry(u32::MAX - 1, 1, false),
5467 Generation::new(1),
5468 None,
5469 None,
5470 );
5471 w.push(
5472 window_entry(u32::MAX, 2, false),
5473 Generation::new(1),
5474 None,
5475 None,
5476 );
5477 assert_eq!(
5478 window_seqs(&w),
5479 vec![(u32::MAX - 1, false), (u32::MAX, false)]
5480 );
5481 w.push(
5482 window_entry(u32::MAX, 3, false),
5483 Generation::new(1),
5484 None,
5485 None,
5486 );
5487 assert_eq!(window_seqs(&w), vec![(u32::MAX, true)]);
5488 assert_eq!(
5489 w.bytes_of(SegmentSeq::new(u32::MAX)),
5490 Some(Bytes::from(vec![3; 4]))
5491 );
5492 }
5493
5494 #[test]
5495 fn window_capacity_is_never_below_three_segments() {
5496 let mut w = Window::new(nz(1));
5497 for seq in 1..=5 {
5498 w.push(window_entry(seq, 0, false), Generation::new(1), None, None);
5499 }
5500 assert_eq!(window_seqs(&w), vec![(3, false), (4, false), (5, false)]);
5501 }
5502
5503 /// A measured peak bitrate above `u32::MAX` must surface as the typed
5504 /// [`HlsMasterError::BandwidthOverflow`] (was an untyped `String`).
5505 #[test]
5506 fn master_playlist_reports_typed_bandwidth_overflow() {
5507 let (_trunk, origin, writer) = make_origin();
5508 // 8 bytes (64 bits) in 10 ns = 6.4e9 b/s > u32::MAX.
5509 writer
5510 .publish_segment(SegmentEntry::new(
5511 Bytes::from(vec![1u8; 8]),
5512 SegmentSeq::new(1),
5513 Duration::from_nanos(10),
5514 Timestamp::from_nanos(0),
5515 SegmentMeta {
5516 discontinuous: false,
5517 },
5518 ))
5519 .unwrap();
5520 assert_eq!(
5521 origin.master_playlist("media.m3u8"),
5522 Err(HlsMasterError::BandwidthOverflow {
5523 bandwidth: 6_400_000_000
5524 })
5525 );
5526 }
5527
5528 #[test]
5529 fn segment_bandwidth_is_rounded_up_and_rejects_degenerate_input() {
5530 // 1001 B * 8 / 3 s = 2669.33 -> 2670.
5531 assert_eq!(
5532 segment_bandwidth_bps(1001, Duration::from_secs(3)),
5533 Some(2670)
5534 );
5535 // Exact division stays exact.
5536 assert_eq!(
5537 segment_bandwidth_bps(1_500_000, Duration::from_secs(1)),
5538 Some(12_000_000)
5539 );
5540 // Zero duration: nothing to measure.
5541 assert_eq!(segment_bandwidth_bps(10, Duration::ZERO), None);
5542 // A one-nanosecond segment of huge size overflows u64 b/s.
5543 assert_eq!(
5544 segment_bandwidth_bps(usize::MAX, Duration::from_nanos(1)),
5545 None
5546 );
5547 }
5548
5549 #[test]
5550 fn bytes_of_addresses_exactly_the_matching_entry() {
5551 let mut w = Window::new(nz(6));
5552 for (seq, tag) in [(5u32, 0x15u8), (6, 0x16), (7, 0x17)] {
5553 w.push(
5554 window_entry(seq, tag, false),
5555 Generation::new(1),
5556 None,
5557 None,
5558 );
5559 }
5560 assert_eq!(
5561 w.bytes_of(SegmentSeq::new(5)),
5562 Some(Bytes::from(vec![0x15; 4]))
5563 );
5564 assert_eq!(
5565 w.bytes_of(SegmentSeq::new(6)),
5566 Some(Bytes::from(vec![0x16; 4]))
5567 );
5568 assert_eq!(
5569 w.bytes_of(SegmentSeq::new(7)),
5570 Some(Bytes::from(vec![0x17; 4]))
5571 );
5572 assert_eq!(w.bytes_of(SegmentSeq::new(4)), None);
5573 assert_eq!(w.bytes_of(SegmentSeq::new(8)), None);
5574 assert_eq!(w.bytes_of(SegmentSeq::new(u32::MAX)), None);
5575 }
5576
5577 // --- r09-O4 / T6 (#1089, #1134): lock discipline ---------------------
5578
5579 fn segment_request(name: &str) -> HlsRequest {
5580 HlsRequest::Resource {
5581 name: name.to_string(),
5582 }
5583 }
5584
5585 /// With nothing published since the last drain, a request must not need
5586 /// the cursor lock. Holding that lock from the test thread makes the old
5587 /// always-drain code block (the request thread then misses the timeout).
5588 #[test]
5589 fn a_request_with_nothing_new_does_not_take_the_cursor_lock() {
5590 let (_trunk, origin, writer) = make_origin();
5591 seg(&writer, 1, 4.0, false);
5592 assert!(matches!(
5593 resolve_now(&origin, segment_request("seg-1-7-1.m4s")),
5594 EgressResponse::Ready { .. }
5595 ));
5596
5597 let guard = locked(&origin.cursor);
5598 let (tx, rx) = std::sync::mpsc::channel();
5599 std::thread::scope(|s| {
5600 s.spawn(|| {
5601 let _ = tx.send(resolve_now(&origin, segment_request("seg-1-7-1.m4s")));
5602 });
5603 let got = rx.recv_timeout(Duration::from_secs(10));
5604 drop(guard); // always release, so the scope can join
5605 assert!(
5606 matches!(got, Ok(EgressResponse::Ready { .. })),
5607 "request blocked on the cursor lock: {got:?}"
5608 );
5609 });
5610 }
5611
5612 /// The skip never hides a real publish: a segment closed after the last
5613 /// drain is served by the very next request.
5614 #[test]
5615 fn a_segment_published_after_a_drain_is_served_by_the_next_request() {
5616 let (_trunk, origin, writer) = make_origin();
5617 seg(&writer, 1, 4.0, false);
5618 assert!(matches!(
5619 resolve_now(&origin, segment_request("seg-1-7-1.m4s")),
5620 EgressResponse::Ready { .. }
5621 ));
5622 assert_eq!(
5623 resolve_now(&origin, segment_request("seg-1-7-2.m4s")),
5624 EgressResponse::NotFound
5625 );
5626 seg(&writer, 2, 4.0, false);
5627 assert!(matches!(
5628 resolve_now(&origin, segment_request("seg-1-7-2.m4s")),
5629 EgressResponse::Ready { .. }
5630 ));
5631 }
5632
5633 /// A panic while some thread held an origin lock must not become a panic
5634 /// in every later request.
5635 #[test]
5636 fn a_poisoned_lock_does_not_cascade_into_later_requests() {
5637 let (_trunk, origin, writer) = make_origin();
5638 seg(&writer, 1, 4.0, false);
5639 std::thread::scope(|s| {
5640 let origin = &origin;
5641 for which in 0..4 {
5642 let handle = s.spawn(move || {
5643 // Hold one of the origin's mutexes across a panic.
5644 match which {
5645 0 => drop(origin.window.lock().map(|_g| panic!("poison window"))),
5646 1 => drop(origin.cursor.lock().map(|_g| panic!("poison cursor"))),
5647 2 => drop(origin.init.lock().map(|_g| panic!("poison init"))),
5648 _ => drop(origin.codecs.lock().map(|_g| panic!("poison codecs"))),
5649 }
5650 });
5651 assert!(handle.join().is_err(), "setup thread {which} must panic");
5652 }
5653 });
5654 assert!(origin.window.is_poisoned());
5655 assert!(origin.cursor.is_poisoned());
5656 assert!(origin.init.is_poisoned());
5657 assert!(origin.codecs.is_poisoned());
5658 // Every public entry point still answers.
5659 seg(&writer, 2, 4.0, false);
5660 assert_eq!(origin.closed_segments().len(), 2);
5661 assert!(origin.init_bytes().is_some());
5662 assert!(origin.master_playlist("media.m3u8").is_ok());
5663 origin.set_init(vec![0xCC; 8]);
5664 assert!(render_body(&origin).contains("seg-1-7-2.m4s"));
5665 assert!(matches!(
5666 resolve_now(&origin, segment_request("seg-1-7-1.m4s")),
5667 EgressResponse::Ready { .. }
5668 ));
5669 }
5670}