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