Skip to main content

hls_runtime/server/
engine.rs

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