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