use super::ids::Generation;
use media_plane::ids::{PartIndex, SegmentSeq};
use std::collections::{BTreeMap, VecDeque};
use std::num::NonZeroUsize;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, Mutex, MutexGuard, PoisonError};
use std::time::Duration;
use transmux::TrackId;
use broadcast_common::{Timestamp, Unpackage};
use broadcast_hls::{
DecimalSeconds, LowLatencyConfig, MapTag, MasterPlaylist, MediaPlaylist, MediaSegment,
OpenSegment, PartSpec, Variant,
};
use bytes::Bytes;
use media_plane::egress::{AwaitPolicy, CachePolicy, EgressResponse, ServedEgress};
use media_plane::trunk::{PartEntry, SegmentCursor, SegmentCursorItem, SegmentEntry, Trunk};
const UNDRAINED: u64 = u64::MAX;
static LAST_INSTANCE: AtomicU64 = AtomicU64::new(0);
fn next_instance_token() -> u64 {
let now_ms = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.ok()
.and_then(|d| u64::try_from(d.as_millis()).ok())
.unwrap_or(0);
let mut previous = LAST_INSTANCE.load(Ordering::Relaxed);
loop {
let token = now_ms.max(previous.saturating_add(1));
match LAST_INSTANCE.compare_exchange_weak(
previous,
token,
Ordering::AcqRel,
Ordering::Relaxed,
) {
Ok(_) => return token,
Err(actual) => previous = actual,
}
}
}
fn locked<T>(m: &Mutex<T>) -> MutexGuard<'_, T> {
m.lock().unwrap_or_else(PoisonError::into_inner)
}
pub const DEFAULT_TRACK_ID: TrackId = TrackId::new(1);
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[non_exhaustive]
pub enum Container {
Fmp4,
MpegTs,
}
impl Container {
pub fn name(&self) -> &'static str {
match self {
Container::Fmp4 => "fmp4",
Container::MpegTs => "mpeg-ts",
}
}
fn segment_extension(self) -> &'static str {
match self {
Container::Fmp4 => "m4s",
Container::MpegTs => "ts",
}
}
}
broadcast_common::impl_spec_display!(Container);
impl Default for Container {
fn default() -> Self {
Container::Fmp4
}
}
const PLACEHOLDER_BANDWIDTH_BPS: u64 = 5_000_000;
const INIT_HISTORY: usize = 8;
const NANOS_PER_SEC: u128 = 1_000_000_000;
const MIN_WINDOW_SEGMENTS: usize = 3;
const PART_RETENTION_TARGET_DURATIONS: f64 = 3.0;
const FUTURE_PART_SEGMENT_BOUND: u32 = 2;
const ADVANCE_PART_LIMIT_NUMERATOR: u64 = 3;
const MILLIS_PER_SEC: u64 = 1_000;
const ABUSE_MSN_FUTURE_BOUND: u64 = 2;
const PART_HOLD_BACK_MULTIPLIER: f64 = 3.0;
pub fn master_playlist_m3u8(media_playlist_name: &str) -> String {
format!(
"#EXTM3U\n#EXT-X-STREAM-INF:BANDWIDTH={PLACEHOLDER_BANDWIDTH_BPS}\n{media_playlist_name}\n"
)
}
#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)]
pub struct BlockingQuery {
pub hls_msn: Option<u64>,
pub hls_part: Option<u32>,
}
#[derive(Debug, Clone, PartialEq, Eq)]
#[non_exhaustive]
pub enum HlsRequest {
Playlist {
track_id: TrackId,
query: BlockingQuery,
},
Resource {
name: String,
},
}
#[derive(Debug, Clone, PartialEq, Eq)]
#[non_exhaustive]
pub enum HlsBody {
Playlist(String),
Resource(Bytes),
}
#[derive(Clone)]
struct WindowSegment {
sequence_number: SegmentSeq,
init_gen: Generation,
bytes: Bytes,
duration_secs: f64,
discontinuous: bool,
start_ns: u64,
}
#[derive(Debug, Clone, Copy, PartialEq)]
#[non_exhaustive]
pub struct ClosedSegment {
pub sequence_number: SegmentSeq,
pub start_ns: u64,
pub duration_secs: f64,
pub discontinuous: bool,
pub init_generation: Generation,
}
impl ClosedSegment {
#[must_use]
pub fn with_init_generation(mut self, generation: Generation) -> Self {
self.init_generation = generation;
self
}
pub fn new(
sequence_number: SegmentSeq,
start_ns: u64,
duration_secs: f64,
discontinuous: bool,
) -> Self {
ClosedSegment {
sequence_number,
start_ns,
duration_secs,
discontinuous,
init_generation: Generation::new(1),
}
}
}
struct Window {
segments: VecDeque<WindowSegment>,
capacity: usize,
max_segment_duration_secs: f64,
peak_bandwidth_bps: u64,
discontinuity_sequence: u64,
gap_pending: bool,
}
impl Window {
fn new(capacity: NonZeroUsize) -> Self {
Window {
segments: VecDeque::new(),
capacity: capacity.get().max(MIN_WINDOW_SEGMENTS),
max_segment_duration_secs: 0.0,
peak_bandwidth_bps: 0,
discontinuity_sequence: 0,
gap_pending: false,
}
}
fn note_gap(&mut self) {
self.gap_pending = true;
}
fn clear_rolling_off(&mut self) {
let discontinuities = self.segments.iter().filter(|s| s.discontinuous).count();
self.discontinuity_sequence = self
.discontinuity_sequence
.saturating_add(u64::try_from(discontinuities).unwrap_or(u64::MAX));
self.segments.clear();
}
fn push(&mut self, entry: SegmentEntry, init_gen: Generation) {
let follows = self
.segments
.back()
.is_none_or(|last| last.sequence_number.checked_next() == Some(entry.sequence_number));
let mut discontinuous = entry.meta.discontinuous;
if !follows || self.gap_pending {
self.clear_rolling_off();
discontinuous = true;
}
self.gap_pending = false;
if self
.segments
.back()
.is_some_and(|last| last.init_gen != init_gen)
{
discontinuous = true;
}
let duration_secs = entry.duration.as_secs_f64();
self.max_segment_duration_secs = self.max_segment_duration_secs.max(duration_secs);
if let Some(bps) = segment_bandwidth_bps(entry.bytes.len(), entry.duration) {
self.peak_bandwidth_bps = self.peak_bandwidth_bps.max(bps);
}
if self.segments.len() >= self.capacity
&& let Some(evicted) = self.segments.pop_front()
&& evicted.discontinuous
{
self.discontinuity_sequence = self.discontinuity_sequence.saturating_add(1);
}
self.segments.push_back(WindowSegment {
sequence_number: entry.sequence_number,
init_gen,
bytes: entry.bytes,
duration_secs,
discontinuous,
start_ns: entry.timeline_position.as_nanos(),
});
}
fn bytes_of(&self, sequence_number: SegmentSeq) -> Option<Bytes> {
let front = self.segments.front()?.sequence_number;
let index = usize::try_from(sequence_number.checked_distance_from(front)?).ok()?;
let segment = self.segments.get(index)?;
(segment.sequence_number == sequence_number).then(|| segment.bytes.clone())
}
}
struct WindowView {
segments: VecDeque<WindowSegment>,
max_segment_duration_secs: f64,
discontinuity_sequence: u64,
}
#[derive(Default)]
struct InitStore {
generation: Generation,
current: Option<Bytes>,
history: VecDeque<(Generation, Bytes)>,
starts: VecDeque<(SegmentSeq, Generation)>,
}
impl InitStore {
fn set(&mut self, bytes: Bytes, next_seq: SegmentSeq) {
if self.current.as_ref() == Some(&bytes) {
return;
}
self.generation = Generation::new(self.generation.value().saturating_add(1));
let first_seq = if self.current.is_none() {
SegmentSeq::new(0)
} else {
next_seq
};
self.current = Some(bytes.clone());
self.history.push_back((self.generation, bytes));
self.starts.push_back((first_seq, self.generation));
while self.history.len() > INIT_HISTORY {
self.history.pop_front();
}
while self.starts.len() > INIT_HISTORY {
self.starts.pop_front();
}
}
fn generation_for(&self, seq: SegmentSeq) -> Generation {
self.starts
.iter()
.rev()
.find(|(first, _)| *first <= seq)
.or_else(|| self.starts.front())
.map_or(Generation::new(1), |&(_, generation)| generation)
}
fn bytes_of(&self, generation: Generation) -> Option<Bytes> {
self.history
.iter()
.find(|(g, _)| *g == generation)
.map(|(_, b)| b.clone())
}
}
fn segment_bandwidth_bps(len: usize, duration: Duration) -> Option<u64> {
let nanos = duration.as_nanos();
if nanos == 0 {
return None;
}
let bits = u128::try_from(len).ok()?.checked_mul(8)?;
let bps = bits.checked_mul(NANOS_PER_SEC)?.div_ceil(nanos);
u64::try_from(bps).ok()
}
fn parse_part(file: &str, container: Container) -> Option<(Option<u64>, u64, PartIndex)> {
let suffix = format!(".{}", container.segment_extension());
let rest = file.strip_prefix("part-")?.strip_suffix(suffix.as_str())?;
let (names, idx) = rest.rsplit_once('.')?;
let (epoch, seq) = parse_epoch_seq(names)?;
Some((epoch, seq, PartIndex::new(idx.parse().ok()?)))
}
fn parse_epoch_seq(names: &str) -> Option<(Option<u64>, u64)> {
let mut fields = names.split('-');
fields.next()?.parse::<u32>().ok()?;
let first = fields.next()?;
match fields.next() {
None => Some((None, first.parse().ok()?)),
Some(seq) => {
if fields.next().is_some() {
return None;
}
Some((Some(first.parse().ok()?), seq.parse().ok()?))
}
}
}
enum ImmediateResource {
CurrentInit,
VersionedInit { epoch: u64, generation: Generation },
Segment { epoch: Option<u64>, seq: u64 },
}
fn parse_immediate(file: &str, container: Container) -> Option<ImmediateResource> {
if container == Container::Fmp4
&& let Some(rest) = file.strip_prefix("init-")
{
let rest = rest.strip_suffix(".mp4")?;
let mut fields = rest.split('-');
fields.next()?.parse::<u32>().ok()?;
return match (fields.next(), fields.next(), fields.next()) {
(None, ..) => Some(ImmediateResource::CurrentInit),
(Some(epoch), Some(generation), None) => Some(ImmediateResource::VersionedInit {
epoch: epoch.parse().ok()?,
generation: Generation::new(generation.parse().ok()?),
}),
_ => None,
};
}
if let Some(rest) = file.strip_prefix("seg-") {
let suffix = format!(".{}", container.segment_extension());
let (epoch, seq) = parse_epoch_seq(rest.strip_suffix(suffix.as_str())?)?;
return Some(ImmediateResource::Segment { epoch, seq });
}
None
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, thiserror::Error)]
#[non_exhaustive]
pub enum HlsOriginBuildError {
#[error("HlsOrigin::builder(...).target_duration_secs(...) is required but was never called")]
MissingTargetDurationSecs,
#[error("HlsOrigin::builder(...).window_segments(...) is required but was never called")]
MissingWindowSegments,
#[error("HlsOrigin::builder(...).target_duration_secs(...) must be finite and positive")]
InvalidTargetDuration,
#[error("HlsOrigin::builder(...).low_latency(0): the part target must be positive")]
ZeroPartTarget,
#[error(
"HlsOrigin::builder(...).media_sequence_offset(...) leaves no room for u32 sequence numbers"
)]
MediaSequenceOffsetTooLarge,
}
#[derive(Debug, PartialEq, thiserror::Error)]
#[non_exhaustive]
pub enum HlsMasterError {
#[error("measured BANDWIDTH {bandwidth} exceeds u32")]
BandwidthOverflow {
bandwidth: u64,
},
#[error("master playlist render failed: {0}")]
Render(broadcast_hls::Error),
}
pub struct HlsOriginBuilder {
trunk: Arc<Trunk>,
target_duration_secs: Option<f64>,
window_segments: Option<NonZeroUsize>,
container: Container,
part_target_ms: Option<u32>,
media_sequence_offset: u64,
instance: Option<u64>,
}
impl HlsOriginBuilder {
fn new(trunk: Arc<Trunk>) -> Self {
HlsOriginBuilder {
trunk,
target_duration_secs: None,
window_segments: None,
container: Container::default(),
part_target_ms: None,
media_sequence_offset: 0,
instance: None,
}
}
pub fn target_duration_secs(mut self, target_duration_secs: f64) -> Self {
self.target_duration_secs = Some(target_duration_secs);
self
}
pub fn window_segments(mut self, window_segments: NonZeroUsize) -> Self {
self.window_segments = Some(window_segments);
self
}
pub fn container(mut self, container: Container) -> Self {
self.container = container;
self
}
pub fn low_latency(mut self, part_target_ms: u32) -> Self {
self.part_target_ms = Some(part_target_ms);
self
}
pub fn media_sequence_offset(mut self, offset: u64) -> Self {
self.media_sequence_offset = offset;
self
}
pub fn instance(mut self, token: u64) -> Self {
self.instance = Some(token);
self
}
pub fn build(self) -> Result<HlsOrigin, HlsOriginBuildError> {
let target_duration_secs = self
.target_duration_secs
.ok_or(HlsOriginBuildError::MissingTargetDurationSecs)?;
if !target_duration_secs.is_finite() || target_duration_secs <= 0.0 {
return Err(HlsOriginBuildError::InvalidTargetDuration);
}
let window_segments = self
.window_segments
.ok_or(HlsOriginBuildError::MissingWindowSegments)?;
if self.part_target_ms == Some(0) {
return Err(HlsOriginBuildError::ZeroPartTarget);
}
if self
.media_sequence_offset
.checked_add(u64::from(u32::MAX))
.is_none()
{
return Err(HlsOriginBuildError::MediaSequenceOffsetTooLarge);
}
let cursor = self.trunk.subscribe_segments();
Ok(HlsOrigin {
trunk: self.trunk,
cursor: Mutex::new(cursor),
drained_through: AtomicU64::new(UNDRAINED),
window: Mutex::new(Window::new(window_segments)),
init: Mutex::new(InitStore::default()),
codecs: Mutex::new(None),
target_duration_secs,
container: self.container,
part_target_ms: self.part_target_ms,
media_sequence_offset: self.media_sequence_offset,
instance: self.instance.unwrap_or_else(next_instance_token),
})
}
}
pub struct HlsOrigin {
trunk: Arc<Trunk>,
cursor: Mutex<SegmentCursor>,
drained_through: AtomicU64,
window: Mutex<Window>,
init: Mutex<InitStore>,
codecs: Mutex<Option<String>>,
target_duration_secs: f64,
container: Container,
part_target_ms: Option<u32>,
media_sequence_offset: u64,
instance: u64,
}
impl HlsOrigin {
pub fn builder(trunk: Arc<Trunk>) -> HlsOriginBuilder {
HlsOriginBuilder::new(trunk)
}
pub fn set_init(&self, bytes: impl Into<Bytes>) {
let bytes = bytes.into();
if self.container == Container::Fmp4 {
let derived = transmux::Fmp4Demux::new()
.unpackage(bytes.as_ref())
.ok()
.map(|media| {
media
.tracks
.into_iter()
.map(|t| t.spec)
.collect::<Vec<transmux::TrackSpec>>()
})
.and_then(|specs| self.set_track_specs(&specs).ok());
if derived.is_none() {
*locked(&self.codecs) = None;
}
}
let next_seq = self
.trunk
.last_closed_segment()
.unwrap_or(SegmentSeq::new(0))
.saturating_next();
locked(&self.init).set(bytes, next_seq);
}
pub fn set_track_specs(&self, specs: &[transmux::TrackSpec]) -> Result<(), transmux::Error> {
let mut ordered: Vec<&transmux::TrackSpec> = specs.iter().collect();
ordered.sort_by_key(|spec| spec.track_id);
let mut codecs: Vec<String> = Vec::new();
for spec in ordered {
match transmux::rfc6381_codec_string(&spec.config) {
Ok(codec) => codecs.push(codec),
Err(transmux::Error::UnsupportedCodec { .. }) => {}
Err(other) => return Err(other),
}
}
*locked(&self.codecs) = (!codecs.is_empty()).then(|| codecs.join(","));
Ok(())
}
pub fn master_playlist(&self, media_playlist_name: &str) -> Result<String, HlsMasterError> {
self.drain();
let measured = locked(&self.window).peak_bandwidth_bps;
let bandwidth = if measured == 0 {
PLACEHOLDER_BANDWIDTH_BPS
} else {
measured
};
let bandwidth = u32::try_from(bandwidth)
.map_err(|_| HlsMasterError::BandwidthOverflow { bandwidth })?;
let master = MasterPlaylist {
variants: vec![Variant {
bandwidth,
codecs: locked(&self.codecs).clone(),
uri: media_playlist_name.to_string(),
..Default::default()
}],
..Default::default()
};
master.to_m3u8().map_err(HlsMasterError::Render)
}
pub fn instance(&self) -> u64 {
self.instance
}
pub fn init_name(&self, track_id: TrackId, generation: Generation) -> String {
format!("init-{track_id}-{}-{generation}.mp4", self.instance)
}
fn segment_name(&self, track_id: TrackId, seq: SegmentSeq, ext: &str) -> String {
format!(
"seg-{track_id}-{}-{}.{ext}",
self.instance,
self.public_msn(seq)
)
}
fn part_name(&self, track_id: TrackId, seq: SegmentSeq, idx: PartIndex, ext: &str) -> String {
format!(
"part-{track_id}-{}-{}.{idx}.{ext}",
self.instance,
self.public_msn(seq)
)
}
pub fn media_sequence_offset(&self) -> u64 {
self.media_sequence_offset
}
pub fn next_media_sequence(&self) -> u64 {
self.public_msn(
self.trunk
.last_closed_segment()
.unwrap_or(SegmentSeq::new(0))
.saturating_next(),
)
}
fn public_msn(&self, seq: SegmentSeq) -> u64 {
self.media_sequence_offset
.saturating_add(u64::from(seq.value()))
}
fn internal_seq(&self, public: u64) -> Option<SegmentSeq> {
u32::try_from(public.checked_sub(self.media_sequence_offset)?)
.ok()
.map(SegmentSeq::new)
}
fn advance_part_limit(&self) -> u64 {
match self.part_target_ms {
None | Some(0) => 0,
Some(ms) if u64::from(ms) < MILLIS_PER_SEC => {
(ADVANCE_PART_LIMIT_NUMERATOR * MILLIS_PER_SEC).div_ceil(u64::from(ms))
}
Some(_) => ADVANCE_PART_LIMIT_NUMERATOR,
}
}
pub fn init_bytes(&self) -> Option<Bytes> {
locked(&self.init).current.clone()
}
fn drain(&self) {
let head = self
.trunk
.last_closed_segment()
.map_or(0, |n| u64::from(n.value()).saturating_add(1));
if self.drained_through.load(Ordering::Acquire) == head {
return;
}
let mut cursor = locked(&self.cursor);
let mut window = locked(&self.window);
while let Some(item) = cursor.poll() {
match item {
SegmentCursorItem::Segment(entry) => {
let init_gen = locked(&self.init).generation_for(entry.sequence_number);
window.push(entry, init_gen);
}
_ => window.note_gap(),
}
}
self.drained_through.store(head, Ordering::Release);
}
fn window_view(&self) -> WindowView {
let w = locked(&self.window);
WindowView {
segments: w.segments.clone(),
max_segment_duration_secs: w.max_segment_duration_secs,
discontinuity_sequence: w.discontinuity_sequence,
}
}
pub fn closed_segments(&self) -> Vec<ClosedSegment> {
self.drain();
locked(&self.window)
.segments
.iter()
.map(|s| ClosedSegment {
sequence_number: s.sequence_number,
start_ns: s.start_ns,
duration_secs: s.duration_secs,
discontinuous: s.discontinuous,
init_generation: s.init_gen,
})
.collect()
}
fn live_edge(&self) -> (SegmentSeq, Vec<PartEntry>) {
let last_closed = self
.trunk
.last_closed_segment()
.unwrap_or(SegmentSeq::new(0));
let candidate = last_closed.saturating_next();
let parts = self.trunk.parts_in_segment(candidate);
if parts.is_empty() {
(last_closed, Vec::new())
} else {
(candidate, parts)
}
}
fn resident_part_specs(&self, track_id: TrackId, seq: SegmentSeq, ext: &str) -> Vec<PartSpec> {
let by_index: BTreeMap<PartIndex, PartEntry> = self
.trunk
.parts_in_segment(seq)
.into_iter()
.map(|p| (p.part_index, p))
.collect();
let contiguous = by_index
.keys()
.copied()
.zip(0u32..)
.all(|(index, expected)| index.value() == expected);
if by_index.is_empty() || !contiguous {
return Vec::new();
}
by_index
.values()
.map(|p| PartSpec {
uri: self.part_name(track_id, p.segment_number, p.part_index, ext),
duration: DecimalSeconds::new(p.duration.as_secs_f64())
.expect("Duration::as_secs_f64 is finite, >= 0"),
independent: p.independent,
..Default::default()
})
.collect()
}
fn render_playlist(&self, track_id: TrackId) -> core::result::Result<String, String> {
self.drain();
let window = self.window_view();
let (open_seq, open_parts) = self.live_edge();
let low_latency_enabled = self.part_target_ms.is_some();
let has_open_parts = low_latency_enabled && !open_parts.is_empty();
let ext = self.container.segment_extension();
let media_sequence = window
.segments
.front()
.map(|s| self.public_msn(s.sequence_number))
.or_else(|| has_open_parts.then(|| self.public_msn(open_seq)))
.unwrap_or_else(|| self.public_msn(SegmentSeq::new(1)));
let target_duration = self
.target_duration_secs
.max(window.max_segment_duration_secs);
let target_duration = target_duration.round().clamp(0.0, f64::from(u32::MAX)) as u32;
let retention_secs = PART_RETENTION_TARGET_DURATIONS * f64::from(target_duration);
let mut behind_secs: f64 = if has_open_parts {
open_parts.iter().map(|p| p.duration.as_secs_f64()).sum()
} else {
0.0
};
let mut closed_parts: Vec<Vec<PartSpec>> = Vec::with_capacity(window.segments.len());
for s in window.segments.iter().rev() {
let keep = low_latency_enabled && behind_secs <= retention_secs;
closed_parts.push(if keep {
self.resident_part_specs(track_id, s.sequence_number, ext)
} else {
Vec::new()
});
behind_secs += s.duration_secs;
}
closed_parts.reverse();
let fmp4 = self.container == Container::Fmp4;
let map_for = |generation: Generation| {
fmp4.then(|| MapTag {
uri: self.init_name(track_id, generation),
byte_range: None,
extra_attrs: Vec::new(),
})
};
let segments: Vec<MediaSegment> = window
.segments
.iter()
.zip(closed_parts)
.map(|(s, parts)| MediaSegment {
map: map_for(s.init_gen),
uri: self.segment_name(track_id, s.sequence_number, ext),
duration: DecimalSeconds::new(s.duration_secs)
.expect("Duration::as_secs_f64 is finite, >= 0"),
discontinuous: s.discontinuous,
parts,
..Default::default()
})
.collect();
let open_gen = locked(&self.init).generation_for(open_seq);
let open_segment = has_open_parts.then(|| {
let open = OpenSegment::new(
open_parts
.iter()
.map(|p| PartSpec {
uri: self.part_name(track_id, p.segment_number, p.part_index, ext),
duration: DecimalSeconds::new(p.duration.as_secs_f64())
.expect("Duration::as_secs_f64 is finite, >= 0"),
independent: p.independent,
..Default::default()
})
.collect(),
);
match map_for(open_gen) {
Some(map) => open.with_map(map),
None => open,
}
});
let next_part_hint = has_open_parts.then(|| {
let next_idx = open_parts
.iter()
.map(|p| p.part_index)
.max()
.map(PartIndex::saturating_next)
.unwrap_or(PartIndex::new(0));
self.part_name(track_id, open_seq, next_idx, ext)
});
let mut extra_tags = Vec::new();
if fmp4 && segments.is_empty() && open_segment.is_none() {
let generation = locked(&self.init).generation.max(Generation::new(1));
extra_tags.push(format!(
"#EXT-X-MAP:URI=\"{}\"",
self.init_name(track_id, generation)
));
}
if let Some(anchor) = self.trunk.time_anchor() {
let timeline = timed_metadata::Timeline::with_anchor(anchor);
for seg in &window.segments {
for entry in self.trunk.events_in_segment(seg.sequence_number) {
if let Ok(dr) = timeline.to_daterange(&entry.event) {
if let Ok(tag) = dr.to_tag_line() {
extra_tags.push(tag);
}
}
}
}
}
let low_latency = low_latency_enabled.then(|| {
let part_target_ms = self
.part_target_ms
.expect("low_latency_enabled implies Some");
let part_target = DecimalSeconds::new(f64::from(part_target_ms) / 1000.0)
.expect("u32 / 1000.0 is finite, >= 0");
let part_hold_back = DecimalSeconds::new(part_target.get() * PART_HOLD_BACK_MULTIPLIER)
.expect("finite, >= 0 times a positive constant is finite, >= 0");
LowLatencyConfig {
part_target: Some(part_target),
part_hold_back: Some(part_hold_back),
preload_hint_part: next_part_hint,
..Default::default()
}
});
let playlist = MediaPlaylist {
target_duration,
media_sequence,
discontinuity_sequence: window.discontinuity_sequence,
segments,
open_segment,
endlist: false,
extra_tags,
low_latency,
iframes_only: false,
..Default::default()
};
playlist.to_m3u8().map_err(|error| error.to_string())
}
fn resolve_playlist(
&self,
track_id: TrackId,
query: BlockingQuery,
now: Timestamp,
await_policy: AwaitPolicy,
) -> EgressResponse<HlsBody> {
if query.hls_part.is_some() && query.hls_msn.is_none() {
return EgressResponse::BadRequest {
reason: "_HLS_part without _HLS_msn is meaningless",
};
}
if let Some(msn) = query.hls_msn {
let (in_progress_seg, live_parts) = self.live_edge();
let in_progress = self.public_msn(in_progress_seg);
if msn > in_progress.saturating_add(ABUSE_MSN_FUTURE_BOUND) {
return EgressResponse::BadRequest {
reason: "_HLS_msn unreasonably far beyond the live edge",
};
}
if let Some(part) = query.hls_part
&& msn >= in_progress
{
let existing = if msn == in_progress {
u64::try_from(live_parts.len()).unwrap_or(u64::MAX)
} else {
0
};
let limit = existing
.saturating_sub(1)
.saturating_add(self.advance_part_limit());
if u64::from(part) > limit {
return EgressResponse::BadRequest {
reason: "_HLS_part beyond the Advance Part Limit",
};
}
}
let satisfied = match query.hls_part {
Some(part) => {
in_progress > msn
|| (in_progress == msn
&& u64::try_from(live_parts.len()).unwrap_or(u64::MAX)
> u64::from(part))
}
None => {
self.public_msn(
self.trunk
.last_closed_segment()
.unwrap_or(SegmentSeq::new(0)),
) >= msn
}
};
if !satisfied {
return EgressResponse::pending(await_policy, now, now);
}
}
match self.render_playlist(track_id) {
Ok(playlist) => EgressResponse::Ready {
body: HlsBody::Playlist(playlist),
cache: CachePolicy::NoCache,
},
Err(_) => EgressResponse::NotFound,
}
}
fn resolve_resource(
&self,
name: &str,
now: Timestamp,
await_policy: AwaitPolicy,
) -> EgressResponse<HlsBody> {
if let Some((epoch, public_seq, idx)) = parse_part(name, self.container) {
if epoch.is_some_and(|e| e != self.instance) {
return EgressResponse::NotFound;
}
let Some(seq) = self.internal_seq(public_seq) else {
return EgressResponse::NotFound;
};
if let Some(bytes) = self.trunk.part_bytes(seq, idx) {
return EgressResponse::Ready {
body: HlsBody::Resource(bytes),
cache: if epoch.is_some() {
CachePolicy::Immutable
} else {
CachePolicy::NoCache
},
};
}
if self.part_target_ms.is_none() {
return EgressResponse::NotFound;
}
let never_will = self.trunk.last_closed_segment().is_some_and(|c| c >= seq);
if never_will {
return EgressResponse::NotFound;
}
let last_closed = self
.trunk
.last_closed_segment()
.unwrap_or(SegmentSeq::new(0));
if last_closed
.checked_add(FUTURE_PART_SEGMENT_BOUND)
.is_some_and(|limit| seq > limit)
{
return EgressResponse::NotFound;
}
let next_hinted = self
.trunk
.parts_in_segment(seq)
.iter()
.map(|p| p.part_index)
.max()
.map_or(PartIndex::new(0), PartIndex::saturating_next);
if idx != next_hinted {
return EgressResponse::NotFound;
}
return EgressResponse::pending(await_policy, now, now);
}
self.drain();
let (bytes, cache) = match parse_immediate(name, self.container) {
Some(ImmediateResource::CurrentInit) => (self.init_bytes(), CachePolicy::NoCache),
Some(ImmediateResource::VersionedInit { epoch, generation })
if epoch == self.instance =>
{
(
locked(&self.init).bytes_of(generation),
CachePolicy::Immutable,
)
}
Some(ImmediateResource::Segment { epoch, seq })
if epoch.is_none_or(|e| e == self.instance) =>
{
(
self.internal_seq(seq)
.and_then(|s| locked(&self.window).bytes_of(s)),
if epoch.is_some() {
CachePolicy::Immutable
} else {
CachePolicy::NoCache
},
)
}
_ => (None, CachePolicy::NoCache),
};
match bytes {
Some(bytes) => EgressResponse::Ready {
body: HlsBody::Resource(bytes),
cache,
},
None => EgressResponse::NotFound,
}
}
}
impl ServedEgress for HlsOrigin {
type Request = HlsRequest;
type Body = HlsBody;
fn resolve(
&self,
request: HlsRequest,
now: Timestamp,
await_policy: AwaitPolicy,
) -> EgressResponse<HlsBody> {
match request {
HlsRequest::Playlist { track_id, query } => {
self.resolve_playlist(track_id, query, now, await_policy)
}
HlsRequest::Resource { name } => self.resolve_resource(&name, now, await_policy),
}
}
}
#[cfg(test)]
mod tests {
use super::*;
use media_plane::trunk::{EventAnchor, TrunkConfig};
use std::time::{Duration, Instant};
use timed_metadata::{MediaTime, TimeAnchor};
use transmux::SegmentMeta;
fn nz(n: usize) -> NonZeroUsize {
NonZeroUsize::new(n).expect("test capacity must be non-zero")
}
fn make_origin() -> (Arc<Trunk>, HlsOrigin, media_plane::trunk::SegmentWriter) {
let trunk = Trunk::new(TrunkConfig::new(nz(64), nz(8), nz(8), nz(8), nz(64)));
let writer = trunk.segment_writer().expect("first segment writer");
let origin = HlsOrigin::builder(Arc::clone(&trunk))
.target_duration_secs(4.0)
.window_segments(nz(4))
.low_latency(500)
.instance(7)
.build()
.expect("both required fields set");
origin.set_init(vec![0xAAu8; 8]);
(trunk, origin, writer)
}
#[test]
fn resource_names_place_every_distinct_id_in_its_own_slot() {
let (_trunk, origin, _writer) = make_origin();
let track = TrackId::new(3);
assert_eq!(origin.instance(), 7);
assert_eq!(
origin.init_name(track, Generation::new(4)),
"init-3-7-4.mp4"
);
assert_eq!(
origin.segment_name(track, SegmentSeq::new(5), "m4s"),
"seg-3-7-5.m4s"
);
assert_eq!(
origin.part_name(track, SegmentSeq::new(5), PartIndex::new(2), "m4s"),
"part-3-7-5.2.m4s"
);
let trunk = Trunk::new(TrunkConfig::new(nz(64), nz(8), nz(8), nz(8), nz(64)));
let offset = HlsOrigin::builder(trunk)
.target_duration_secs(4.0)
.window_segments(nz(4))
.instance(7)
.media_sequence_offset(100)
.build()
.expect("both required fields set");
assert_eq!(
offset.segment_name(track, SegmentSeq::new(5), "m4s"),
"seg-3-7-105.m4s",
"the Media Sequence Number offset applies to the sequence slot only"
);
assert_eq!(
offset.part_name(track, SegmentSeq::new(5), PartIndex::new(2), "m4s"),
"part-3-7-105.2.m4s"
);
assert_eq!(
offset.init_name(track, Generation::new(4)),
"init-3-7-4.mp4",
"the offset never touches the generation"
);
}
fn seg(
writer: &media_plane::trunk::SegmentWriter,
seq: u32,
duration_secs: f64,
discontinuous: bool,
) {
writer
.publish_segment(SegmentEntry::new(
Bytes::from(vec![seq as u8; 8]),
SegmentSeq::new(seq),
Duration::from_secs_f64(duration_secs),
Timestamp::from_nanos(0),
SegmentMeta { discontinuous },
))
.unwrap();
}
fn part(writer: &media_plane::trunk::SegmentWriter, seg_no: u32, idx: u32, independent: bool) {
writer.publish_part(PartEntry::new(
Bytes::from(vec![idx as u8; 4]),
SegmentSeq::new(seg_no),
PartIndex::new(idx),
Duration::from_millis(500),
independent,
));
}
fn resolve_now(origin: &HlsOrigin, request: HlsRequest) -> EgressResponse<HlsBody> {
origin.resolve(
request,
Timestamp::from_nanos(0),
AwaitPolicy::new(Timestamp::from_nanos(0)),
)
}
#[test]
fn trunk_events_render_as_ext_x_daterange() {
let hex = "FC302100000000000000FFF01005000007D27FEF7F7E0020F580C0000000000088B9661D";
let splice_bytes: Vec<u8> = (0..hex.len())
.step_by(2)
.map(|i| u8::from_str_radix(&hex[i..i + 2], 16).unwrap())
.collect();
let (trunk, origin, writer) = make_origin();
writer.set_time_anchor(TimeAnchor {
pts_90k: 0,
utc_epoch_ms: 1_000_000_000_000,
});
let mut timeline = timed_metadata::Timeline::new();
let ev = timeline.push_scte35(&splice_bytes).unwrap();
let trunk_writer = trunk.writer().expect("first trunk writer");
trunk_writer.publish_event(
ev.clone(),
EventAnchor::Media(ev.at.unwrap_or(MediaTime(0))),
);
writer.note_segment_start(SegmentSeq::new(1), MediaTime(0));
writer.note_segment_start(SegmentSeq::new(2), MediaTime(4_000_000));
seg(&writer, 1, 4.0, false);
let playlist = origin.render_playlist(DEFAULT_TRACK_ID).expect("render");
assert!(
playlist.contains("#EXT-X-DATERANGE"),
"playlist should contain EXT-X-DATERANGE, got: {playlist}"
);
assert!(
playlist.contains("SCTE35-OUT="),
"daterange should carry SCTE35-OUT attribute: {playlist}"
);
assert!(
playlist.contains("ID=\"2002\""),
"daterange ID should be the event_id: {playlist}"
);
}
#[test]
fn non_scte35_events_are_silently_skipped() {
use timed_metadata::event::{EventKind, SourcePayload};
use timed_metadata::{MediaDuration, TimedEvent};
let (trunk, origin, writer) = make_origin();
writer.set_time_anchor(TimeAnchor {
pts_90k: 0,
utc_epoch_ms: 1_000_000_000_000,
});
let emsg_ev = TimedEvent {
id: Some(7),
kind: EventKind::Unspecified,
at: Some(MediaTime(2_000_000)),
duration: Some(MediaDuration(2_000_000)),
source: SourcePayload::Emsg {
scheme_id_uri: "urn:example".to_string(),
value: "1".to_string(),
raw: vec![0x01, 0x02],
},
};
let trunk_writer = trunk.writer().expect("first trunk writer");
trunk_writer.publish_event(emsg_ev, EventAnchor::Media(MediaTime(2_000_000)));
writer.note_segment_start(SegmentSeq::new(1), MediaTime(0));
writer.note_segment_start(SegmentSeq::new(2), MediaTime(4_000_000));
seg(&writer, 1, 4.0, false);
let playlist = origin.render_playlist(DEFAULT_TRACK_ID).expect("render");
assert!(
!playlist.contains("#EXT-X-DATERANGE"),
"no DATERANGE expected for a non-SCTE-35 event: {playlist}"
);
}
#[test]
fn master_playlist_has_stream_inf() {
let m = master_playlist_m3u8("media.m3u8");
assert!(m.contains("#EXTM3U"));
assert!(m.contains("#EXT-X-STREAM-INF"));
assert!(m.contains("media.m3u8"));
}
#[test]
fn master_playlist_points_at_configured_playlist_name() {
let m = master_playlist_m3u8("index.m3u8");
assert!(m.contains("index.m3u8"));
assert!(!m.contains("media.m3u8"));
}
#[test]
fn playlist_rendered_from_populated_trunk_matches_expected_shape() {
let (_trunk, origin, writer) = make_origin();
seg(&writer, 1, 4.0, false);
part(&writer, 2, 0, true);
part(&writer, 2, 1, false);
let body = match resolve_now(
&origin,
HlsRequest::Playlist {
track_id: DEFAULT_TRACK_ID,
query: BlockingQuery::default(),
},
) {
EgressResponse::Ready {
body: HlsBody::Playlist(m),
cache,
} => {
assert_eq!(cache, CachePolicy::NoCache);
m
}
other => panic!("expected Ready(Playlist), got {other:?}"),
};
assert!(body.contains("#EXT-X-VERSION:6"), "body: {body}");
assert!(!body.contains("#EXT-X-VERSION:9"), "body: {body}");
assert!(body.contains("#EXT-X-TARGETDURATION:4"), "body: {body}");
assert!(
body.contains("#EXT-X-SERVER-CONTROL:CAN-BLOCK-RELOAD=YES,PART-HOLD-BACK=1.5"),
"body: {body}"
);
assert!(
body.contains("#EXT-X-PART-INF:PART-TARGET=0.5"),
"body: {body}"
);
assert!(
body.contains("#EXT-X-MAP:URI=\"init-1-7-1.mp4\""),
"body: {body}"
);
assert!(body.contains("seg-1-7-1.m4s"), "body: {body}");
assert!(
body.contains("#EXT-X-PART:DURATION=0.5") && body.contains("INDEPENDENT=YES"),
"body: {body}"
);
assert!(body.contains("#EXT-X-PRELOAD-HINT"), "body: {body}");
assert!(
body.contains("part-1-7-2.2.m4s"),
"preload hint for the next part: {body}"
);
}
#[test]
fn preload_hinted_part_blocks_until_produced_then_serves() {
let (trunk, origin, writer) = make_origin();
let origin = Arc::new(origin);
let deadline = Timestamp::from_nanos(5_000_000_000);
let policy = AwaitPolicy::new(deadline);
let first = origin.resolve(
HlsRequest::Resource {
name: "part-1-7-1.0.m4s".to_string(),
},
Timestamp::from_nanos(0),
policy,
);
assert!(
matches!(first, EgressResponse::Await { .. }),
"expected Await before the part exists, got {first:?}"
);
let listener = trunk.listen().expect("listener slot available");
let woken = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false));
let woken2 = std::sync::Arc::clone(&woken);
let waiter = std::thread::spawn(move || {
let ok = listener.wait_deadline(Instant::now() + Duration::from_secs(60));
woken2.store(ok, std::sync::atomic::Ordering::SeqCst);
});
part(&writer, 1, 0, true);
waiter.join().expect("waiter thread must not panic");
assert!(
woken.load(std::sync::atomic::Ordering::SeqCst),
"Trunk::listen() must wake once publish_part lands"
);
match origin.resolve(
HlsRequest::Resource {
name: "part-1-7-1.0.m4s".to_string(),
},
Timestamp::from_nanos(1),
policy,
) {
EgressResponse::Ready {
body: HlsBody::Resource(bytes),
cache,
} => {
assert_eq!(bytes, Bytes::from(vec![0u8; 4]));
assert_eq!(cache, CachePolicy::Immutable);
}
other => panic!("expected Ready once produced, got {other:?}"),
}
}
#[test]
fn awaiting_part_is_bounded_by_await_policy_deadline() {
let (_trunk, origin, _writer) = make_origin();
let deadline = Timestamp::from_nanos(1_000_000_000);
let policy = AwaitPolicy::new(deadline);
let still_waiting = origin.resolve(
HlsRequest::Resource {
name: "part-1-7-1.0.m4s".to_string(),
},
Timestamp::from_nanos(999_999_999),
policy,
);
assert!(matches!(still_waiting, EgressResponse::Await { .. }));
let expired = origin.resolve(
HlsRequest::Resource {
name: "part-1-7-1.0.m4s".to_string(),
},
deadline,
policy,
);
assert!(
matches!(expired, EgressResponse::NotFound),
"expected NotFound once the deadline passed, got {expired:?}"
);
}
#[test]
fn just_closed_segment_final_part_still_serves() {
let (_trunk, origin, writer) = make_origin();
part(&writer, 1, 0, true);
part(&writer, 1, 1, false); seg(&writer, 1, 4.0, false);
match resolve_now(
&origin,
HlsRequest::Resource {
name: "part-1-7-1.1.m4s".to_string(),
},
) {
EgressResponse::Ready {
body: HlsBody::Resource(bytes),
..
} => assert_eq!(bytes, Bytes::from(vec![1u8; 4])),
other => panic!("the just-closed segment's final part must still serve, got {other:?}"),
}
assert_eq!(
resolve_now(
&origin,
HlsRequest::Resource {
name: "part-1-7-1.9.m4s".to_string(),
}
),
EgressResponse::NotFound
);
let body = match resolve_now(
&origin,
HlsRequest::Playlist {
track_id: DEFAULT_TRACK_ID,
query: BlockingQuery::default(),
},
) {
EgressResponse::Ready {
body: HlsBody::Playlist(m),
..
} => m,
other => panic!("expected Ready(Playlist), got {other:?}"),
};
assert!(
body.contains("seg-1-7-1.m4s"),
"closed segment rendered whole: {body}"
);
assert_eq!(
body.lines().skip(4).collect::<Vec<_>>(),
vec![
"#EXT-X-SERVER-CONTROL:CAN-BLOCK-RELOAD=YES,PART-HOLD-BACK=1.5",
"#EXT-X-PART-INF:PART-TARGET=0.5",
"#EXT-X-MAP:URI=\"init-1-7-1.mp4\"",
"#EXT-X-PART:DURATION=0.5,URI=\"part-1-7-1.0.m4s\",INDEPENDENT=YES",
"#EXT-X-PART:DURATION=0.5,URI=\"part-1-7-1.1.m4s\"",
"#EXTINF:4,",
"seg-1-7-1.m4s",
],
"{body}"
);
}
#[test]
fn media_sequence_and_discontinuity_sequence_advance_as_window_rolls() {
let (_trunk, origin, writer) = make_origin();
seg(&writer, 1, 4.0, false);
seg(&writer, 2, 4.0, true); seg(&writer, 3, 4.0, false);
seg(&writer, 4, 4.0, false);
let body = match resolve_now(
&origin,
HlsRequest::Playlist {
track_id: DEFAULT_TRACK_ID,
query: BlockingQuery::default(),
},
) {
EgressResponse::Ready {
body: HlsBody::Playlist(m),
..
} => m,
other => panic!("expected Ready(Playlist), got {other:?}"),
};
assert!(body.contains("#EXT-X-MEDIA-SEQUENCE:1"), "body: {body}");
assert!(
!body.contains("#EXT-X-DISCONTINUITY-SEQUENCE"),
"nothing has rolled off the window yet: {body}"
);
assert!(body.contains("#EXT-X-DISCONTINUITY\n"), "body: {body}");
seg(&writer, 5, 4.0, false);
let body = match resolve_now(
&origin,
HlsRequest::Playlist {
track_id: DEFAULT_TRACK_ID,
query: BlockingQuery::default(),
},
) {
EgressResponse::Ready {
body: HlsBody::Playlist(m),
..
} => m,
other => panic!("expected Ready(Playlist), got {other:?}"),
};
assert!(body.contains("#EXT-X-MEDIA-SEQUENCE:2"), "body: {body}");
assert!(
!body.contains("#EXT-X-DISCONTINUITY-SEQUENCE"),
"evicted segment 1 was not discontinuous: {body}"
);
seg(&writer, 6, 4.0, false);
let body = match resolve_now(
&origin,
HlsRequest::Playlist {
track_id: DEFAULT_TRACK_ID,
query: BlockingQuery::default(),
},
) {
EgressResponse::Ready {
body: HlsBody::Playlist(m),
..
} => m,
other => panic!("expected Ready(Playlist), got {other:?}"),
};
assert!(body.contains("#EXT-X-MEDIA-SEQUENCE:3"), "body: {body}");
assert!(
body.contains("#EXT-X-DISCONTINUITY-SEQUENCE:1"),
"segment 2 (discontinuous) has now rolled off the window: {body}"
);
}
#[test]
fn target_duration_is_max_of_configured_and_actual_segment_duration() {
let (_trunk, origin, writer) = make_origin(); seg(&writer, 1, 7.5, false);
let body = match resolve_now(
&origin,
HlsRequest::Playlist {
track_id: DEFAULT_TRACK_ID,
query: BlockingQuery::default(),
},
) {
EgressResponse::Ready {
body: HlsBody::Playlist(m),
..
} => m,
other => panic!("expected Ready(Playlist), got {other:?}"),
};
assert!(
body.contains("#EXT-X-TARGETDURATION:8"),
"TARGETDURATION must be round(7.5)=8, not the configured target: {body}"
);
}
#[test]
fn far_future_msn_rejected() {
let (_trunk, origin, writer) = make_origin();
seg(&writer, 1, 4.0, false);
let outcome = resolve_now(
&origin,
HlsRequest::Playlist {
track_id: DEFAULT_TRACK_ID,
query: BlockingQuery {
hls_msn: Some(1002),
hls_part: None,
},
},
);
assert!(matches!(outcome, EgressResponse::BadRequest { .. }));
}
#[test]
fn msn_at_spec_bound_is_accepted() {
let (_trunk, origin, writer) = make_origin();
seg(&writer, 1, 4.0, false);
let outcome = resolve_now(
&origin,
HlsRequest::Playlist {
track_id: DEFAULT_TRACK_ID,
query: BlockingQuery {
hls_msn: Some(3),
hls_part: None,
},
},
);
assert!(
!matches!(outcome, EgressResponse::BadRequest { .. }),
"msn at spec bound (last_closed+2) must be accepted, not rejected"
);
}
#[test]
fn msn_one_beyond_spec_bound_is_rejected() {
let (_trunk, origin, writer) = make_origin();
seg(&writer, 1, 4.0, false);
let outcome = resolve_now(
&origin,
HlsRequest::Playlist {
track_id: DEFAULT_TRACK_ID,
query: BlockingQuery {
hls_msn: Some(4),
hls_part: None,
},
},
);
assert!(
matches!(outcome, EgressResponse::BadRequest { .. }),
"msn at spec bound + 1 (last_closed+3) must be rejected"
);
}
#[test]
fn part_without_msn_rejected() {
let (_trunk, origin, _writer) = make_origin();
let outcome = resolve_now(
&origin,
HlsRequest::Playlist {
track_id: DEFAULT_TRACK_ID,
query: BlockingQuery {
hls_msn: None,
hls_part: Some(0),
},
},
);
assert!(matches!(outcome, EgressResponse::BadRequest { .. }));
}
#[test]
fn resolve_resource_init_present() {
let (_trunk, origin, _writer) = make_origin();
match resolve_now(
&origin,
HlsRequest::Resource {
name: "init-1.mp4".to_string(),
},
) {
EgressResponse::Ready {
body: HlsBody::Resource(bytes),
cache,
} => {
assert_eq!(bytes, Bytes::from(vec![0xAAu8; 8]));
assert_eq!(cache, CachePolicy::NoCache);
}
other => panic!("expected Ready, got {other:?}"),
}
}
#[test]
fn resolve_resource_unmatched_filename_not_found() {
let (_trunk, origin, _writer) = make_origin();
assert_eq!(
resolve_now(
&origin,
HlsRequest::Resource {
name: "not-a-thing.txt".to_string(),
}
),
EgressResponse::NotFound
);
}
fn make_origin_with(
container: Container,
low_latency_ms: Option<u32>,
) -> (Arc<Trunk>, HlsOrigin, media_plane::trunk::SegmentWriter) {
let trunk = Trunk::new(TrunkConfig::new(nz(64), nz(8), nz(8), nz(8), nz(64)));
let writer = trunk.segment_writer().expect("first segment writer");
let mut builder = HlsOrigin::builder(Arc::clone(&trunk))
.target_duration_secs(4.0)
.window_segments(nz(4))
.container(container);
if let Some(ms) = low_latency_ms {
builder = builder.low_latency(ms);
}
let origin = builder
.instance(7)
.build()
.expect("both required fields set");
(trunk, origin, writer)
}
fn render_body(origin: &HlsOrigin) -> String {
match resolve_now(
origin,
HlsRequest::Playlist {
track_id: DEFAULT_TRACK_ID,
query: BlockingQuery::default(),
},
) {
EgressResponse::Ready {
body: HlsBody::Playlist(m),
..
} => m,
other => panic!("expected Ready(Playlist), got {other:?}"),
}
}
fn extract_version_tag(body: &str) -> Option<u8> {
body.lines()
.find_map(|l| l.strip_prefix("#EXT-X-VERSION:")?.parse::<u8>().ok())
}
fn segment_uris(body: &str) -> Vec<String> {
let lines: Vec<&str> = body.lines().collect();
let mut out = Vec::new();
for i in 0..lines.len() {
if lines[i].starts_with("#EXTINF:")
&& let Some(next) = lines.get(i + 1)
&& !next.starts_with('#')
{
out.push((*next).to_string());
}
}
out
}
fn part_uris(body: &str) -> Vec<String> {
body.lines()
.filter(|l| l.starts_with("#EXT-X-PART:"))
.filter_map(|l| {
let start = l.find("URI=\"")? + "URI=\"".len();
let rest = &l[start..];
let end = rest.find('"')?;
Some(rest[..end].to_string())
})
.collect()
}
fn assert_version_matches_broadcast_hls_derivation(body: &str) -> Option<u8> {
let parsed = MediaPlaylist::parse(body).expect("rendered body must round-trip parse");
let rendered = extract_version_tag(body);
assert_eq!(
rendered,
parsed.computed_version(),
"rendered #EXT-X-VERSION must equal broadcast_hls's own derivation, body: {body}"
);
rendered
}
fn assert_version_present_and_matches_derivation(body: &str) -> u8 {
assert_version_matches_broadcast_hls_derivation(body).unwrap_or_else(|| {
panic!(
"this cell must trigger a real RFC 8216bis §8 version rule -- a \
missing #EXT-X-VERSION makes the derivation check vacuous, body: {body}"
)
})
}
#[test]
fn mpegts_classic_no_map_ts_uris_no_ll_tags() {
let (_trunk, origin, writer) = make_origin_with(Container::MpegTs, None);
seg(&writer, 1, 4.004, false);
let body = render_body(&origin);
assert!(!body.contains("#EXT-X-MAP"), "body: {body}");
assert!(!body.contains("#EXT-X-PART"), "body: {body}");
assert!(!body.contains("#EXT-X-SERVER-CONTROL"), "body: {body}");
assert!(!body.contains("#EXT-X-PRELOAD-HINT"), "body: {body}");
assert!(!body.contains(".m4s"), "body: {body}");
assert!(body.contains("seg-1-7-1.ts"), "body: {body}");
assert_version_present_and_matches_derivation(&body);
let uris = segment_uris(&body);
assert_eq!(uris, vec!["seg-1-7-1.ts".to_string()]);
for uri in uris {
let ImmediateResource::Segment { seq, .. } = parse_immediate(&uri, Container::MpegTs)
.expect("advertised segment URI must parse under MpegTs")
else {
panic!("expected a Segment resource for {uri}");
};
match resolve_now(&origin, HlsRequest::Resource { name: uri.clone() }) {
EgressResponse::Ready {
body: HlsBody::Resource(bytes),
..
} => assert_eq!(bytes, Bytes::from(vec![seq as u8; 8])),
other => panic!("expected Ready for {uri}, got {other:?}"),
}
}
}
#[test]
fn mpegts_low_latency_part_ts_uris_blocking_part_requests_resolve() {
let (trunk, origin, writer) = make_origin_with(Container::MpegTs, Some(500));
let origin = Arc::new(origin);
seg(&writer, 1, 4.004, false);
part(&writer, 2, 0, true);
part(&writer, 2, 1, false);
let body = render_body(&origin);
assert!(!body.contains("#EXT-X-MAP"), "body: {body}");
assert!(body.contains("#EXT-X-PART-INF"), "body: {body}");
assert!(body.contains("#EXT-X-PART:"), "body: {body}");
assert!(body.contains("seg-1-7-1.ts"), "body: {body}");
assert!(body.contains("part-1-7-2.0.ts"), "body: {body}");
assert!(!body.contains(".m4s"), "body: {body}");
assert_version_present_and_matches_derivation(&body);
let parts = part_uris(&body);
assert!(!parts.is_empty(), "body: {body}");
for uri in &parts {
let (_epoch, seq, idx) = parse_part(uri, Container::MpegTs)
.unwrap_or_else(|| panic!("advertised part URI {uri} must parse under MpegTs"));
match resolve_now(&origin, HlsRequest::Resource { name: uri.clone() }) {
EgressResponse::Ready {
body: HlsBody::Resource(bytes),
..
} => {
assert_eq!(bytes, Bytes::from(vec![idx.value() as u8; 4]));
assert_eq!(seq, 2);
}
other => panic!("expected Ready for {uri}, got {other:?}"),
}
}
let deadline = Timestamp::from_nanos(5_000_000_000);
let policy = AwaitPolicy::new(deadline);
let pending = origin.resolve(
HlsRequest::Resource {
name: "part-1-7-2.2.ts".to_string(),
},
Timestamp::from_nanos(0),
policy,
);
assert!(
matches!(pending, EgressResponse::Await { .. }),
"expected Await before the part exists, got {pending:?}"
);
let listener = trunk.listen().expect("listener slot available");
let woken = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false));
let woken2 = std::sync::Arc::clone(&woken);
let waiter = std::thread::spawn(move || {
let ok = listener.wait_deadline(Instant::now() + Duration::from_secs(60));
woken2.store(ok, std::sync::atomic::Ordering::SeqCst);
});
part(&writer, 2, 2, false);
waiter.join().expect("waiter thread must not panic");
assert!(
woken.load(std::sync::atomic::Ordering::SeqCst),
"Trunk::listen() must wake once publish_part lands"
);
match origin.resolve(
HlsRequest::Resource {
name: "part-1-7-2.2.ts".to_string(),
},
Timestamp::from_nanos(1),
policy,
) {
EgressResponse::Ready {
body: HlsBody::Resource(bytes),
..
} => assert_eq!(bytes, Bytes::from(vec![2u8; 4])),
other => panic!("expected Ready once produced, got {other:?}"),
}
}
#[test]
fn mpegts_classic_integral_extinf_emits_no_version_tag() {
let (_trunk, origin, writer) = make_origin_with(Container::MpegTs, None);
seg(&writer, 1, 4.0, false);
let body = render_body(&origin);
assert!(body.contains("#EXTINF:4,"), "body: {body}");
assert!(!body.contains("4.000"), "body: {body}");
assert_eq!(
assert_version_matches_broadcast_hls_derivation(&body),
None,
"nothing in this playlist triggers an RFC 8216bis §8 row: {body}"
);
}
#[test]
fn low_latency_does_not_raise_the_derived_version() {
let (_t1, classic, w1) = make_origin_with(Container::MpegTs, None);
seg(&w1, 1, 4.004, false);
let classic_version = assert_version_present_and_matches_derivation(&render_body(&classic));
let (_t2, low_latency, w2) = make_origin_with(Container::MpegTs, Some(500));
seg(&w2, 1, 4.004, false);
part(&w2, 2, 0, true);
let ll_body = render_body(&low_latency);
assert!(ll_body.contains("#EXT-X-PART:"), "body: {ll_body}");
assert_eq!(
assert_version_present_and_matches_derivation(&ll_body),
classic_version,
"enabling low latency must not raise the derived version"
);
}
#[test]
fn fmp4_classic_map_present_no_ll_tags() {
let (_trunk, origin, writer) = make_origin_with(Container::Fmp4, None);
origin.set_init(vec![0xBBu8; 8]);
seg(&writer, 1, 4.004, false);
let body = render_body(&origin);
assert!(
body.contains("#EXT-X-MAP:URI=\"init-1-7-1.mp4\""),
"body: {body}"
);
assert!(!body.contains("#EXT-X-PART"), "body: {body}");
assert!(!body.contains("#EXT-X-SERVER-CONTROL"), "body: {body}");
assert!(!body.contains("#EXT-X-PRELOAD-HINT"), "body: {body}");
assert!(body.contains("seg-1-7-1.m4s"), "body: {body}");
assert_version_present_and_matches_derivation(&body);
match resolve_now(
&origin,
HlsRequest::Resource {
name: "init-1.mp4".to_string(),
},
) {
EgressResponse::Ready {
body: HlsBody::Resource(bytes),
..
} => assert_eq!(bytes, Bytes::from(vec![0xBBu8; 8])),
other => panic!("expected Ready(init), got {other:?}"),
}
for uri in segment_uris(&body) {
let ImmediateResource::Segment { seq, .. } = parse_immediate(&uri, Container::Fmp4)
.expect("advertised segment URI must parse under Fmp4")
else {
panic!("expected a Segment resource for {uri}");
};
match resolve_now(&origin, HlsRequest::Resource { name: uri.clone() }) {
EgressResponse::Ready {
body: HlsBody::Resource(bytes),
..
} => assert_eq!(bytes, Bytes::from(vec![seq as u8; 8])),
other => panic!("expected Ready for {uri}, got {other:?}"),
}
}
}
#[test]
fn fmp4_low_latency_existing_behaviour_preserved() {
let (_trunk, origin, writer) = make_origin_with(Container::Fmp4, Some(500));
origin.set_init(vec![0xAAu8; 8]);
seg(&writer, 1, 4.004, false);
part(&writer, 2, 0, true);
part(&writer, 2, 1, false);
let body = render_body(&origin);
assert!(
body.contains("#EXT-X-MAP:URI=\"init-1-7-1.mp4\""),
"body: {body}"
);
assert!(body.contains("#EXT-X-PART-INF"), "body: {body}");
assert!(body.contains("#EXT-X-SERVER-CONTROL"), "body: {body}");
assert!(body.contains("#EXT-X-PRELOAD-HINT"), "body: {body}");
assert!(body.contains("seg-1-7-1.m4s"), "body: {body}");
assert!(body.contains("part-1-7-2.0.m4s"), "body: {body}");
assert!(!body.contains(".ts\""), "body: {body}");
assert_version_present_and_matches_derivation(&body);
match resolve_now(
&origin,
HlsRequest::Resource {
name: "init-1.mp4".to_string(),
},
) {
EgressResponse::Ready {
body: HlsBody::Resource(bytes),
..
} => assert_eq!(bytes, Bytes::from(vec![0xAAu8; 8])),
other => panic!("expected Ready(init), got {other:?}"),
}
for uri in segment_uris(&body) {
let ImmediateResource::Segment { seq, .. } = parse_immediate(&uri, Container::Fmp4)
.expect("advertised segment URI must parse under Fmp4")
else {
panic!("expected a Segment resource for {uri}");
};
match resolve_now(&origin, HlsRequest::Resource { name: uri.clone() }) {
EgressResponse::Ready {
body: HlsBody::Resource(bytes),
..
} => assert_eq!(bytes, Bytes::from(vec![seq as u8; 8])),
other => panic!("expected Ready for {uri}, got {other:?}"),
}
}
for uri in part_uris(&body) {
let (_epoch, _seq, idx) = parse_part(&uri, Container::Fmp4)
.unwrap_or_else(|| panic!("advertised part URI {uri} must parse under Fmp4"));
match resolve_now(&origin, HlsRequest::Resource { name: uri.clone() }) {
EgressResponse::Ready {
body: HlsBody::Resource(bytes),
..
} => assert_eq!(bytes, Bytes::from(vec![idx.value() as u8; 4])),
other => panic!("expected Ready for {uri}, got {other:?}"),
}
}
}
#[test]
fn cross_container_refusal_mp4_init_under_mpegts_not_found() {
let (_trunk, origin, _writer) = make_origin_with(Container::MpegTs, None);
origin.set_init(vec![0xCCu8; 8]);
assert_eq!(
resolve_now(
&origin,
HlsRequest::Resource {
name: "init-1.mp4".to_string(),
}
),
EgressResponse::NotFound
);
}
#[test]
fn builder_errors_on_missing_required_fields() {
let trunk = Trunk::new(TrunkConfig::new(nz(64), nz(8), nz(8), nz(8), nz(64)));
match HlsOrigin::builder(Arc::clone(&trunk))
.window_segments(nz(4))
.instance(7)
.build()
{
Err(e) => assert_eq!(e, HlsOriginBuildError::MissingTargetDurationSecs),
Ok(_) => panic!("expected an error: target_duration_secs was never set"),
}
match HlsOrigin::builder(Arc::clone(&trunk))
.target_duration_secs(4.0)
.instance(7)
.build()
{
Err(e) => assert_eq!(e, HlsOriginBuildError::MissingWindowSegments),
Ok(_) => panic!("expected an error: window_segments was never set"),
}
}
#[test]
fn container_label_and_display() {
assert_eq!(Container::Fmp4.name(), "fmp4");
assert_eq!(Container::MpegTs.name(), "mpeg-ts");
assert_eq!(Container::Fmp4.to_string(), "fmp4");
assert_eq!(Container::default(), Container::Fmp4);
}
#[test]
fn closed_segments_snapshot_matches_published_segments_ascending() {
let (_trunk, origin, writer) = make_origin();
writer
.publish_segment(SegmentEntry::new(
Bytes::from(vec![1u8; 8]),
SegmentSeq::new(1),
Duration::from_secs_f64(4.0),
Timestamp::from_nanos(0),
SegmentMeta {
discontinuous: false,
},
))
.unwrap();
writer
.publish_segment(SegmentEntry::new(
Bytes::from(vec![2u8; 8]),
SegmentSeq::new(2),
Duration::from_secs_f64(4.0),
Timestamp::from_nanos(4_000_000_000),
SegmentMeta {
discontinuous: true,
},
))
.unwrap();
let snapshot = origin.closed_segments();
assert_eq!(snapshot.len(), 2);
assert_eq!(snapshot[0].sequence_number, SegmentSeq::new(1));
assert_eq!(snapshot[0].start_ns, 0);
assert!(!snapshot[0].discontinuous);
assert_eq!(snapshot[1].sequence_number, SegmentSeq::new(2));
assert_eq!(snapshot[1].start_ns, 4_000_000_000);
assert!(snapshot[1].discontinuous);
}
#[test]
fn closed_segments_reflects_window_eviction() {
let (_trunk, origin, writer) = make_origin(); for seq in 1..=5 {
seg(&writer, seq, 4.0, false);
}
let snapshot = origin.closed_segments();
let seqs: Vec<u32> = snapshot.iter().map(|s| s.sequence_number.value()).collect();
assert_eq!(
seqs,
vec![2, 3, 4, 5],
"oldest segment 1 must have rolled off"
);
}
fn window_entry(seq: u32, tag: u8, discontinuous: bool) -> SegmentEntry {
SegmentEntry::new(
Bytes::from(vec![tag; 4]),
SegmentSeq::new(seq),
Duration::from_secs(4),
Timestamp::from_nanos(0),
SegmentMeta { discontinuous },
)
}
fn window_seqs(w: &Window) -> Vec<(u32, bool)> {
w.segments
.iter()
.map(|s| (s.sequence_number.value(), s.discontinuous))
.collect()
}
#[test]
fn window_restarts_on_a_repeated_or_decreasing_sequence_number() {
let mut w = Window::new(nz(6));
w.push(window_entry(5, 0xA5, false), Generation::new(1));
w.push(window_entry(6, 0xA6, true), Generation::new(1));
assert_eq!(window_seqs(&w), vec![(5, false), (6, true)]);
w.push(window_entry(3, 0xB3, false), Generation::new(1));
assert_eq!(window_seqs(&w), vec![(3, true)]);
assert_eq!(w.discontinuity_sequence, 1);
assert_eq!(
w.bytes_of(SegmentSeq::new(3)),
Some(Bytes::from(vec![0xB3; 4]))
);
assert_eq!(w.bytes_of(SegmentSeq::new(5)), None);
assert_eq!(w.bytes_of(SegmentSeq::new(6)), None);
w.push(window_entry(3, 0xC3, false), Generation::new(1));
assert_eq!(window_seqs(&w), vec![(3, true)]);
assert_eq!(
w.bytes_of(SegmentSeq::new(3)),
Some(Bytes::from(vec![0xC3; 4]))
);
}
#[test]
fn window_handles_u32_max_sequence_numbers() {
let mut w = Window::new(nz(4));
w.push(window_entry(u32::MAX - 1, 1, false), Generation::new(1));
w.push(window_entry(u32::MAX, 2, false), Generation::new(1));
assert_eq!(
window_seqs(&w),
vec![(u32::MAX - 1, false), (u32::MAX, false)]
);
w.push(window_entry(u32::MAX, 3, false), Generation::new(1));
assert_eq!(window_seqs(&w), vec![(u32::MAX, true)]);
assert_eq!(
w.bytes_of(SegmentSeq::new(u32::MAX)),
Some(Bytes::from(vec![3; 4]))
);
}
#[test]
fn window_capacity_is_never_below_three_segments() {
let mut w = Window::new(nz(1));
for seq in 1..=5 {
w.push(window_entry(seq, 0, false), Generation::new(1));
}
assert_eq!(window_seqs(&w), vec![(3, false), (4, false), (5, false)]);
}
#[test]
fn master_playlist_reports_typed_bandwidth_overflow() {
let (_trunk, origin, writer) = make_origin();
writer
.publish_segment(SegmentEntry::new(
Bytes::from(vec![1u8; 8]),
SegmentSeq::new(1),
Duration::from_nanos(10),
Timestamp::from_nanos(0),
SegmentMeta {
discontinuous: false,
},
))
.unwrap();
assert_eq!(
origin.master_playlist("media.m3u8"),
Err(HlsMasterError::BandwidthOverflow {
bandwidth: 6_400_000_000
})
);
}
#[test]
fn segment_bandwidth_is_rounded_up_and_rejects_degenerate_input() {
assert_eq!(
segment_bandwidth_bps(1001, Duration::from_secs(3)),
Some(2670)
);
assert_eq!(
segment_bandwidth_bps(1_500_000, Duration::from_secs(1)),
Some(12_000_000)
);
assert_eq!(segment_bandwidth_bps(10, Duration::ZERO), None);
assert_eq!(
segment_bandwidth_bps(usize::MAX, Duration::from_nanos(1)),
None
);
}
#[test]
fn bytes_of_addresses_exactly_the_matching_entry() {
let mut w = Window::new(nz(6));
for (seq, tag) in [(5u32, 0x15u8), (6, 0x16), (7, 0x17)] {
w.push(window_entry(seq, tag, false), Generation::new(1));
}
assert_eq!(
w.bytes_of(SegmentSeq::new(5)),
Some(Bytes::from(vec![0x15; 4]))
);
assert_eq!(
w.bytes_of(SegmentSeq::new(6)),
Some(Bytes::from(vec![0x16; 4]))
);
assert_eq!(
w.bytes_of(SegmentSeq::new(7)),
Some(Bytes::from(vec![0x17; 4]))
);
assert_eq!(w.bytes_of(SegmentSeq::new(4)), None);
assert_eq!(w.bytes_of(SegmentSeq::new(8)), None);
assert_eq!(w.bytes_of(SegmentSeq::new(u32::MAX)), None);
}
fn segment_request(name: &str) -> HlsRequest {
HlsRequest::Resource {
name: name.to_string(),
}
}
#[test]
fn a_request_with_nothing_new_does_not_take_the_cursor_lock() {
let (_trunk, origin, writer) = make_origin();
seg(&writer, 1, 4.0, false);
assert!(matches!(
resolve_now(&origin, segment_request("seg-1-7-1.m4s")),
EgressResponse::Ready { .. }
));
let guard = locked(&origin.cursor);
let (tx, rx) = std::sync::mpsc::channel();
std::thread::scope(|s| {
s.spawn(|| {
let _ = tx.send(resolve_now(&origin, segment_request("seg-1-7-1.m4s")));
});
let got = rx.recv_timeout(Duration::from_secs(10));
drop(guard); assert!(
matches!(got, Ok(EgressResponse::Ready { .. })),
"request blocked on the cursor lock: {got:?}"
);
});
}
#[test]
fn a_segment_published_after_a_drain_is_served_by_the_next_request() {
let (_trunk, origin, writer) = make_origin();
seg(&writer, 1, 4.0, false);
assert!(matches!(
resolve_now(&origin, segment_request("seg-1-7-1.m4s")),
EgressResponse::Ready { .. }
));
assert_eq!(
resolve_now(&origin, segment_request("seg-1-7-2.m4s")),
EgressResponse::NotFound
);
seg(&writer, 2, 4.0, false);
assert!(matches!(
resolve_now(&origin, segment_request("seg-1-7-2.m4s")),
EgressResponse::Ready { .. }
));
}
#[test]
fn a_poisoned_lock_does_not_cascade_into_later_requests() {
let (_trunk, origin, writer) = make_origin();
seg(&writer, 1, 4.0, false);
std::thread::scope(|s| {
let origin = &origin;
for which in 0..4 {
let handle = s.spawn(move || {
match which {
0 => drop(origin.window.lock().map(|_g| panic!("poison window"))),
1 => drop(origin.cursor.lock().map(|_g| panic!("poison cursor"))),
2 => drop(origin.init.lock().map(|_g| panic!("poison init"))),
_ => drop(origin.codecs.lock().map(|_g| panic!("poison codecs"))),
}
});
assert!(handle.join().is_err(), "setup thread {which} must panic");
}
});
assert!(origin.window.is_poisoned());
assert!(origin.cursor.is_poisoned());
assert!(origin.init.is_poisoned());
assert!(origin.codecs.is_poisoned());
seg(&writer, 2, 4.0, false);
assert_eq!(origin.closed_segments().len(), 2);
assert!(origin.init_bytes().is_some());
assert!(origin.master_playlist("media.m3u8").is_ok());
origin.set_init(vec![0xCC; 8]);
assert!(render_body(&origin).contains("seg-1-7-2.m4s"));
assert!(matches!(
resolve_now(&origin, segment_request("seg-1-7-1.m4s")),
EgressResponse::Ready { .. }
));
}
}