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