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