Skip to main content

hls_runtime/server/
engine.rs

1//! [`HlsOrigin`] — the LL-HLS origin [`ServedEgress`] (plan step 4):
2//! blocking-reload/part-availability *decision* logic and playlist rendering,
3//! rendered directly from a shared [`Trunk`] instead of the deleted
4//! `MediaStore` push-fed rolling window.
5//!
6//! Master/media playlist tags are RFC 8216 §4.3.4 (`#EXT-X-STREAM-INF`) and
7//! §4.3.3 (`#EXTM3U`/`#EXT-X-VERSION`, rendered by [`MediaPlaylist::to_m3u8`]);
8//! the blocking reload query parameters (`_HLS_msn`/`_HLS_part`) are the
9//! Blocking Playlist Reload mechanism of RFC 8216bis §6.2.5.2 — the client
10//! asks the origin to hold the response open until the requested Media
11//! Sequence Number/part is available, bounded by the caller's own
12//! [`AwaitPolicy`] so the origin never hangs indefinitely.
13//!
14//! # What comes straight from the `Trunk`, with no cache at all
15//!
16//! Every part-availability and blocking-reload decision reads the `Trunk`
17//! directly, `&self`-shaped, every call:
18//!
19//! - **Live parts of the open segment** — [`Trunk::part_bytes`]/
20//!   [`Trunk::parts_in_segment`] (step 3b-iv's live-part log). This is the
21//!   whole reason step 3b-iv exists: before it, nothing in `Trunk` could
22//!   answer "does part 3 of the segment currently being written exist",
23//!   which is exactly what forced `MediaStore` to keep its own
24//!   `live_parts`/`recent_parts` buffers in the first place.
25//! - **Whether a segment has closed** — [`Trunk::last_closed_segment`].
26//! - **The "in-progress-or-last-active segment" `MediaStore::latest_progress`
27//!   used to track as a push-fed field** — [`HlsOrigin::live_edge`] derives
28//!   it from the two queries above alone (`last_closed_segment() + 1`, probed
29//!   via `parts_in_segment`), needing no field of its own. See that method's
30//!   doc for the derivation and why it is exact, not a heuristic.
31//! - **A just-closed segment's final part still resolving** — falls out of
32//!   [`Trunk::part_bytes`] for free: [`media_plane::trunk::SegmentWriter::publish_segment`]
33//!   deliberately never touches the live-part log (see `trunk`'s own module
34//!   doc, "The live-part log"), so this crate no longer needs `MediaStore`'s
35//!   separate `recent_parts` buffer at all — that buffer existed *only* to
36//!   simulate exactly the guarantee the `Trunk` now gives natively.
37//!
38//! # The one thing that genuinely cannot come from the `Trunk` alone
39//!
40//! [`Trunk::subscribe_segments`] hands back a moving, single-consumer
41//! [`SegmentCursor`] — there is no snapshot query over the segment log the
42//! way [`Trunk::events_between`] gives the event log (see
43//! `media_plane::egress`'s own module doc, "`ServedEgress::resolve` does not
44//! take `&Trunk`", which anticipated exactly this). Rendering a Media
45//! Playlist needs the **window** of currently-advertised closed segments
46//! (their bytes, durations, and discontinuity bits), plus two numbers that
47//! must survive eviction from that window: the lifetime-max segment
48//! duration (RFC 8216bis §4.4.3.1's `TARGETDURATION` MUST) and the
49//! cumulative discontinuity count that has rolled off the front
50//! (`#EXT-X-DISCONTINUITY-SEQUENCE`, RFC 8216 §4.3.3.3). None of that is
51//! answerable by a fresh `&self` call on `Trunk` — it has to be assembled by
52//! draining a cursor over time.
53//!
54//! `Window` is that assembly, and it is **not** a second `MediaStore`: it
55//! holds only bytes/duration/discontinuity-bit for the segments currently in
56//! the advertised window, fed by exactly **one** [`SegmentCursor`] this
57//! `HlsOrigin` owns — precisely the shape `media_plane::egress`'s module
58//! doc prescribes ("a `ServedEgress` implementation... keeps its own
59//! resolvable window in sync by draining [cursors]... `resolve` only ever
60//! reads that already-synced state"). It carries none of `MediaStore`'s
61//! other fields (`health`, `track_specs`, `created_at`, `window_segments()`
62//! diagnostics) — those served `multimux`'s DASH/ll-DASH outputs, not
63//! LL-HLS rendering, and are out of this step's scope (Step 5's problem, if
64//! still needed once `multimux` is rewritten).
65//!
66//! The fMP4 **init segment** bytes are the other thing this module holds
67//! outside the `Trunk`: an init segment is neither a sample, a finished
68//! segment, an event, nor a live part — it is produced once by the
69//! segmenter and never changes, so it was never in scope for any of
70//! `Trunk`'s four rings. [`HlsOrigin::set_init`] is the (small, honest) side
71//! channel for it — not a duplicate of anything `Trunk` holds.
72
73use std::collections::VecDeque;
74use std::num::NonZeroUsize;
75use std::sync::{Arc, Mutex};
76
77use broadcast_common::Timestamp;
78use broadcast_hls::{LowLatencyConfig, MediaPlaylist, MediaSegment, OpenSegment, PartSpec};
79use bytes::Bytes;
80use media_plane::egress::{AwaitPolicy, CachePolicy, EgressResponse, ServedEgress};
81use media_plane::trunk::{PartEntry, SegmentCursor, SegmentCursorItem, SegmentEntry, Trunk};
82
83/// Track id for the single rendition served per stream (no multi-track/
84/// multi-rendition support yet).
85pub const DEFAULT_TRACK_ID: u32 = 1;
86
87/// Which container [`HlsOrigin`] serves segments/parts as — orthogonal to
88/// whether LL-HLS is enabled ([`HlsOriginBuilder::low_latency`]); issue #873.
89///
90/// RFC 8216bis §3.1.1 / §3.1.2 give the two containers different
91/// `#EXT-X-MAP` obligations:
92///
93/// - fMP4 (§3.1.2): "Each fMP4 Segment in a Media Playlist MUST have an
94///   `EXT-X-MAP` tag applied to it" — unconditional, so [`Container::Fmp4`]
95///   always emits one.
96/// - MPEG-2 TS (§3.1.1): "Each Transport Stream Segment MUST contain a PAT
97///   and a PMT, **or** have an `EXT-X-MAP` tag applied to it" — a
98///   disjunction, not a container restriction. `EXT-X-MAP` is legal for TS;
99///   it is not required when the segments carry their own PAT/PMT.
100///
101/// [`Container::MpegTs`] omits `#EXT-X-MAP` **by default**, on the
102/// assumption that segments come from a self-initialising source (e.g.
103/// `transmux`'s TS segmenter, which re-emits PAT+PMT at the head of every
104/// segment) — it does not *forbid* the tag; a future caller feeding
105/// pre-segmented TS without in-band PSI would need a way to opt back in,
106/// which is not implemented here (out of scope for issue #873; the current
107/// wiring never calls `set_init` from a `MpegTs`-configured pipeline, so the
108/// gap has no live caller yet).
109#[derive(Debug, Clone, Copy, PartialEq, Eq)]
110#[non_exhaustive]
111pub enum Container {
112    /// Fragmented MP4 / CMAF segments (`.m4s`), with an fMP4 init segment
113    /// (`.mp4`) referenced by an always-present `#EXT-X-MAP`.
114    Fmp4,
115    /// Whole MPEG-2 Transport Stream segments (`.ts`), self-initialising
116    /// (in-band PAT/PMT) — no `#EXT-X-MAP`, no init segment served.
117    MpegTs,
118}
119
120impl Container {
121    /// The spec token for this container — `"fmp4"` / `"mpeg-ts"`.
122    pub fn name(&self) -> &'static str {
123        match self {
124            Container::Fmp4 => "fmp4",
125            Container::MpegTs => "mpeg-ts",
126        }
127    }
128
129    /// The dynamic-filename extension (without the leading `.`) this
130    /// container's segments/parts are named with.
131    fn segment_extension(self) -> &'static str {
132        match self {
133            Container::Fmp4 => "m4s",
134            Container::MpegTs => "ts",
135        }
136    }
137}
138
139broadcast_common::impl_spec_display!(Container);
140
141impl Default for Container {
142    /// [`Container::Fmp4`] — preserves every pre-#873 caller's behaviour.
143    fn default() -> Self {
144        Container::Fmp4
145    }
146}
147
148/// Placeholder `BANDWIDTH` (bits/second) advertised in the master playlist's
149/// `#EXT-X-STREAM-INF` — actual encoded bitrate isn't measured, so a single
150/// fixed estimate is used for the single variant served.
151const PLACEHOLDER_BANDWIDTH_BPS: u64 = 5_000_000;
152
153/// RFC 8216bis §6.2.5.2 (SHOULD): a `_HLS_msn` greater than "the Media
154/// Sequence Number of the last Media Segment in the current Playlist plus
155/// two" should be rejected rather than always blocking to the caller's
156/// timeout — a legitimate client only ever asks for the segment/part right
157/// after the one it already has, so anything more than two segments beyond
158/// the current last closed segment is either a malfunctioning client or abuse.
159const ABUSE_MSN_FUTURE_BOUND: u64 = 2;
160
161/// RFC 8216bis / Apple LL-HLS §4.4.3.7: `#EXT-X-SERVER-CONTROL`'s
162/// `PART-HOLD-BACK` attribute MUST be at least 3x the part target duration
163/// (`#EXT-X-PART-INF`'s `PART-TARGET`).
164const PART_HOLD_BACK_MULTIPLIER: f64 = 3.0;
165
166/// A minimal single-variant master playlist pointing at `media_playlist_name`
167/// (the caller's configured media-playlist filename — e.g. multimux's
168/// `Config::playlist_name`, defaulting to `"media.m3u8"`) — the same
169/// regardless of any stream state (no multi-rendition support yet), so this
170/// takes no `Trunk`/origin argument.
171pub fn master_playlist_m3u8(media_playlist_name: &str) -> String {
172    format!(
173        "#EXTM3U\n#EXT-X-STREAM-INF:BANDWIDTH={PLACEHOLDER_BANDWIDTH_BPS}\n{media_playlist_name}\n"
174    )
175}
176
177/// Blocking playlist reload query parameters (RFC 8216bis §6.2.5.2) — the
178/// sans-IO counterpart of an adapter's own (likely serde-`Deserialize`)
179/// query-string type; the adapter maps its wire query params into this.
180#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)]
181pub struct BlockingQuery {
182    /// The Media Sequence Number the client already has, plus one — the
183    /// origin should not respond until a segment/part beyond this is ready.
184    pub hls_msn: Option<u64>,
185    /// The part index (within `hls_msn`) the client is waiting for.
186    pub hls_part: Option<u32>,
187}
188
189/// [`ServedEgress::Request`] for [`HlsOrigin`]: which wire resource is
190/// being asked for. A data-carrying dispatch ADT (matches this crate's
191/// `client::action::Action`/`ResourceId` convention) — see
192/// `tests/label_coverage.rs`'s SKIP list.
193#[derive(Debug, Clone, PartialEq, Eq)]
194#[non_exhaustive]
195pub enum HlsRequest {
196    /// `GET <media playlist>`, optionally carrying a blocking-reload query.
197    Playlist {
198        /// The track id to render the playlist for (a naming parameter only —
199        /// see [`DEFAULT_TRACK_ID`]).
200        track_id: u32,
201        /// The blocking-reload query parameters, if any.
202        query: BlockingQuery,
203    },
204    /// `GET` a dynamic origin resource by its wire filename (`init-{track}.mp4`,
205    /// `seg-{track}-{seq}.m4s`, `part-{track}-{seq}.{idx}.m4s`).
206    Resource {
207        /// The requested filename, exactly as it appeared in the request path.
208        name: String,
209    },
210}
211
212/// [`ServedEgress::Body`] for [`HlsOrigin`]: the resolved body, typed by
213/// which [`HlsRequest`] produced it. A data-carrying ADT — see
214/// `tests/label_coverage.rs`'s SKIP list.
215#[derive(Debug, Clone, PartialEq, Eq)]
216#[non_exhaustive]
217pub enum HlsBody {
218    /// A rendered Media Playlist (`#EXTM3U` text).
219    Playlist(String),
220    /// Resolved resource bytes (init/segment/part).
221    Resource(Bytes),
222}
223
224/// One playlist-window-resident **closed** segment's identity/bytes —
225/// `Window`'s per-entry shape. Deliberately narrower than the old
226/// `MediaStore`'s `SegmentInfo`-derived window entries: this crate only ever
227/// needs bytes + duration + the discontinuity bit to render a Media
228/// Playlist, so that is all this holds.
229struct WindowSegment {
230    sequence_number: u32,
231    bytes: Bytes,
232    duration_secs: f64,
233    discontinuous: bool,
234}
235
236/// The small per-[`HlsOrigin`] synced window this module's own doc
237/// ("The one thing that genuinely cannot come from the `Trunk` alone")
238/// explains the need for — fed by draining exactly one [`SegmentCursor`],
239/// never pushed into directly.
240struct Window {
241    segments: VecDeque<WindowSegment>,
242    capacity: usize,
243    /// Largest segment duration ever drained, surviving window eviction —
244    /// RFC 8216bis §4.4.3.1's `TARGETDURATION` MUST holds for *every*
245    /// segment this origin has ever advertised, not just the ones still in
246    /// the window (mirrors the deleted `MediaStore::max_segment_duration`).
247    max_segment_duration_secs: f64,
248    /// Cumulative count of discontinuities that have rolled off the front of
249    /// the window — RFC 8216 §4.3.3.3's `#EXT-X-DISCONTINUITY-SEQUENCE`.
250    /// Incremented exactly once per **evicted** entry whose
251    /// [`WindowSegment::discontinuous`] was `true`; a discontinuity still
252    /// inside the window is rendered as a per-segment `#EXT-X-DISCONTINUITY`
253    /// tag instead (see [`MediaPlaylist::to_m3u8`]), never double-counted
254    /// here.
255    discontinuity_sequence: u64,
256}
257
258impl Window {
259    fn new(capacity: NonZeroUsize) -> Self {
260        Window {
261            segments: VecDeque::new(),
262            capacity: capacity.get(),
263            max_segment_duration_secs: 0.0,
264            discontinuity_sequence: 0,
265        }
266    }
267
268    /// Absorb one drained [`SegmentEntry`], evicting the oldest window entry
269    /// first if already at `capacity` — same evict-then-push shape as every
270    /// ring in `trunk.rs` itself.
271    fn push(&mut self, entry: SegmentEntry) {
272        let duration_secs = entry.duration.as_secs_f64();
273        self.max_segment_duration_secs = self.max_segment_duration_secs.max(duration_secs);
274        if self.segments.len() == self.capacity {
275            if let Some(evicted) = self.segments.pop_front() {
276                if evicted.discontinuous {
277                    self.discontinuity_sequence += 1;
278                }
279            }
280        }
281        self.segments.push_back(WindowSegment {
282            sequence_number: entry.sequence_number,
283            bytes: entry.bytes,
284            duration_secs,
285            discontinuous: entry.meta.discontinuous,
286        });
287    }
288
289    fn bytes_of(&self, sequence_number: u32) -> Option<Bytes> {
290        self.segments
291            .iter()
292            .find(|s| s.sequence_number == sequence_number)
293            .map(|s| s.bytes.clone())
294    }
295}
296
297/// Parse a `part-{track}-{seq}.{idx}.{ext}` dynamic filename into
298/// `(seq, idx)`, or `None` if it isn't a part filename in `container`'s own
299/// extension (or its numeric fields don't parse). `{track}` is validated but
300/// unused (matches every other dynamic-filename resource in this module).
301fn parse_part(file: &str, container: Container) -> Option<(u32, u32)> {
302    let suffix = format!(".{}", container.segment_extension());
303    let rest = file.strip_prefix("part-")?.strip_suffix(suffix.as_str())?;
304    let (track_seq, idx) = rest.rsplit_once('.')?;
305    let (track, seq) = track_seq.split_once('-')?;
306    track.parse::<u32>().ok()?;
307    Some((seq.parse().ok()?, idx.parse().ok()?))
308}
309
310/// Parse a `init-{track}.mp4`/`seg-{track}-{seq}.{ext}` dynamic filename;
311/// `part-…` filenames are handled separately by [`parse_part`] (they can
312/// block until available). `{track}` is validated as a number but otherwise
313/// unused: an [`HlsOrigin`] holds a single track's data (see
314/// [`DEFAULT_TRACK_ID`]).
315///
316/// The `Init` variant is only ever recognised under [`Container::Fmp4`] — a
317/// `MpegTs` origin's grammar has no init resource at all (its segments are
318/// self-initialising; see [`Container`]'s own doc), so `init-*.mp4` under
319/// `MpegTs` falls through to `None` regardless of whether
320/// [`HlsOrigin::set_init`] was ever called. This is issue #873's
321/// cross-container refusal: advertised == servable, and an `MpegTs` origin
322/// never advertises an init segment to begin with.
323enum ImmediateResource {
324    Init,
325    Segment(u32),
326}
327
328fn parse_immediate(file: &str, container: Container) -> Option<ImmediateResource> {
329    if container == Container::Fmp4 {
330        if let Some(rest) = file.strip_prefix("init-") {
331            let track = rest.strip_suffix(".mp4")?;
332            track.parse::<u32>().ok()?;
333            return Some(ImmediateResource::Init);
334        }
335    }
336    if let Some(rest) = file.strip_prefix("seg-") {
337        let suffix = format!(".{}", container.segment_extension());
338        let rest = rest.strip_suffix(suffix.as_str())?;
339        let (track, seq) = rest.split_once('-')?;
340        track.parse::<u32>().ok()?;
341        return Some(ImmediateResource::Segment(seq.parse().ok()?));
342    }
343    None
344}
345
346/// Error returned by [`HlsOriginBuilder::build`] when a required field was
347/// never set — never a silently-defaulted value (issue #873).
348#[derive(Debug, Clone, Copy, PartialEq, Eq, thiserror::Error)]
349#[non_exhaustive]
350pub enum HlsOriginBuildError {
351    /// [`HlsOriginBuilder::target_duration_secs`] was never called.
352    #[error("HlsOrigin::builder(...).target_duration_secs(...) is required but was never called")]
353    MissingTargetDurationSecs,
354    /// [`HlsOriginBuilder::window_segments`] was never called.
355    #[error("HlsOrigin::builder(...).window_segments(...) is required but was never called")]
356    MissingWindowSegments,
357}
358
359/// Fluent builder for [`HlsOrigin`] (issue #873) — replaces the old
360/// four-positional `HlsOrigin::new` (deleted; this crate is at 0.4.0
361/// unpublished, so there is no compatibility burden), which could not
362/// express "classic HLS, no low latency" at all since `part_target_ms` was a
363/// mandatory positional argument.
364///
365/// ```
366/// # use std::num::NonZeroUsize;
367/// # use std::sync::Arc;
368/// # use hls_runtime::server::{Container, HlsOrigin};
369/// # use media_plane::trunk::{Trunk, TrunkConfig};
370/// # let nz = |n: usize| NonZeroUsize::new(n).unwrap();
371/// # let trunk = Arc::new(Trunk::new(TrunkConfig::new(nz(16), nz(4), nz(8), nz(4), nz(16))));
372/// let classic_ts = HlsOrigin::builder(Arc::clone(&trunk))
373///     .target_duration_secs(6.0)
374///     .window_segments(nz(4))
375///     .container(Container::MpegTs)
376///     // `.low_latency(..)` omitted entirely -> classic HLS.
377///     .build()
378///     .expect("both required fields were set");
379/// # let _ = classic_ts;
380/// ```
381pub struct HlsOriginBuilder {
382    trunk: Arc<Trunk>,
383    target_duration_secs: Option<f64>,
384    window_segments: Option<NonZeroUsize>,
385    container: Container,
386    part_target_ms: Option<u32>,
387}
388
389impl HlsOriginBuilder {
390    fn new(trunk: Arc<Trunk>) -> Self {
391        HlsOriginBuilder {
392            trunk,
393            target_duration_secs: None,
394            window_segments: None,
395            container: Container::default(),
396            part_target_ms: None,
397        }
398    }
399
400    /// `#EXT-X-TARGETDURATION`'s configured floor (RFC 8216bis §4.4.3.1) —
401    /// required; [`HlsOriginBuilder::build`] errors if this is never called.
402    /// The actually-rendered value is raised to the largest real segment
403    /// duration seen, if that ever exceeds this (see `render_playlist`).
404    pub fn target_duration_secs(mut self, target_duration_secs: f64) -> Self {
405        self.target_duration_secs = Some(target_duration_secs);
406        self
407    }
408
409    /// How many closed segments this origin advertises in a rendered Media
410    /// Playlist — required; independent of
411    /// [`media_plane::trunk::TrunkConfig::segment_capacity`] (the `Trunk`'s
412    /// own retention bound): a caller may legitimately want a shorter
413    /// advertised window than the `Trunk` retains for other consumers (e.g. a
414    /// DVR `SegmentEgress` reading the same `Trunk`).
415    pub fn window_segments(mut self, window_segments: NonZeroUsize) -> Self {
416        self.window_segments = Some(window_segments);
417        self
418    }
419
420    /// Which container this origin serves segments/parts as. Defaults to
421    /// [`Container::Fmp4`] if never called, matching every pre-#873 caller's
422    /// behaviour. Orthogonal to [`Self::low_latency`] — all four
423    /// `{Fmp4, MpegTs} x {classic, low-latency}` combinations are valid.
424    pub fn container(mut self, container: Container) -> Self {
425        self.container = container;
426        self
427    }
428
429    /// Opt into LL-HLS: `part_target_ms` becomes `#EXT-X-PART-INF`'s
430    /// `PART-TARGET` (milliseconds). **Omit this call entirely for classic
431    /// HLS** — no `#EXT-X-PART`/`#EXT-X-PART-INF`/`#EXT-X-SERVER-CONTROL`/
432    /// `#EXT-X-PRELOAD-HINT` tags are then rendered, regardless of
433    /// [`Self::container`]. This is what the old constructor's mandatory
434    /// `part_target_ms` positional could not express.
435    pub fn low_latency(mut self, part_target_ms: u32) -> Self {
436        self.part_target_ms = Some(part_target_ms);
437        self
438    }
439
440    /// Build the [`HlsOrigin`], subscribing its one [`SegmentCursor`]
441    /// immediately (so the window starts empty but never misses a segment
442    /// published from this point on).
443    ///
444    /// Errors, never silently defaults, if [`Self::target_duration_secs`] or
445    /// [`Self::window_segments`] was never called.
446    pub fn build(self) -> Result<HlsOrigin, HlsOriginBuildError> {
447        let target_duration_secs = self
448            .target_duration_secs
449            .ok_or(HlsOriginBuildError::MissingTargetDurationSecs)?;
450        let window_segments = self
451            .window_segments
452            .ok_or(HlsOriginBuildError::MissingWindowSegments)?;
453        let cursor = self.trunk.subscribe_segments();
454        Ok(HlsOrigin {
455            trunk: self.trunk,
456            cursor: Mutex::new(cursor),
457            window: Mutex::new(Window::new(window_segments)),
458            init: Mutex::new(None),
459            target_duration_secs,
460            container: self.container,
461            part_target_ms: self.part_target_ms,
462        })
463    }
464}
465
466/// The LL-HLS origin [`ServedEgress`]: renders playlists and resolves
467/// blocking-reload/part-availability requests for one stream, backed by a
468/// shared [`Trunk`]. See this module's own doc for exactly what comes
469/// straight from the `Trunk` and what needs the small synced `Window`.
470pub struct HlsOrigin {
471    trunk: Arc<Trunk>,
472    /// This origin's **one** [`SegmentCursor`] — see [`Trunk::subscribe_segments`]'s
473    /// own docs (and this crate's `media_plane::egress` module doc) for why a
474    /// `ServedEgress` must never take one per request/peer.
475    cursor: Mutex<SegmentCursor>,
476    window: Mutex<Window>,
477    /// The fMP4 init segment — see this module's doc for why this, alone, is
478    /// not answerable by any `Trunk` ring.
479    init: Mutex<Option<Bytes>>,
480    target_duration_secs: f64,
481    container: Container,
482    /// `Some(part_target_ms)` enables LL-HLS; `None` renders classic HLS —
483    /// no `#EXT-X-PART`/`#EXT-X-PART-INF`/`#EXT-X-SERVER-CONTROL`/
484    /// `#EXT-X-PRELOAD-HINT` at all, orthogonal to [`Self::container`] (issue
485    /// #873).
486    part_target_ms: Option<u32>,
487}
488
489impl HlsOrigin {
490    /// Start building an [`HlsOrigin`] over `trunk` — see [`HlsOriginBuilder`]
491    /// for the required fields (`target_duration_secs`/`window_segments`),
492    /// the container choice, and how to opt into LL-HLS.
493    pub fn builder(trunk: Arc<Trunk>) -> HlsOriginBuilder {
494        HlsOriginBuilder::new(trunk)
495    }
496
497    /// Store the init segment bytes — see this module's doc for why an init
498    /// segment is not something any `Trunk` ring holds.
499    ///
500    /// **Documented no-op under [`Container::MpegTs`]**: this origin's
501    /// `MpegTs` grammar has no init resource and never emits `#EXT-X-MAP`
502    /// by default (see [`Container`]'s own doc), so bytes stored here are
503    /// never advertised or served in that mode. The method stays callable
504    /// regardless of container so a caller sharing one code path across
505    /// both (e.g. a segmenter that always calls `set_init` once available)
506    /// does not need to branch on which container it configured.
507    pub fn set_init(&self, bytes: impl Into<Bytes>) {
508        *self.init.lock().unwrap() = Some(bytes.into());
509    }
510
511    /// The fMP4 init segment bytes, if set.
512    pub fn init_bytes(&self) -> Option<Bytes> {
513        self.init.lock().unwrap().clone()
514    }
515
516    /// Drain this origin's [`SegmentCursor`] into `Window` — called at the
517    /// top of every [`ServedEgress::resolve`] so a render always reflects
518    /// whatever has published since the last call. Non-blocking, bounded by
519    /// however many segments actually published since the last drain.
520    ///
521    /// A [`SegmentCursorItem::Lagged`] report (this origin's `window_segments`/
522    /// polling cadence fell behind the `Trunk`'s own
523    /// `segment_capacity` eviction) is accepted, not treated as an error:
524    /// exactly like every other lossy cursor in this workspace, the honest
525    /// response is to resume from the next segment, not to fabricate the
526    /// lost entries' duration/discontinuity data.
527    fn drain(&self) {
528        let mut cursor = self.cursor.lock().unwrap();
529        let mut window = self.window.lock().unwrap();
530        while let Some(item) = cursor.poll() {
531            if let SegmentCursorItem::Segment(entry) = item {
532                window.push(entry);
533            }
534        }
535    }
536
537    /// `(in-progress-or-last-active segment sequence number, its currently
538    /// resident live parts)` — the `Trunk`-only replacement for the deleted
539    /// `MediaStore::latest_progress`.
540    ///
541    /// Derivation: the only segment that can possibly have live, not-yet-
542    /// closed parts is the one immediately after
543    /// [`Trunk::last_closed_segment`] (a segmenter never opens segment N+2's
544    /// parts before N+1 closes) — so probing exactly that one candidate via
545    /// [`Trunk::parts_in_segment`] is exact, not a heuristic. If that probe
546    /// is empty (nothing has started for the next segment yet — e.g. the
547    /// instant after a close, before its successor's first part lands), the
548    /// answer falls back to `last_closed_segment` itself, with an empty part
549    /// list — exactly the degenerate state `MediaStore::latest_progress`
550    /// also returned right after `add_segment` cleared `live_parts`.
551    fn live_edge(&self) -> (u32, Vec<PartEntry>) {
552        let last_closed = self.trunk.last_closed_segment().unwrap_or(0);
553        let candidate = last_closed + 1;
554        let parts = self.trunk.parts_in_segment(candidate);
555        if parts.is_empty() {
556            (last_closed, Vec::new())
557        } else {
558            (candidate, parts)
559        }
560    }
561
562    /// Render the LL-HLS media playlist for `track_id` from this origin's
563    /// current `Window` (closed segments) and the `Trunk`'s live edge (open
564    /// segment's parts + preload hint).
565    ///
566    /// RFC 8216bis §4.4.4.9: an in-progress (not yet closed) segment MUST NOT
567    /// be advertised with an `#EXTINF`/URI pair — that segment has no
568    /// fetchable resource yet — it may only appear as trailing `#EXT-X-PART`
569    /// lines. `broadcast_hls::MediaPlaylist::open_segment` is exactly this
570    /// representation: its parts render as trailing `#EXT-X-PART` lines with
571    /// no `#EXTINF`/URI, so the in-progress segment's parts and the
572    /// `#EXT-X-PRELOAD-HINT` for the next, not-yet-available part are both
573    /// rendered by `to_m3u8()` itself — this method only supplies the URI
574    /// scheme (`part-<track>-<seq>.<idx>.m4s`) and the part metadata.
575    fn render_playlist(&self, track_id: u32) -> String {
576        self.drain();
577        let window = self.window.lock().unwrap();
578        let (open_seq, open_parts) = self.live_edge();
579        // Only render an open segment/preload-hint once the live edge is
580        // genuinely a not-yet-closed segment with at least one live part —
581        // never re-render an already-closed segment's lingering parts (the
582        // `Trunk`'s live-part log deliberately does not evict them on close;
583        // see `trunk`'s own module doc) as if they were still open.
584        // Classic HLS (no `.low_latency(...)` call, issue #873) never
585        // advertises an in-progress segment at all — RFC 8216bis §4.4.4.9's
586        // trailing-`#EXT-X-PART`-only representation is itself an LL-HLS
587        // directive, so it is gated on low latency being enabled, not merely
588        // on the Trunk happening to have live parts.
589        let low_latency_enabled = self.part_target_ms.is_some();
590        let has_open_parts = low_latency_enabled && !open_parts.is_empty();
591        let ext = self.container.segment_extension();
592
593        let media_sequence = window
594            .segments
595            .front()
596            .map(|s| u64::from(s.sequence_number))
597            .or_else(|| has_open_parts.then_some(u64::from(open_seq)))
598            .unwrap_or(1);
599        let segments: Vec<MediaSegment> = window
600            .segments
601            .iter()
602            .map(|s| MediaSegment {
603                uri: format!("seg-{track_id}-{}.{ext}", s.sequence_number),
604                duration: s.duration_secs,
605                discontinuous: s.discontinuous,
606                parts: Vec::new(),
607                ..Default::default()
608            })
609            .collect();
610        let open_segment = has_open_parts.then(|| {
611            OpenSegment::new(
612                open_parts
613                    .iter()
614                    .map(|p| PartSpec {
615                        uri: format!(
616                            "part-{track_id}-{}.{}.{ext}",
617                            p.segment_number, p.part_index
618                        ),
619                        duration: p.duration.as_secs_f64(),
620                        independent: p.independent,
621                        ..Default::default()
622                    })
623                    .collect(),
624            )
625        });
626        let next_part_hint = has_open_parts.then(|| {
627            let next_idx = open_parts
628                .iter()
629                .map(|p| p.part_index)
630                .max()
631                .map(|idx| idx + 1)
632                .unwrap_or(0);
633            format!("part-{track_id}-{open_seq}.{next_idx}.{ext}")
634        });
635        // RFC 8216bis §4.4.3.1 (MUST): every Media Segment's EXTINF duration,
636        // rounded to the nearest integer, MUST be <= TARGETDURATION. The
637        // segmenter cuts on the next keyframe *after* the configured target,
638        // so a real segment routinely exceeds it — advertising the
639        // configured target alone can under-declare. Use whichever is
640        // larger, rounded (not the configured value's `ceil()` alone).
641        let target_duration = self
642            .target_duration_secs
643            .max(window.max_segment_duration_secs)
644            .round() as u32;
645        // `#EXT-X-MAP`: unconditional under Fmp4 (RFC 8216bis §3.1.2 MUST);
646        // omitted under MpegTs by default (§3.1.1's PAT/PMT-or-MAP
647        // disjunction — see `Container`'s own doc for why this is a default,
648        // not a hard restriction).
649        let extra_tags = match self.container {
650            Container::Fmp4 => vec![format!("#EXT-X-MAP:URI=\"init-{track_id}.mp4\"")],
651            Container::MpegTs => Vec::new(),
652        };
653        let low_latency = low_latency_enabled.then(|| {
654            let part_target_ms = self
655                .part_target_ms
656                .expect("low_latency_enabled implies Some");
657            let part_target = f64::from(part_target_ms) / 1000.0;
658            LowLatencyConfig {
659                part_target,
660                part_hold_back: part_target * PART_HOLD_BACK_MULTIPLIER,
661                preload_hint_part: next_part_hint,
662                ..Default::default()
663            }
664        });
665        let playlist = MediaPlaylist {
666            // No explicit floor: `broadcast_hls::MediaPlaylist::to_m3u8`
667            // computes `EXT-X-VERSION` from the content actually emitted
668            // (RFC 8216bis §8) rather than this origin choosing a value
669            // ahead of time (issue #871) — none of the LL-HLS directives
670            // below (`EXT-X-PART`/`EXT-X-PART-INF`/`EXT-X-PRELOAD-HINT`/
671            // `EXT-X-SERVER-CONTROL`) carry any version requirement at all.
672            target_duration,
673            media_sequence,
674            discontinuity_sequence: window.discontinuity_sequence,
675            segments,
676            open_segment,
677            endlist: false,
678            extra_tags,
679            low_latency,
680            iframes_only: false,
681            ..Default::default()
682        };
683        playlist.to_m3u8()
684    }
685
686    fn resolve_playlist(
687        &self,
688        track_id: u32,
689        query: BlockingQuery,
690        now: Timestamp,
691        await_policy: AwaitPolicy,
692    ) -> EgressResponse<HlsBody> {
693        if query.hls_part.is_some() && query.hls_msn.is_none() {
694            return EgressResponse::BadRequest {
695                reason: "_HLS_part without _HLS_msn is meaningless",
696            };
697        }
698        if let Some(msn) = query.hls_msn {
699            let (in_progress_seg, live_parts) = self.live_edge();
700            if msn > u64::from(in_progress_seg) + ABUSE_MSN_FUTURE_BOUND {
701                return EgressResponse::BadRequest {
702                    reason: "_HLS_msn unreasonably far beyond the live edge",
703                };
704            }
705            let satisfied = match query.hls_part {
706                Some(part) => {
707                    u64::from(in_progress_seg) > msn
708                        || (u64::from(in_progress_seg) == msn
709                            && live_parts.len() as u64 > u64::from(part))
710                }
711                None => self.trunk.last_closed_segment().unwrap_or(0) as u64 >= msn,
712            };
713            if !satisfied {
714                return EgressResponse::pending(await_policy, now, now);
715            }
716        }
717        EgressResponse::Ready {
718            body: HlsBody::Playlist(self.render_playlist(track_id)),
719            cache: CachePolicy::NoCache,
720        }
721    }
722
723    /// A part request is the preload-hinted Partial Segment a client fetches
724    /// ahead of time (RFC 8216bis §6.2.2, §6.3.1). If the origin promised it
725    /// via `#EXT-X-PRELOAD-HINT` but hasn't produced it yet,
726    /// [`EgressResponse::Await`] — the caller should hold the request open
727    /// (not 404 immediately, which spams errors and defeats low latency).
728    /// [`EgressResponse::NotFound`] is returned **promptly** (without the
729    /// caller needing to wait out its own [`AwaitPolicy`]) once the part can
730    /// no longer appear: its segment has closed (now only addressable as a
731    /// whole segment via `seg-…`) — a legitimate 404 the client answers by
732    /// fetching the next segment/part.
733    fn resolve_resource(
734        &self,
735        name: &str,
736        now: Timestamp,
737        await_policy: AwaitPolicy,
738    ) -> EgressResponse<HlsBody> {
739        if let Some((seq, idx)) = parse_part(name, self.container) {
740            if let Some(bytes) = self.trunk.part_bytes(seq, idx) {
741                return EgressResponse::Ready {
742                    body: HlsBody::Resource(bytes),
743                    cache: CachePolicy::Immutable,
744                };
745            }
746            // The requested part's segment has already closed (whether or
747            // not this origin's own `Window` still retains its bytes) -> it
748            // will never be produced. `Trunk::last_closed_segment` answers
749            // this exactly, with no dependence on `Window`'s retention.
750            let never_will = self.trunk.last_closed_segment().is_some_and(|c| c >= seq);
751            return if never_will {
752                EgressResponse::NotFound
753            } else {
754                EgressResponse::pending(await_policy, now, now)
755            };
756        }
757        self.drain();
758        let bytes = match parse_immediate(name, self.container) {
759            Some(ImmediateResource::Init) => self.init_bytes(),
760            Some(ImmediateResource::Segment(seq)) => self.window.lock().unwrap().bytes_of(seq),
761            None => None,
762        };
763        match bytes {
764            Some(bytes) => EgressResponse::Ready {
765                body: HlsBody::Resource(bytes),
766                cache: CachePolicy::Immutable,
767            },
768            None => EgressResponse::NotFound,
769        }
770    }
771}
772
773impl ServedEgress for HlsOrigin {
774    type Request = HlsRequest;
775    type Body = HlsBody;
776
777    fn resolve(
778        &self,
779        request: HlsRequest,
780        now: Timestamp,
781        await_policy: AwaitPolicy,
782    ) -> EgressResponse<HlsBody> {
783        match request {
784            HlsRequest::Playlist { track_id, query } => {
785                self.resolve_playlist(track_id, query, now, await_policy)
786            }
787            HlsRequest::Resource { name } => self.resolve_resource(&name, now, await_policy),
788        }
789    }
790}
791
792#[cfg(test)]
793mod tests {
794    use super::*;
795    use media_plane::trunk::TrunkConfig;
796    use std::time::{Duration, Instant};
797    use transmux::SegmentMeta;
798
799    fn nz(n: usize) -> NonZeroUsize {
800        NonZeroUsize::new(n).expect("test capacity must be non-zero")
801    }
802
803    /// A fresh `Trunk` sized generously for these tests, plus the one
804    /// `HlsOrigin` under test — `Fmp4` + low-latency, matching every
805    /// pre-#873 test in this module (regression guard).
806    fn make_origin() -> (Arc<Trunk>, HlsOrigin, media_plane::trunk::SegmentWriter) {
807        let trunk = Trunk::new(TrunkConfig::new(nz(64), nz(8), nz(8), nz(8), nz(64)));
808        let writer = trunk.segment_writer().expect("first segment writer");
809        let origin = HlsOrigin::builder(Arc::clone(&trunk))
810            .target_duration_secs(4.0)
811            .window_segments(nz(4))
812            .low_latency(500)
813            .build()
814            .expect("both required fields set");
815        origin.set_init(vec![0xAAu8; 8]);
816        (trunk, origin, writer)
817    }
818
819    fn seg(
820        writer: &media_plane::trunk::SegmentWriter,
821        seq: u32,
822        duration_secs: f64,
823        discontinuous: bool,
824    ) {
825        writer.publish_segment(SegmentEntry::new(
826            Bytes::from(vec![seq as u8; 8]),
827            seq,
828            Duration::from_secs_f64(duration_secs),
829            Timestamp::from_nanos(0),
830            SegmentMeta { discontinuous },
831        ));
832    }
833
834    fn part(writer: &media_plane::trunk::SegmentWriter, seg_no: u32, idx: u32, independent: bool) {
835        writer.publish_part(PartEntry::new(
836            Bytes::from(vec![idx as u8; 4]),
837            seg_no,
838            idx,
839            Duration::from_millis(500),
840            independent,
841        ));
842    }
843
844    fn resolve_now(origin: &HlsOrigin, request: HlsRequest) -> EgressResponse<HlsBody> {
845        origin.resolve(
846            request,
847            Timestamp::from_nanos(0),
848            AwaitPolicy::new(Timestamp::from_nanos(0)),
849        )
850    }
851
852    // --- master playlist (unaffected by the Trunk migration) -------------
853
854    #[test]
855    fn master_playlist_has_stream_inf() {
856        let m = master_playlist_m3u8("media.m3u8");
857        assert!(m.contains("#EXTM3U"));
858        assert!(m.contains("#EXT-X-STREAM-INF"));
859        assert!(m.contains("media.m3u8"));
860    }
861
862    #[test]
863    fn master_playlist_points_at_configured_playlist_name() {
864        let m = master_playlist_m3u8("index.m3u8");
865        assert!(m.contains("index.m3u8"));
866        assert!(!m.contains("media.m3u8"));
867    }
868
869    // --- 1. playlist rendered from a populated Trunk matches the expected
870    //        shape ---------------------------------------------------------
871
872    /// MUTATION VERIFIED: changing `render_playlist`'s
873    /// `low_latency: Some(...)` to `None` makes this test's
874    /// `assert!(m.contains("#EXT-X-PART-INF"))` (and every other
875    /// LL-HLS-tag assertion) fail — `to_m3u8()` omits the entire
876    /// low-latency header block when `low_latency` is `None`, so none of
877    /// `#EXT-X-PART-INF`/`#EXT-X-SERVER-CONTROL`/`#EXT-X-PART` appear in the
878    /// rendered body. Recompiled and re-run to confirm the failure, then
879    /// reverted.
880    #[test]
881    fn playlist_rendered_from_populated_trunk_matches_expected_shape() {
882        let (_trunk, origin, writer) = make_origin();
883        seg(&writer, 1, 4.0, false);
884        part(&writer, 2, 0, true);
885        part(&writer, 2, 1, false);
886
887        let body = match resolve_now(
888            &origin,
889            HlsRequest::Playlist {
890                track_id: DEFAULT_TRACK_ID,
891                query: BlockingQuery::default(),
892            },
893        ) {
894            EgressResponse::Ready {
895                body: HlsBody::Playlist(m),
896                cache,
897            } => {
898                assert_eq!(cache, CachePolicy::NoCache);
899                m
900            }
901            other => panic!("expected Ready(Playlist), got {other:?}"),
902        };
903
904        // RFC 8216bis §8 (issue #871): this playlist's true minimum is 6
905        // (EXT-X-MAP without EXT-X-I-FRAMES-ONLY) — none of the LL-HLS
906        // directives it also carries require any version at all. The old
907        // hardcoded `EXT-X-VERSION:9` over-declared and would have locked
908        // out every client on protocol version 6, 7, or 8.
909        assert!(body.contains("#EXT-X-VERSION:6"), "body: {body}");
910        assert!(!body.contains("#EXT-X-VERSION:9"), "body: {body}");
911        assert!(body.contains("#EXT-X-TARGETDURATION:4"), "body: {body}");
912        assert!(
913            body.contains("#EXT-X-SERVER-CONTROL:CAN-BLOCK-RELOAD=YES,PART-HOLD-BACK=1.5"),
914            "body: {body}"
915        );
916        assert!(
917            body.contains("#EXT-X-PART-INF:PART-TARGET=0.5"),
918            "body: {body}"
919        );
920        assert!(
921            body.contains("#EXT-X-MAP:URI=\"init-1.mp4\""),
922            "body: {body}"
923        );
924        assert!(body.contains("seg-1-1.m4s"), "body: {body}");
925        assert!(
926            body.contains("#EXT-X-PART:DURATION=0.5") && body.contains("INDEPENDENT=YES"),
927            "body: {body}"
928        );
929        assert!(body.contains("#EXT-X-PRELOAD-HINT"), "body: {body}");
930        assert!(
931            body.contains("part-1-2.2.m4s"),
932            "preload hint for the next part: {body}"
933        );
934    }
935
936    // --- 2. a preload-hinted part BLOCKS until produced, then serves -----
937
938    /// MUTATION VERIFIED: changing `resolve_resource`'s `never_will` check
939    /// (whether the requested part's segment has already closed, via
940    /// `last_closed_segment`) to always `true` ("never will produce this
941    /// part") makes this test's first assertion fail: the not-yet-produced
942    /// part resolves `NotFound` immediately instead of `Await`, so
943    /// `assert!(matches!(first, EgressResponse::Await { .. }))` sees
944    /// `NotFound` and fails. Recompiled and re-run to confirm the failure,
945    /// then reverted. This is the RFC 8216bis section 6.2.2 behaviour that
946    /// shipped as multimux 0.2.1's bug fix — regressing it would break the
947    /// live camera route.
948    #[test]
949    fn preload_hinted_part_blocks_until_produced_then_serves() {
950        let (trunk, origin, writer) = make_origin();
951        let origin = Arc::new(origin);
952
953        // Not produced yet: must Await, not NotFound.
954        let deadline = Timestamp::from_nanos(5_000_000_000);
955        let policy = AwaitPolicy::new(deadline);
956        let first = origin.resolve(
957            HlsRequest::Resource {
958                name: "part-1-1.0.m4s".to_string(),
959            },
960            Timestamp::from_nanos(0),
961            policy,
962        );
963        assert!(
964            matches!(first, EgressResponse::Await { .. }),
965            "expected Await before the part exists, got {first:?}"
966        );
967
968        // Register a real Trunk::listen() wake-up and block a worker thread
969        // on it -- the actual mechanism a real adapter (Step 5) uses, not a
970        // poll loop -- to prove the part genuinely blocks rather than
971        // merely returning Await once and never resolving.
972        let listener = trunk.listen().expect("listener slot available");
973        let woken = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false));
974        let woken2 = std::sync::Arc::clone(&woken);
975        // HANG GUARD (issue #807): deliberately generous, same reasoning as
976        // `media-plane/src/trunk.rs`'s own `Trunk::listen()` wake tests --
977        // the claim is "wakes rather than parking forever", not "wakes
978        // within N seconds"; the publish happens on another thread, so a
979        // tight bound would measure the machine's scheduler, not this code.
980        let waiter = std::thread::spawn(move || {
981            let ok = listener.wait_deadline(Instant::now() + Duration::from_secs(60));
982            woken2.store(ok, std::sync::atomic::Ordering::SeqCst);
983        });
984
985        // Produce the part the request was waiting on.
986        part(&writer, 1, 0, true);
987
988        waiter.join().expect("waiter thread must not panic");
989        assert!(
990            woken.load(std::sync::atomic::Ordering::SeqCst),
991            "Trunk::listen() must wake once publish_part lands"
992        );
993
994        // Re-resolving now must serve it -- not 404.
995        match origin.resolve(
996            HlsRequest::Resource {
997                name: "part-1-1.0.m4s".to_string(),
998            },
999            Timestamp::from_nanos(1),
1000            policy,
1001        ) {
1002            EgressResponse::Ready {
1003                body: HlsBody::Resource(bytes),
1004                cache,
1005            } => {
1006                assert_eq!(bytes, Bytes::from(vec![0u8; 4]));
1007                assert_eq!(cache, CachePolicy::Immutable);
1008            }
1009            other => panic!("expected Ready once produced, got {other:?}"),
1010        }
1011    }
1012
1013    /// MUTATION VERIFIED: removing `EgressResponse::pending`'s expiry check
1014    /// (i.e. always returning `Await`) would make a client wait forever for
1015    /// a part that will never exist -- this test proves the OTHER half of
1016    /// the bound: once `now` reaches the caller's own `AwaitPolicy::deadline`,
1017    /// resolve must stop Awaiting. Changing the deadline comparison in
1018    /// `resolve_resource`'s `EgressResponse::pending(await_policy, now, now)`
1019    /// call to ignore `now` (always pass `Timestamp::from_nanos(0)`) makes
1020    /// this test's final assertion fail: `resolve` at `now == deadline`
1021    /// keeps returning `Await` instead of `NotFound`. Recompiled and re-run
1022    /// to confirm the failure, then reverted.
1023    #[test]
1024    fn awaiting_part_is_bounded_by_await_policy_deadline() {
1025        let (_trunk, origin, _writer) = make_origin();
1026        let deadline = Timestamp::from_nanos(1_000_000_000);
1027        let policy = AwaitPolicy::new(deadline);
1028
1029        let still_waiting = origin.resolve(
1030            HlsRequest::Resource {
1031                name: "part-1-9.0.m4s".to_string(),
1032            },
1033            Timestamp::from_nanos(999_999_999),
1034            policy,
1035        );
1036        assert!(matches!(still_waiting, EgressResponse::Await { .. }));
1037
1038        let expired = origin.resolve(
1039            HlsRequest::Resource {
1040                name: "part-1-9.0.m4s".to_string(),
1041            },
1042            deadline,
1043            policy,
1044        );
1045        assert!(
1046            matches!(expired, EgressResponse::NotFound),
1047            "expected NotFound once the deadline passed, got {expired:?}"
1048        );
1049    }
1050
1051    // --- 3. a just-closed segment's final part still serves ---------------
1052
1053    /// MUTATION VERIFIED: this behaviour depends entirely on
1054    /// `media_plane::trunk::SegmentWriter::publish_segment` (`media-plane/src/trunk.rs`) never
1055    /// touching the live-part log. Simulating the old `MediaStore` bug here
1056    /// by having `resolve_resource` check `last_closed_segment() >= seq`
1057    /// ("this segment already closed -> NotFound") **before** checking
1058    /// `Trunk::part_bytes` (i.e. swapping the two checks' order) makes this
1059    /// test's first assertion fail: the just-closed segment's final part
1060    /// resolves `NotFound` instead of `Ready` (`panicked at ...: the
1061    /// just-closed segment's final part must still serve, got NotFound`),
1062    /// because the eager closed-check now shadows the still-valid
1063    /// `part_bytes` hit. Recompiled and re-run to confirm the failure, then
1064    /// reverted. This is the RFC 8216bis boundary behaviour that shipped as
1065    /// multimux 0.2.2's bug fix — regressing it would break the live camera
1066    /// route (its own `#EXT-X-PRELOAD-HINT` part races exactly this
1067    /// boundary every segment).
1068    #[test]
1069    fn just_closed_segment_final_part_still_serves() {
1070        let (_trunk, origin, writer) = make_origin();
1071        part(&writer, 1, 0, true);
1072        part(&writer, 1, 1, false); // segment 1's final part
1073        seg(&writer, 1, 4.0, false); // close segment 1
1074
1075        match resolve_now(
1076            &origin,
1077            HlsRequest::Resource {
1078                name: "part-1-1.1.m4s".to_string(),
1079            },
1080        ) {
1081            EgressResponse::Ready {
1082                body: HlsBody::Resource(bytes),
1083                ..
1084            } => assert_eq!(bytes, Bytes::from(vec![1u8; 4])),
1085            other => panic!("the just-closed segment's final part must still serve, got {other:?}"),
1086        }
1087
1088        // A genuinely-nonexistent part of the closed segment is NotFound.
1089        assert_eq!(
1090            resolve_now(
1091                &origin,
1092                HlsRequest::Resource {
1093                    name: "part-1-1.9.m4s".to_string(),
1094                }
1095            ),
1096            EgressResponse::NotFound
1097        );
1098
1099        // The playlist must not resurrect the closed segment's parts as
1100        // "open" -- it is rendered whole.
1101        let body = match resolve_now(
1102            &origin,
1103            HlsRequest::Playlist {
1104                track_id: DEFAULT_TRACK_ID,
1105                query: BlockingQuery::default(),
1106            },
1107        ) {
1108            EgressResponse::Ready {
1109                body: HlsBody::Playlist(m),
1110                ..
1111            } => m,
1112            other => panic!("expected Ready(Playlist), got {other:?}"),
1113        };
1114        assert!(
1115            body.contains("seg-1-1.m4s"),
1116            "closed segment rendered whole: {body}"
1117        );
1118        assert!(
1119            !body.contains("part-1-1."),
1120            "closed parts not rendered as open: {body}"
1121        );
1122    }
1123
1124    // --- 4. MEDIA-SEQUENCE / DISCONTINUITY-SEQUENCE advance as the window
1125    //        rolls -----------------------------------------------------
1126
1127    /// MUTATION VERIFIED: changing `Window::push`'s eviction guard from
1128    /// `if evicted.discontinuous` to `if false` (never counting an evicted
1129    /// discontinuity) makes this test's
1130    /// `assert!(body.contains("#EXT-X-DISCONTINUITY-SEQUENCE:1"))` fail --
1131    /// the tag is omitted entirely (the renderer only emits it when
1132    /// `discontinuity_sequence > 0`), because the counter never advances
1133    /// past `0`. Recompiled and re-run to confirm the failure, then
1134    /// reverted.
1135    #[test]
1136    fn media_sequence_and_discontinuity_sequence_advance_as_window_rolls() {
1137        let (_trunk, origin, writer) = make_origin(); // window_segments = 4
1138
1139        seg(&writer, 1, 4.0, false);
1140        seg(&writer, 2, 4.0, true); // discontinuous
1141        seg(&writer, 3, 4.0, false);
1142        seg(&writer, 4, 4.0, false);
1143
1144        // Window (capacity 4) holds exactly 1..=4 -- MEDIA-SEQUENCE=1, and
1145        // segment 2's own #EXT-X-DISCONTINUITY renders in-window (no
1146        // DISCONTINUITY-SEQUENCE yet, nothing has rolled off).
1147        let body = match resolve_now(
1148            &origin,
1149            HlsRequest::Playlist {
1150                track_id: DEFAULT_TRACK_ID,
1151                query: BlockingQuery::default(),
1152            },
1153        ) {
1154            EgressResponse::Ready {
1155                body: HlsBody::Playlist(m),
1156                ..
1157            } => m,
1158            other => panic!("expected Ready(Playlist), got {other:?}"),
1159        };
1160        assert!(body.contains("#EXT-X-MEDIA-SEQUENCE:1"), "body: {body}");
1161        assert!(
1162            !body.contains("#EXT-X-DISCONTINUITY-SEQUENCE"),
1163            "nothing has rolled off the window yet: {body}"
1164        );
1165        assert!(body.contains("#EXT-X-DISCONTINUITY\n"), "body: {body}");
1166
1167        // Roll the window: segment 5 evicts segment 1 (not discontinuous;
1168        // DISCONTINUITY-SEQUENCE stays 0), segment 6 evicts segment 2
1169        // (discontinuous -- DISCONTINUITY-SEQUENCE becomes 1).
1170        seg(&writer, 5, 4.0, false);
1171        let body = match resolve_now(
1172            &origin,
1173            HlsRequest::Playlist {
1174                track_id: DEFAULT_TRACK_ID,
1175                query: BlockingQuery::default(),
1176            },
1177        ) {
1178            EgressResponse::Ready {
1179                body: HlsBody::Playlist(m),
1180                ..
1181            } => m,
1182            other => panic!("expected Ready(Playlist), got {other:?}"),
1183        };
1184        assert!(body.contains("#EXT-X-MEDIA-SEQUENCE:2"), "body: {body}");
1185        assert!(
1186            !body.contains("#EXT-X-DISCONTINUITY-SEQUENCE"),
1187            "evicted segment 1 was not discontinuous: {body}"
1188        );
1189
1190        seg(&writer, 6, 4.0, false);
1191        let body = match resolve_now(
1192            &origin,
1193            HlsRequest::Playlist {
1194                track_id: DEFAULT_TRACK_ID,
1195                query: BlockingQuery::default(),
1196            },
1197        ) {
1198            EgressResponse::Ready {
1199                body: HlsBody::Playlist(m),
1200                ..
1201            } => m,
1202            other => panic!("expected Ready(Playlist), got {other:?}"),
1203        };
1204        assert!(body.contains("#EXT-X-MEDIA-SEQUENCE:3"), "body: {body}");
1205        assert!(
1206            body.contains("#EXT-X-DISCONTINUITY-SEQUENCE:1"),
1207            "segment 2 (discontinuous) has now rolled off the window: {body}"
1208        );
1209    }
1210
1211    // --- misc: target-duration MUST, abuse bound, bad request -------------
1212
1213    #[test]
1214    fn target_duration_is_max_of_configured_and_actual_segment_duration() {
1215        let (_trunk, origin, writer) = make_origin(); // configured target 4.0
1216        seg(&writer, 1, 7.5, false);
1217        let body = match resolve_now(
1218            &origin,
1219            HlsRequest::Playlist {
1220                track_id: DEFAULT_TRACK_ID,
1221                query: BlockingQuery::default(),
1222            },
1223        ) {
1224            EgressResponse::Ready {
1225                body: HlsBody::Playlist(m),
1226                ..
1227            } => m,
1228            other => panic!("expected Ready(Playlist), got {other:?}"),
1229        };
1230        assert!(
1231            body.contains("#EXT-X-TARGETDURATION:8"),
1232            "TARGETDURATION must be round(7.5)=8, not the configured target: {body}"
1233        );
1234    }
1235
1236    #[test]
1237    fn far_future_msn_rejected() {
1238        let (_trunk, origin, writer) = make_origin();
1239        seg(&writer, 1, 4.0, false);
1240        let outcome = resolve_now(
1241            &origin,
1242            HlsRequest::Playlist {
1243                track_id: DEFAULT_TRACK_ID,
1244                query: BlockingQuery {
1245                    hls_msn: Some(1002),
1246                    hls_part: None,
1247                },
1248            },
1249        );
1250        assert!(matches!(outcome, EgressResponse::BadRequest { .. }));
1251    }
1252
1253    /// RFC 8216bis §6.2.5.2: `_HLS_msn` at `last_closed + 2` is the
1254    /// spec's bound — it MUST be accepted. With one segment closed (seq=1),
1255    /// the last closed is 1, so `_HLS_msn=3` (last_closed + 2) is accepted
1256    /// and the request blocks (returns Pending, not BadRequest).
1257    #[test]
1258    fn msn_at_spec_bound_is_accepted() {
1259        let (_trunk, origin, writer) = make_origin();
1260        seg(&writer, 1, 4.0, false);
1261        let outcome = resolve_now(
1262            &origin,
1263            HlsRequest::Playlist {
1264                track_id: DEFAULT_TRACK_ID,
1265                query: BlockingQuery {
1266                    hls_msn: Some(3),
1267                    hls_part: None,
1268                },
1269            },
1270        );
1271        assert!(
1272            !matches!(outcome, EgressResponse::BadRequest { .. }),
1273            "msn at spec bound (last_closed+2) must be accepted, not rejected"
1274        );
1275    }
1276
1277    /// RFC 8216bis §6.2.5.2: `_HLS_msn` one beyond the spec's +2 SHOULD
1278    /// boundary is rejected. With one segment closed (seq=1, last_closed=1),
1279    /// the live edge in_progress_seg is 1 (not 2, because no parts exist to
1280    /// advance it), so `_HLS_msn=4` (1 + 2 + 1) is rejected.
1281    #[test]
1282    fn msn_one_beyond_spec_bound_is_rejected() {
1283        let (_trunk, origin, writer) = make_origin();
1284        seg(&writer, 1, 4.0, false);
1285        // With segments up to 1 closed, in_progress_seg is 1.
1286        // _HLS_msn > 1 + 2 == 3 → rejected. So msn=4 is rejected.
1287        let outcome = resolve_now(
1288            &origin,
1289            HlsRequest::Playlist {
1290                track_id: DEFAULT_TRACK_ID,
1291                query: BlockingQuery {
1292                    hls_msn: Some(4),
1293                    hls_part: None,
1294                },
1295            },
1296        );
1297        assert!(
1298            matches!(outcome, EgressResponse::BadRequest { .. }),
1299            "msn at spec bound + 1 (last_closed+3) must be rejected"
1300        );
1301    }
1302
1303    #[test]
1304    fn part_without_msn_rejected() {
1305        let (_trunk, origin, _writer) = make_origin();
1306        let outcome = resolve_now(
1307            &origin,
1308            HlsRequest::Playlist {
1309                track_id: DEFAULT_TRACK_ID,
1310                query: BlockingQuery {
1311                    hls_msn: None,
1312                    hls_part: Some(0),
1313                },
1314            },
1315        );
1316        assert!(matches!(outcome, EgressResponse::BadRequest { .. }));
1317    }
1318
1319    #[test]
1320    fn resolve_resource_init_present() {
1321        let (_trunk, origin, _writer) = make_origin();
1322        match resolve_now(
1323            &origin,
1324            HlsRequest::Resource {
1325                name: "init-1.mp4".to_string(),
1326            },
1327        ) {
1328            EgressResponse::Ready {
1329                body: HlsBody::Resource(bytes),
1330                cache,
1331            } => {
1332                assert_eq!(bytes, Bytes::from(vec![0xAAu8; 8]));
1333                assert_eq!(cache, CachePolicy::Immutable);
1334            }
1335            other => panic!("expected Ready, got {other:?}"),
1336        }
1337    }
1338
1339    #[test]
1340    fn resolve_resource_unmatched_filename_not_found() {
1341        let (_trunk, origin, _writer) = make_origin();
1342        assert_eq!(
1343            resolve_now(
1344                &origin,
1345                HlsRequest::Resource {
1346                    name: "not-a-thing.txt".to_string(),
1347                }
1348            ),
1349            EgressResponse::NotFound
1350        );
1351    }
1352
1353    // --- issue #873: Container x low-latency matrix -----------------------
1354    //
1355    // Every cell below asserts on rendered playlist text AND on request
1356    // resolution (fetch every URI the rendered text itself advertised, never
1357    // a hard-coded filename) -- "advertised == servable" is the entire
1358    // reason `HlsOrigin` exists, per the issue. `EXT-X-VERSION` is checked
1359    // against `broadcast_hls::MediaPlaylist::computed_version()` re-derived
1360    // from a round-trip parse of the rendered text -- never an independently
1361    // guessed integer literal, which is exactly the bug issue #871 removed.
1362
1363    fn make_origin_with(
1364        container: Container,
1365        low_latency_ms: Option<u32>,
1366    ) -> (Arc<Trunk>, HlsOrigin, media_plane::trunk::SegmentWriter) {
1367        let trunk = Trunk::new(TrunkConfig::new(nz(64), nz(8), nz(8), nz(8), nz(64)));
1368        let writer = trunk.segment_writer().expect("first segment writer");
1369        let mut builder = HlsOrigin::builder(Arc::clone(&trunk))
1370            .target_duration_secs(4.0)
1371            .window_segments(nz(4))
1372            .container(container);
1373        if let Some(ms) = low_latency_ms {
1374            builder = builder.low_latency(ms);
1375        }
1376        let origin = builder.build().expect("both required fields set");
1377        (trunk, origin, writer)
1378    }
1379
1380    fn render_body(origin: &HlsOrigin) -> String {
1381        match resolve_now(
1382            origin,
1383            HlsRequest::Playlist {
1384                track_id: DEFAULT_TRACK_ID,
1385                query: BlockingQuery::default(),
1386            },
1387        ) {
1388            EgressResponse::Ready {
1389                body: HlsBody::Playlist(m),
1390                ..
1391            } => m,
1392            other => panic!("expected Ready(Playlist), got {other:?}"),
1393        }
1394    }
1395
1396    /// The `#EXT-X-VERSION:<n>` value present in `body`, if any.
1397    fn extract_version_tag(body: &str) -> Option<u8> {
1398        body.lines()
1399            .find_map(|l| l.strip_prefix("#EXT-X-VERSION:")?.parse::<u8>().ok())
1400    }
1401
1402    /// Every segment URI (the line immediately after a `#EXTINF:` line), in
1403    /// playlist order -- exactly what a real client would fetch next.
1404    fn segment_uris(body: &str) -> Vec<String> {
1405        let lines: Vec<&str> = body.lines().collect();
1406        let mut out = Vec::new();
1407        for i in 0..lines.len() {
1408            if lines[i].starts_with("#EXTINF:") {
1409                if let Some(next) = lines.get(i + 1) {
1410                    if !next.starts_with('#') {
1411                        out.push((*next).to_string());
1412                    }
1413                }
1414            }
1415        }
1416        out
1417    }
1418
1419    /// Every `#EXT-X-PART:` line's `URI="..."` value, in order.
1420    fn part_uris(body: &str) -> Vec<String> {
1421        body.lines()
1422            .filter(|l| l.starts_with("#EXT-X-PART:"))
1423            .filter_map(|l| {
1424                let start = l.find("URI=\"")? + "URI=\"".len();
1425                let rest = &l[start..];
1426                let end = rest.find('"')?;
1427                Some(rest[..end].to_string())
1428            })
1429            .collect()
1430    }
1431
1432    /// Assert the rendered `#EXT-X-VERSION` equals `broadcast_hls`'s own
1433    /// `computed_version()`, re-derived by round-trip parsing the rendered
1434    /// text -- never an integer this test independently guessed. Returns the
1435    /// rendered value so a caller can make a further, non-vacuous claim
1436    /// about it.
1437    fn assert_version_matches_broadcast_hls_derivation(body: &str) -> Option<u8> {
1438        let parsed = MediaPlaylist::parse(body).expect("rendered body must round-trip parse");
1439        let rendered = extract_version_tag(body);
1440        assert_eq!(
1441            rendered,
1442            parsed.computed_version(),
1443            "rendered #EXT-X-VERSION must equal broadcast_hls's own derivation, body: {body}"
1444        );
1445        rendered
1446    }
1447
1448    /// [`assert_version_matches_broadcast_hls_derivation`] **plus a
1449    /// non-vacuity guard**: the playlist must actually trigger at least one
1450    /// RFC 8216bis §8 version rule, so the equality above cannot pass by
1451    /// comparing `None` against `None`.
1452    ///
1453    /// Without this, a cell whose every `#EXTINF` is integral (e.g. a
1454    /// duration of exactly `4.0`, which renders `#EXTINF:4,`) trips no §8
1455    /// row at all, and the derivation could be entirely broken while the
1456    /// test stayed green. Real segmenters cut on keyframes, not whole
1457    /// seconds, so a fractional `#EXTINF` is also the realistic shape --
1458    /// cf. `fixtures/hls/spec/9.1-simple-media-playlist.m3u8` (`#EXTINF:9.009`,
1459    /// `#EXT-X-VERSION:3`).
1460    fn assert_version_present_and_matches_derivation(body: &str) -> u8 {
1461        assert_version_matches_broadcast_hls_derivation(body).unwrap_or_else(|| {
1462            panic!(
1463                "this cell must trigger a real RFC 8216bis §8 version rule -- a \
1464                 missing #EXT-X-VERSION makes the derivation check vacuous, body: {body}"
1465            )
1466        })
1467    }
1468
1469    /// MUTATION VERIFIED (issue #873): making `HlsOriginBuilder::container`
1470    /// a no-op (so every origin renders as `Fmp4` regardless of the
1471    /// `Container` passed to it) makes this test's
1472    /// `assert!(!body.contains("#EXT-X-MAP"))` fail -- the mutated build
1473    /// unconditionally emits `#EXT-X-MAP:URI="init-1.mp4"`, and the `.ts`
1474    /// URI assertions fail too (segments render as `seg-1-1.m4s` instead of
1475    /// `seg-1-1.ts`). Recompiled and re-run to confirm the failure (see the
1476    /// PR description for the pasted `cargo test` output), then reverted.
1477    /// The mutation bites four tests in total -- this one, the low-latency
1478    /// `MpegTs` cell, the integral-`EXTINF` version case, and the
1479    /// cross-container refusal.
1480    #[test]
1481    fn mpegts_classic_no_map_ts_uris_no_ll_tags() {
1482        let (_trunk, origin, writer) = make_origin_with(Container::MpegTs, None);
1483        // Fractional, like every real keyframe-cut segment (and like the
1484        // RFC's own classic examples) -- so RFC 8216bis §8 row 3
1485        // (floating-point EXTINF) genuinely fires and the version assertion
1486        // below is not `None == None`.
1487        seg(&writer, 1, 4.004, false);
1488
1489        let body = render_body(&origin);
1490        assert!(!body.contains("#EXT-X-MAP"), "body: {body}");
1491        assert!(!body.contains("#EXT-X-PART"), "body: {body}");
1492        assert!(!body.contains("#EXT-X-SERVER-CONTROL"), "body: {body}");
1493        assert!(!body.contains("#EXT-X-PRELOAD-HINT"), "body: {body}");
1494        assert!(!body.contains(".m4s"), "body: {body}");
1495        assert!(body.contains("seg-1-1.ts"), "body: {body}");
1496        assert_version_present_and_matches_derivation(&body);
1497
1498        // advertised == servable: fetch every URI the rendered text itself
1499        // named, and check its bytes against what was actually published
1500        // (via this crate's own `parse_immediate`, not a hard-coded filename).
1501        let uris = segment_uris(&body);
1502        assert_eq!(uris, vec!["seg-1-1.ts".to_string()]);
1503        for uri in uris {
1504            let ImmediateResource::Segment(seq) = parse_immediate(&uri, Container::MpegTs)
1505                .expect("advertised segment URI must parse under MpegTs")
1506            else {
1507                panic!("expected a Segment resource for {uri}");
1508            };
1509            match resolve_now(&origin, HlsRequest::Resource { name: uri.clone() }) {
1510                EgressResponse::Ready {
1511                    body: HlsBody::Resource(bytes),
1512                    ..
1513                } => assert_eq!(bytes, Bytes::from(vec![seq as u8; 8])),
1514                other => panic!("expected Ready for {uri}, got {other:?}"),
1515            }
1516        }
1517    }
1518
1519    /// MUTATION VERIFIED (issue #873): making `HlsOriginBuilder::container`
1520    /// a no-op makes this test's `assert!(!body.contains("#EXT-X-MAP"))`
1521    /// fail identically to the classic-`MpegTs` test above, and the part
1522    /// URIs render as `part-1-2.0.m4s` instead of `part-1-2.0.ts`, so
1523    /// `assert!(body.contains("part-1-2.0.ts"))` also fails. Recompiled and
1524    /// re-run to confirm, then reverted.
1525    #[test]
1526    fn mpegts_low_latency_part_ts_uris_blocking_part_requests_resolve() {
1527        let (trunk, origin, writer) = make_origin_with(Container::MpegTs, Some(500));
1528        let origin = Arc::new(origin);
1529        // A closed segment with a fractional (keyframe-cut) duration, so
1530        // RFC 8216bis §8 row 3 fires and the version assertion below is not
1531        // `None == None`; segment 2 is then the open one carrying live parts.
1532        seg(&writer, 1, 4.004, false);
1533        part(&writer, 2, 0, true);
1534        part(&writer, 2, 1, false);
1535
1536        let body = render_body(&origin);
1537        assert!(!body.contains("#EXT-X-MAP"), "body: {body}");
1538        assert!(body.contains("#EXT-X-PART-INF"), "body: {body}");
1539        assert!(body.contains("#EXT-X-PART:"), "body: {body}");
1540        assert!(body.contains("seg-1-1.ts"), "body: {body}");
1541        assert!(body.contains("part-1-2.0.ts"), "body: {body}");
1542        assert!(!body.contains(".m4s"), "body: {body}");
1543        // LL-HLS directives add no §8 version requirement of their own --
1544        // the finding that killed the old hardcoded `EXT-X-VERSION:9` --
1545        // so this must derive to exactly the same value as the classic
1546        // MpegTs cell above.
1547        assert_version_present_and_matches_derivation(&body);
1548
1549        // advertised == servable for every advertised part.
1550        let parts = part_uris(&body);
1551        assert!(!parts.is_empty(), "body: {body}");
1552        for uri in &parts {
1553            let (seq, idx) = parse_part(uri, Container::MpegTs)
1554                .unwrap_or_else(|| panic!("advertised part URI {uri} must parse under MpegTs"));
1555            match resolve_now(&origin, HlsRequest::Resource { name: uri.clone() }) {
1556                EgressResponse::Ready {
1557                    body: HlsBody::Resource(bytes),
1558                    ..
1559                } => {
1560                    assert_eq!(bytes, Bytes::from(vec![idx as u8; 4]));
1561                    assert_eq!(seq, 2);
1562                }
1563                other => panic!("expected Ready for {uri}, got {other:?}"),
1564            }
1565        }
1566
1567        // A blocking request for a `.ts` part not yet produced must Await,
1568        // then resolve once produced -- the same guarantee the pre-#873
1569        // fMP4-only test proves, now exercised over the `.ts` URI scheme.
1570        let deadline = Timestamp::from_nanos(5_000_000_000);
1571        let policy = AwaitPolicy::new(deadline);
1572        let pending = origin.resolve(
1573            HlsRequest::Resource {
1574                name: "part-1-2.2.ts".to_string(),
1575            },
1576            Timestamp::from_nanos(0),
1577            policy,
1578        );
1579        assert!(
1580            matches!(pending, EgressResponse::Await { .. }),
1581            "expected Await before the part exists, got {pending:?}"
1582        );
1583
1584        let listener = trunk.listen().expect("listener slot available");
1585        let woken = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false));
1586        let woken2 = std::sync::Arc::clone(&woken);
1587        let waiter = std::thread::spawn(move || {
1588            let ok = listener.wait_deadline(Instant::now() + Duration::from_secs(60));
1589            woken2.store(ok, std::sync::atomic::Ordering::SeqCst);
1590        });
1591        part(&writer, 2, 2, false);
1592        waiter.join().expect("waiter thread must not panic");
1593        assert!(
1594            woken.load(std::sync::atomic::Ordering::SeqCst),
1595            "Trunk::listen() must wake once publish_part lands"
1596        );
1597
1598        match origin.resolve(
1599            HlsRequest::Resource {
1600                name: "part-1-2.2.ts".to_string(),
1601            },
1602            Timestamp::from_nanos(1),
1603            policy,
1604        ) {
1605            EgressResponse::Ready {
1606                body: HlsBody::Resource(bytes),
1607                ..
1608            } => assert_eq!(bytes, Bytes::from(vec![2u8; 4])),
1609            other => panic!("expected Ready once produced, got {other:?}"),
1610        }
1611    }
1612
1613    /// RFC 8216bis §8's opening rule: a playlist that triggers no version
1614    /// row at all is version-1 compatible and need not carry the tag. Kept
1615    /// as its own case (rather than as the `MpegTs` cells' only version
1616    /// check, which made those assertions vacuous) because an integral
1617    /// `#EXTINF` is a real, if unusual, shape.
1618    ///
1619    /// MUTATION VERIFIED (issue #873): making `HlsOriginBuilder::container`
1620    /// a no-op makes this fail with `left: Some(6), right: None` -- the
1621    /// mutated build emits `#EXT-X-MAP`, which trips §8 row 6. Recompiled
1622    /// and re-run to confirm, then reverted.
1623    #[test]
1624    fn mpegts_classic_integral_extinf_emits_no_version_tag() {
1625        let (_trunk, origin, writer) = make_origin_with(Container::MpegTs, None);
1626        seg(&writer, 1, 4.0, false);
1627        let body = render_body(&origin);
1628        // A whole number of seconds renders as an integer, so the playlist
1629        // genuinely contains no floating-point EXTINF value -- §8 row 3
1630        // does not fire, and omitting `EXT-X-VERSION` is honest rather than
1631        // a lie to a v1/v2 client. (`broadcast-hls` used to render `4.000`
1632        // here while still reporting no version requirement; fixed in the
1633        // same PR as this test.)
1634        assert!(body.contains("#EXTINF:4,"), "body: {body}");
1635        assert!(!body.contains("4.000"), "body: {body}");
1636        assert_eq!(
1637            assert_version_matches_broadcast_hls_derivation(&body),
1638            None,
1639            "nothing in this playlist triggers an RFC 8216bis §8 row: {body}"
1640        );
1641    }
1642
1643    /// LL-HLS directives carry no RFC 8216bis §8 version requirement of
1644    /// their own -- the finding that killed the old hardcoded
1645    /// `EXT-X-VERSION:9`. Enabling low latency must therefore not change
1646    /// the derived version for otherwise-identical content.
1647    #[test]
1648    fn low_latency_does_not_raise_the_derived_version() {
1649        let (_t1, classic, w1) = make_origin_with(Container::MpegTs, None);
1650        seg(&w1, 1, 4.004, false);
1651        let classic_version = assert_version_present_and_matches_derivation(&render_body(&classic));
1652
1653        let (_t2, low_latency, w2) = make_origin_with(Container::MpegTs, Some(500));
1654        seg(&w2, 1, 4.004, false);
1655        part(&w2, 2, 0, true);
1656        let ll_body = render_body(&low_latency);
1657        assert!(ll_body.contains("#EXT-X-PART:"), "body: {ll_body}");
1658        assert_eq!(
1659            assert_version_present_and_matches_derivation(&ll_body),
1660            classic_version,
1661            "enabling low latency must not raise the derived version"
1662        );
1663    }
1664
1665    #[test]
1666    fn fmp4_classic_map_present_no_ll_tags() {
1667        let (_trunk, origin, writer) = make_origin_with(Container::Fmp4, None);
1668        origin.set_init(vec![0xBBu8; 8]);
1669        // Fractional, so §8 row 3 fires alongside row 6 (EXT-X-MAP without
1670        // EXT-X-I-FRAMES-ONLY). `max(6, 3) = 6`, so the rendered value is
1671        // unchanged -- this removes the ambiguity of an integral EXTINF
1672        // leaving row 3 entirely untested here.
1673        seg(&writer, 1, 4.004, false);
1674
1675        let body = render_body(&origin);
1676        assert!(
1677            body.contains("#EXT-X-MAP:URI=\"init-1.mp4\""),
1678            "body: {body}"
1679        );
1680        assert!(!body.contains("#EXT-X-PART"), "body: {body}");
1681        assert!(!body.contains("#EXT-X-SERVER-CONTROL"), "body: {body}");
1682        assert!(!body.contains("#EXT-X-PRELOAD-HINT"), "body: {body}");
1683        assert!(body.contains("seg-1-1.m4s"), "body: {body}");
1684        assert_version_present_and_matches_derivation(&body);
1685
1686        // advertised == servable, including the init segment the MAP names.
1687        match resolve_now(
1688            &origin,
1689            HlsRequest::Resource {
1690                name: "init-1.mp4".to_string(),
1691            },
1692        ) {
1693            EgressResponse::Ready {
1694                body: HlsBody::Resource(bytes),
1695                ..
1696            } => assert_eq!(bytes, Bytes::from(vec![0xBBu8; 8])),
1697            other => panic!("expected Ready(init), got {other:?}"),
1698        }
1699        for uri in segment_uris(&body) {
1700            let ImmediateResource::Segment(seq) = parse_immediate(&uri, Container::Fmp4)
1701                .expect("advertised segment URI must parse under Fmp4")
1702            else {
1703                panic!("expected a Segment resource for {uri}");
1704            };
1705            match resolve_now(&origin, HlsRequest::Resource { name: uri.clone() }) {
1706                EgressResponse::Ready {
1707                    body: HlsBody::Resource(bytes),
1708                    ..
1709                } => assert_eq!(bytes, Bytes::from(vec![seq as u8; 8])),
1710                other => panic!("expected Ready for {uri}, got {other:?}"),
1711            }
1712        }
1713    }
1714
1715    /// Regression guard (issue #873, matrix cell 4): the pre-#873 default
1716    /// shape (`Fmp4` + low-latency) must render unchanged.
1717    #[test]
1718    fn fmp4_low_latency_existing_behaviour_preserved() {
1719        let (_trunk, origin, writer) = make_origin_with(Container::Fmp4, Some(500));
1720        origin.set_init(vec![0xAAu8; 8]);
1721        // Fractional for the same reason as the Fmp4-classic cell above.
1722        seg(&writer, 1, 4.004, false);
1723        part(&writer, 2, 0, true);
1724        part(&writer, 2, 1, false);
1725
1726        let body = render_body(&origin);
1727        assert!(
1728            body.contains("#EXT-X-MAP:URI=\"init-1.mp4\""),
1729            "body: {body}"
1730        );
1731        assert!(body.contains("#EXT-X-PART-INF"), "body: {body}");
1732        assert!(body.contains("#EXT-X-SERVER-CONTROL"), "body: {body}");
1733        assert!(body.contains("#EXT-X-PRELOAD-HINT"), "body: {body}");
1734        assert!(body.contains("seg-1-1.m4s"), "body: {body}");
1735        assert!(body.contains("part-1-2.0.m4s"), "body: {body}");
1736        assert!(!body.contains(".ts\""), "body: {body}");
1737        assert_version_present_and_matches_derivation(&body);
1738
1739        match resolve_now(
1740            &origin,
1741            HlsRequest::Resource {
1742                name: "init-1.mp4".to_string(),
1743            },
1744        ) {
1745            EgressResponse::Ready {
1746                body: HlsBody::Resource(bytes),
1747                ..
1748            } => assert_eq!(bytes, Bytes::from(vec![0xAAu8; 8])),
1749            other => panic!("expected Ready(init), got {other:?}"),
1750        }
1751        for uri in segment_uris(&body) {
1752            let ImmediateResource::Segment(seq) = parse_immediate(&uri, Container::Fmp4)
1753                .expect("advertised segment URI must parse under Fmp4")
1754            else {
1755                panic!("expected a Segment resource for {uri}");
1756            };
1757            match resolve_now(&origin, HlsRequest::Resource { name: uri.clone() }) {
1758                EgressResponse::Ready {
1759                    body: HlsBody::Resource(bytes),
1760                    ..
1761                } => assert_eq!(bytes, Bytes::from(vec![seq as u8; 8])),
1762                other => panic!("expected Ready for {uri}, got {other:?}"),
1763            }
1764        }
1765        for uri in part_uris(&body) {
1766            let (_seq, idx) = parse_part(&uri, Container::Fmp4)
1767                .unwrap_or_else(|| panic!("advertised part URI {uri} must parse under Fmp4"));
1768            match resolve_now(&origin, HlsRequest::Resource { name: uri.clone() }) {
1769                EgressResponse::Ready {
1770                    body: HlsBody::Resource(bytes),
1771                    ..
1772                } => assert_eq!(bytes, Bytes::from(vec![idx as u8; 4])),
1773                other => panic!("expected Ready for {uri}, got {other:?}"),
1774            }
1775        }
1776    }
1777
1778    /// MUTATION VERIFIED (issue #873): making `HlsOriginBuilder::container`
1779    /// a no-op makes every origin behave as `Fmp4`, so `init-1.mp4` under a
1780    /// nominally-`MpegTs` origin resolves `Ready` instead of `NotFound` --
1781    /// this test's final assertion fails. Recompiled and re-run to confirm,
1782    /// then reverted.
1783    #[test]
1784    fn cross_container_refusal_mp4_init_under_mpegts_not_found() {
1785        let (_trunk, origin, _writer) = make_origin_with(Container::MpegTs, None);
1786        // Still callable (documented no-op) -- the bytes are simply never
1787        // advertised or served under `MpegTs`.
1788        origin.set_init(vec![0xCCu8; 8]);
1789        assert_eq!(
1790            resolve_now(
1791                &origin,
1792                HlsRequest::Resource {
1793                    name: "init-1.mp4".to_string(),
1794                }
1795            ),
1796            EgressResponse::NotFound
1797        );
1798    }
1799
1800    #[test]
1801    fn builder_errors_on_missing_required_fields() {
1802        let trunk = Trunk::new(TrunkConfig::new(nz(64), nz(8), nz(8), nz(8), nz(64)));
1803        match HlsOrigin::builder(Arc::clone(&trunk))
1804            .window_segments(nz(4))
1805            .build()
1806        {
1807            Err(e) => assert_eq!(e, HlsOriginBuildError::MissingTargetDurationSecs),
1808            Ok(_) => panic!("expected an error: target_duration_secs was never set"),
1809        }
1810        match HlsOrigin::builder(Arc::clone(&trunk))
1811            .target_duration_secs(4.0)
1812            .build()
1813        {
1814            Err(e) => assert_eq!(e, HlsOriginBuildError::MissingWindowSegments),
1815            Ok(_) => panic!("expected an error: window_segments was never set"),
1816        }
1817    }
1818
1819    #[test]
1820    fn container_label_and_display() {
1821        assert_eq!(Container::Fmp4.name(), "fmp4");
1822        assert_eq!(Container::MpegTs.name(), "mpeg-ts");
1823        assert_eq!(Container::Fmp4.to_string(), "fmp4");
1824        assert_eq!(Container::default(), Container::Fmp4);
1825    }
1826}