feat: add shared hls cache sessions (#782)

Shared HLS cache session feature and aligns the runtime, cache, provisioning, recovery, custom-response, and documentation paths with the new shared-session model.
This commit is contained in:
knylbyte
2026-07-08 11:52:22 +02:00
committed by GitHub
parent cbd82fbdfa
commit dc49189f63
201 changed files with 48613 additions and 2097 deletions
+10 -2
View File
@@ -1,10 +1,17 @@
use crate::model::ProxyUserCredentials;
use shared::concat_string;
use shared::utils::{deobfuscate_text, extract_extension_from_url, obfuscate_text, CONSTANTS, HLS_PREFIX};
use shared::{
utils::{deobfuscate_text, extract_extension_from_url, obfuscate_text, CONSTANTS},
defaults::{HLS_PREFIX}
};
use std::borrow::Cow;
use std::str;
use url::Url;
pub mod origin_manifest;
pub mod initial_strip;
pub mod transient_manifest;
const TOKEN_SEPARATOR: char = '\x1F';
const TOKEN_SEPARATOR_STR: &str = "\x1F";
@@ -148,7 +155,8 @@ mod test {
get_hls_session_token_and_url_from_token, rewrite_hls, rewrite_hls_url, RewriteHlsProps,
};
use rand::RngCore;
use shared::utils::{u32_to_base64, HLS_PREFIX};
use shared::utils::{u32_to_base64};
use shared::defaults::{HLS_PREFIX};
#[test]
fn test_token_size() {
@@ -0,0 +1,226 @@
use super::transient_manifest::{
configured_strip_segments, manifest_lines, media_segment_units, strip_mode_log_value,
MIN_HLS_INITIAL_VISIBLE_SEGMENTS,
};
use crate::model::StripConfig;
use shared::model::HlsStripMode;
#[derive(Debug, Clone, Eq, PartialEq)]
pub struct HlsInitialStripView {
pub body: String,
pub outcome: HlsInitialStripOutcome,
}
#[derive(Debug, Clone, Eq, PartialEq)]
pub enum HlsInitialStripOutcome {
Applied {
mode: &'static str,
configured: u64,
effective: usize,
visible_segments: usize,
},
Skipped {
reason: HlsInitialStripSkipReason,
visible_segments: usize,
},
}
#[derive(Debug, Clone, Copy, Eq, PartialEq)]
pub enum HlsInitialStripSkipReason {
StripDisabled,
NotEnoughSegments,
}
impl HlsInitialStripSkipReason {
pub const fn as_log_reason(self) -> &'static str {
match self {
Self::StripDisabled => "strip-disabled",
Self::NotEnoughSegments => "not-enough-segments",
}
}
}
pub fn materialize_initial_hls_strip_view(body: &str, strip: &StripConfig) -> HlsInitialStripView {
let lines = manifest_lines(body);
let units = media_segment_units(&lines);
let media_segment_count = units.len();
if strip.value == 0 {
return HlsInitialStripView {
body: body.to_string(),
outcome: HlsInitialStripOutcome::Skipped {
reason: HlsInitialStripSkipReason::StripDisabled,
visible_segments: media_segment_count,
},
};
}
let effective_strip_segments = effective_initial_hls_strip_segments(strip, &lines, &units);
if effective_strip_segments == 0 {
return HlsInitialStripView {
body: body.to_string(),
outcome: HlsInitialStripOutcome::Skipped {
reason: HlsInitialStripSkipReason::NotEnoughSegments,
visible_segments: media_segment_count,
},
};
}
let visible_segments = media_segment_count.saturating_sub(effective_strip_segments);
let strip_start_unit = media_segment_count.saturating_sub(effective_strip_segments);
let strip_ranges = &units[strip_start_unit..];
let mut stripped = String::with_capacity(body.len());
for (index, line) in lines.iter().enumerate() {
if strip_ranges.iter().any(|unit| unit.contains(index)) {
continue;
}
stripped.push_str(line.line);
stripped.push_str(line.ending);
}
HlsInitialStripView {
body: stripped,
outcome: HlsInitialStripOutcome::Applied {
mode: strip_mode_log_value(strip.mode),
configured: strip.value,
effective: effective_strip_segments,
visible_segments,
},
}
}
pub fn initial_hls_strip_segments_for_durations(strip: &StripConfig, segment_durations_ms: &[u64]) -> usize {
if strip.value == 0 {
return 0;
}
let configured_segments = match strip.mode {
HlsStripMode::Segments => usize::try_from(strip.value).unwrap_or(usize::MAX),
HlsStripMode::Seconds => configured_strip_segments_from_durations(strip.value, segment_durations_ms),
};
let max_removable_segments = segment_durations_ms.len().saturating_sub(MIN_HLS_INITIAL_VISIBLE_SEGMENTS);
configured_segments.min(max_removable_segments)
}
fn effective_initial_hls_strip_segments(
strip: &StripConfig,
lines: &[super::transient_manifest::ManifestLine<'_>],
units: &[super::transient_manifest::MediaSegmentUnit],
) -> usize {
let configured_segments = configured_strip_segments(strip, lines, units);
let max_removable_segments = units.len().saturating_sub(MIN_HLS_INITIAL_VISIBLE_SEGMENTS);
configured_segments.min(max_removable_segments)
}
fn configured_strip_segments_from_durations(strip_seconds: u64, segment_durations_ms: &[u64]) -> usize {
let target_ms = strip_seconds.saturating_mul(1_000);
let mut accumulated_ms = 0_u64;
let mut strip_segments = 0_usize;
for duration_ms in segment_durations_ms.iter().rev() {
strip_segments = strip_segments.saturating_add(1);
accumulated_ms = accumulated_ms.saturating_add(*duration_ms);
if accumulated_ms >= target_ms {
break;
}
}
strip_segments
}
#[cfg(test)]
mod tests {
use super::{materialize_initial_hls_strip_view, HlsInitialStripOutcome, HlsInitialStripSkipReason};
use crate::model::StripConfig;
use shared::model::HlsStripMode;
use std::fmt::Write as _;
fn strip_segments(value: u64) -> StripConfig { StripConfig { mode: HlsStripMode::Segments, value } }
fn strip_seconds(value: u64) -> StripConfig { StripConfig { mode: HlsStripMode::Seconds, value } }
fn manifest_with_segments(count: usize) -> String {
let mut body = "#EXTM3U\n#EXT-X-MEDIA-SEQUENCE:100\n".to_string();
for index in 0..count {
body.push_str("#EXTINF:10.0,\n");
writeln!(&mut body, "seg{index}.ts").expect("write segment URI");
}
body
}
fn media_segment_count(body: &str) -> usize {
body.lines().filter(|line| !line.is_empty() && !line.starts_with('#')).count()
}
#[test]
fn pending_strip_segments_keeps_visible_head_window() {
let view = materialize_initial_hls_strip_view(&manifest_with_segments(6), &strip_segments(3));
assert_eq!(media_segment_count(&view.body), 3);
assert!(matches!(
view.outcome,
HlsInitialStripOutcome::Applied {
mode: "segments",
configured: 3,
effective: 3,
visible_segments: 3,
}
));
}
#[test]
fn pending_strip_segments_never_keeps_less_than_three_segments() {
let four_segment_view = materialize_initial_hls_strip_view(&manifest_with_segments(4), &strip_segments(3));
let three_segment_view = materialize_initial_hls_strip_view(&manifest_with_segments(3), &strip_segments(3));
assert_eq!(media_segment_count(&four_segment_view.body), 3);
assert_eq!(media_segment_count(&three_segment_view.body), 3);
assert!(matches!(
three_segment_view.outcome,
HlsInitialStripOutcome::Skipped {
reason: HlsInitialStripSkipReason::NotEnoughSegments,
visible_segments: 3,
}
));
}
#[test]
fn pending_strip_seconds_counts_tail_extinf_durations() {
let body = "#EXTM3U\n#EXT-X-MEDIA-SEQUENCE:100\n#EXTINF:10.0,\nseg100.ts\n#EXTINF:9.0,\nseg101.ts\n#EXTINF:9.0,\nseg102.ts\n#EXTINF:9.0,\nseg103.ts\n#EXTINF:9.0,\nseg104.ts\n#EXTINF:9.0,\nseg105.ts\n";
let view = materialize_initial_hls_strip_view(body, &strip_seconds(30));
assert_eq!(media_segment_count(&view.body), 3);
assert!(matches!(
view.outcome,
HlsInitialStripOutcome::Applied {
mode: "seconds",
configured: 30,
effective: 3,
visible_segments: 3,
}
));
}
#[test]
fn pending_strip_disabled_keeps_full_body() {
let view = materialize_initial_hls_strip_view(&manifest_with_segments(4), &strip_segments(0));
assert_eq!(media_segment_count(&view.body), 4);
assert!(matches!(
view.outcome,
HlsInitialStripOutcome::Skipped {
reason: HlsInitialStripSkipReason::StripDisabled,
visible_segments: 4,
}
));
}
#[test]
fn pending_strip_preserves_media_sequence_and_byterange_semantics() {
let body = "#EXTM3U\n#EXT-X-MEDIA-SEQUENCE:42\n#EXT-X-BYTERANGE:100@200\n#EXTINF:4.0,\nseg42.ts\n#EXT-X-BYTERANGE:100\n#EXTINF:4.0,\nseg43.ts\n#EXT-X-BYTERANGE:100\n#EXTINF:4.0,\nseg44.ts\n#EXT-X-BYTERANGE:100\n#EXTINF:4.0,\nseg45.ts\n";
let view = materialize_initial_hls_strip_view(body, &strip_segments(1));
assert!(view.body.contains("#EXT-X-MEDIA-SEQUENCE:42"));
assert!(view.body.contains("#EXT-X-BYTERANGE:100@200"));
assert!(view.body.contains("#EXT-X-BYTERANGE:100\n#EXTINF:4.0,\nseg43.ts"));
assert!(!view.body.contains("seg45.ts"));
}
}
@@ -0,0 +1,826 @@
#![allow(dead_code)]
use super::rewrite_hls_url;
use shared::utils::CONSTANTS;
use std::{borrow::Cow, collections::HashMap, fmt};
/// Result of parsing an origin media playlist for the live HLS cache proxy.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum OriginManifestParseOutcome {
Normal(ParsedOriginManifest),
TransientPassthrough { reason: OriginManifestTransientReason },
}
/// Reason why an origin manifest cannot enter the normal cache timeline.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum OriginManifestTransientReason {
ExtXKey,
UnsupportedTag { tag: String },
ParserUnsupportedFeature { feature: String },
}
/// Parsed normal-timeline view of a live HLS origin media playlist.
#[derive(Clone, PartialEq, Eq)]
pub struct ParsedOriginManifest {
pub origin_manifest_sequence: u64,
pub origin_manifest_segment_cnt: usize,
pub version: Option<u16>,
pub target_duration: Option<u32>,
pub discontinuity_sequence: Option<u64>,
pub independent_segments: bool,
/// Parsed MAP fetch references before proxy MAP ID assignment.
pub maps: Vec<ParsedOriginMap>,
/// Parsed segment fetch references before proxy sequence mapping.
pub segments: Vec<ParsedOriginSegment>,
}
/// Timing values parsed from HLS manifest attributes used for origin refresh debounce.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub struct ParsedManifestTiming {
pub target_duration_ms: Option<u64>,
pub last_segment_duration_ms: Option<u64>,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ParsedManifestValiditySource {
ExtInfSum,
TargetDuration,
Fallback,
}
/// Parsed validity window for a committed HLS manifest.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct ParsedManifestValidity {
pub playlist_duration_ms: u64,
pub segment_count: usize,
pub source: ParsedManifestValiditySource,
}
/// Lightweight timeline markers used to validate transient passthrough manifests
/// without applying normal timeline parsing or URI normalization.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct ParsedOriginManifestTimeline {
pub origin_manifest_sequence: u64,
pub origin_manifest_segment_cnt: usize,
}
impl ParsedOriginManifestTimeline {
pub fn origin_highwater(self) -> Option<u64> {
let segment_count = u64::try_from(self.origin_manifest_segment_cnt).ok()?;
if segment_count == 0 {
return None;
}
self.origin_manifest_sequence.checked_add(segment_count.saturating_sub(1))
}
}
impl fmt::Debug for ParsedOriginManifest {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.debug_struct("ParsedOriginManifest")
.field("origin_manifest_sequence", &self.origin_manifest_sequence)
.field("origin_manifest_segment_cnt", &self.origin_manifest_segment_cnt)
.field("version", &self.version)
.field("target_duration", &self.target_duration)
.field("discontinuity_sequence", &self.discontinuity_sequence)
.field("independent_segments", &self.independent_segments)
.field("maps_len", &self.maps.len())
.field("segments_len", &self.segments.len())
.finish()
}
}
/// Parsed origin segment metadata before proxy sequence mapping and rendering.
#[derive(Clone, PartialEq, Eq)]
pub struct ParsedOriginSegment {
pub origin_seq: u64,
pub duration_ms: u64,
/// Concrete absolute segment fetch URL resolved against the final manifest URL after redirects.
///
/// This may intentionally contain a provider mirror, redirect target, or CDN host. It is request-local fetch
/// metadata, not HLS session identity, account binding, or provider-failover state.
pub resolved_origin_url: String,
pub discontinuity_before: bool,
pub program_date_time: Option<String>,
pub daterange_tags_before: Vec<String>,
pub origin_byte_range: Option<ParsedByteRange>,
pub map_ref: Option<usize>,
}
impl fmt::Debug for ParsedOriginSegment {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.debug_struct("ParsedOriginSegment")
.field("origin_seq", &self.origin_seq)
.field("duration_ms", &self.duration_ms)
.field("resolved_origin_url", &"<redacted>")
.field("discontinuity_before", &self.discontinuity_before)
.field("program_date_time", &self.program_date_time)
.field("daterange_tags_before", &self.daterange_tags_before)
.field("origin_byte_range", &self.origin_byte_range)
.field("map_ref", &self.map_ref)
.finish()
}
}
/// Parsed origin MAP metadata before proxy MAP ID assignment and rendering.
#[derive(Clone, PartialEq, Eq)]
pub struct ParsedOriginMap {
pub map_id: usize,
/// Concrete absolute MAP URI resolved against the final manifest URL after
/// redirects. It is a fetch URI, not a provider-failover identity.
pub resolved_origin_uri: String,
pub byte_range: Option<ParsedByteRange>,
}
impl fmt::Debug for ParsedOriginMap {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.debug_struct("ParsedOriginMap")
.field("map_id", &self.map_id)
.field("resolved_origin_uri", &"<redacted>")
.field("byte_range", &self.byte_range)
.finish()
}
}
/// Absolute origin byte range normalized from an `EXT-X-BYTERANGE` tag.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct ParsedByteRange {
pub length: u64,
pub offset: u64,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum PendingByteRange {
WithOffset(ParsedByteRange),
WithoutOffset(u64),
}
pub fn parse_origin_media_manifest(body: &str, final_manifest_url: &str) -> OriginManifestParseOutcome {
match parse_origin_media_manifest_result(body, final_manifest_url) {
Ok(manifest) => OriginManifestParseOutcome::Normal(manifest),
Err(reason) => OriginManifestParseOutcome::TransientPassthrough { reason },
}
}
pub fn parse_origin_manifest_timeline(
body: &str,
) -> Result<ParsedOriginManifestTimeline, OriginManifestTransientReason> {
let mut seen_extm3u = false;
let mut media_sequence = None;
let mut pending_extinf = false;
let mut segment_count = 0_usize;
for line in body.lines().map(str::trim).filter(|line| !line.is_empty()) {
if line.starts_with('#') {
let tag = tag_name(line);
match tag {
"#EXTM3U" => seen_extm3u = true,
"#EXT-X-MEDIA-SEQUENCE" => {
media_sequence = Some(parse_tag_value(line, tag)?.parse_numeric("invalid_media_sequence")?);
}
"#EXTINF" => pending_extinf = true,
_ => {}
}
continue;
}
if !pending_extinf {
return Err(parser_feature("segment_uri_without_extinf"));
}
pending_extinf = false;
segment_count = segment_count.saturating_add(1);
}
if !seen_extm3u {
return Err(parser_feature("missing_extm3u"));
}
if pending_extinf {
return Err(parser_feature("extinf_without_segment_uri"));
}
Ok(ParsedOriginManifestTimeline {
origin_manifest_sequence: media_sequence.unwrap_or(0),
origin_manifest_segment_cnt: segment_count,
})
}
pub fn parse_manifest_timing(body: &str) -> ParsedManifestTiming {
let mut timing = ParsedManifestTiming::default();
for line in body.lines().map(str::trim).filter(|line| !line.is_empty()) {
if let Some(value) = line.strip_prefix("#EXT-X-TARGETDURATION:") {
if let Ok(duration) = value.trim().parse::<u64>() {
timing.target_duration_ms = Some(duration.saturating_mul(1_000));
}
} else if let Some(value) = line.strip_prefix("#EXTINF:").and_then(parse_extinf_duration_millis) {
timing.last_segment_duration_ms = Some(value);
}
}
timing
}
pub fn parse_manifest_validity(body: &str) -> Option<ParsedManifestValidity> {
let mut target_duration_ms = None;
let mut pending_extinf_duration_ms = None;
let mut segment_count = 0_usize;
let mut extinf_sum_ms = 0_u64;
let mut extinf_duration_count = 0_usize;
for line in body.lines().map(str::trim).filter(|line| !line.is_empty()) {
if let Some(value) = line.strip_prefix("#EXT-X-TARGETDURATION:") {
if let Ok(duration) = value.trim().parse::<u64>() {
target_duration_ms = Some(duration.saturating_mul(1_000));
}
continue;
}
if let Some(value) = line.strip_prefix("#EXTINF:").and_then(parse_extinf_duration_millis) {
pending_extinf_duration_ms = Some(value);
continue;
}
if line.starts_with('#') {
continue;
}
segment_count = segment_count.saturating_add(1);
if let Some(duration_ms) = pending_extinf_duration_ms.take() {
extinf_sum_ms = extinf_sum_ms.saturating_add(duration_ms);
extinf_duration_count = extinf_duration_count.saturating_add(1);
}
}
if segment_count == 0 {
return None;
}
if extinf_duration_count == segment_count && extinf_sum_ms > 0 {
return Some(ParsedManifestValidity {
playlist_duration_ms: extinf_sum_ms,
segment_count,
source: ParsedManifestValiditySource::ExtInfSum,
});
}
if let Some(target_duration_ms) = target_duration_ms {
return Some(ParsedManifestValidity {
playlist_duration_ms: target_duration_ms.saturating_mul(u64::try_from(segment_count).ok()?),
segment_count,
source: ParsedManifestValiditySource::TargetDuration,
});
}
Some(ParsedManifestValidity {
playlist_duration_ms: 15_000_u64.saturating_mul(u64::try_from(segment_count).ok()?),
segment_count,
source: ParsedManifestValiditySource::Fallback,
})
}
fn parse_extinf_duration_millis(value: &str) -> Option<u64> {
let duration = value.split(',').next()?.trim();
let (seconds, fraction) = duration.split_once('.').unwrap_or((duration, ""));
let seconds = seconds.parse::<u64>().ok()?;
let millis = fraction.chars().take(3).try_fold((0_u64, 0_u32), |(value, digits), digit| {
digit.to_digit(10).map(|digit| (value.saturating_mul(10).saturating_add(u64::from(digit)), digits + 1))
})?;
let millis = millis.0.saturating_mul(10_u64.saturating_pow(3_u32.saturating_sub(millis.1)));
Some(seconds.saturating_mul(1_000).saturating_add(millis))
}
fn parse_origin_media_manifest_result(
body: &str,
final_manifest_url: &str,
) -> Result<ParsedOriginManifest, OriginManifestTransientReason> {
let mut parser = OriginManifestParser::new(final_manifest_url);
for line in body.lines().map(str::trim).filter(|line| !line.is_empty()) {
parser.parse_line(line)?;
}
parser.finish()
}
struct OriginManifestParser<'a> {
final_manifest_url: &'a str,
seen_extm3u: bool,
version: Option<u16>,
target_duration: Option<u32>,
media_sequence: Option<u64>,
discontinuity_sequence: Option<u64>,
independent_segments: bool,
pending_extinf_duration_ms: Option<u64>,
pending_discontinuity: bool,
pending_program_date_time: Option<String>,
pending_daterange_tags: Vec<String>,
pending_byte_range: Option<PendingByteRange>,
current_map_ref: Option<usize>,
maps: Vec<ParsedOriginMap>,
segments_without_seq: Vec<ParsedOriginSegment>,
next_byte_range_offset_by_uri: HashMap<String, u64>,
}
impl<'a> OriginManifestParser<'a> {
fn new(final_manifest_url: &'a str) -> Self {
Self {
final_manifest_url,
seen_extm3u: false,
version: None,
target_duration: None,
media_sequence: None,
discontinuity_sequence: None,
independent_segments: false,
pending_extinf_duration_ms: None,
pending_discontinuity: false,
pending_program_date_time: None,
pending_daterange_tags: Vec::new(),
pending_byte_range: None,
current_map_ref: None,
maps: Vec::new(),
segments_without_seq: Vec::new(),
next_byte_range_offset_by_uri: HashMap::new(),
}
}
fn parse_line(&mut self, line: &str) -> Result<(), OriginManifestTransientReason> {
if line.starts_with('#') {
return self.parse_tag_line(line);
}
self.parse_segment_uri(line)
}
fn parse_tag_line(&mut self, line: &str) -> Result<(), OriginManifestTransientReason> {
if !line.starts_with("#EXT") {
return Ok(());
}
let tag = tag_name(line);
if tag == "#EXT-X-KEY" {
return Err(OriginManifestTransientReason::ExtXKey);
}
if !is_allowed_normal_timeline_tag(tag) {
return Err(OriginManifestTransientReason::UnsupportedTag { tag: tag.to_string() });
}
match tag {
"#EXTM3U" => self.seen_extm3u = true,
"#EXT-X-VERSION" => self.version = Some(parse_tag_value(line, tag)?.parse_numeric("invalid_version")?),
"#EXT-X-TARGETDURATION" => {
self.target_duration = Some(parse_tag_value(line, tag)?.parse_numeric("invalid_target_duration")?);
}
"#EXT-X-MEDIA-SEQUENCE" => {
self.media_sequence = Some(parse_tag_value(line, tag)?.parse_numeric("invalid_media_sequence")?);
}
"#EXTINF" => self.pending_extinf_duration_ms = Some(parse_extinf_duration_ms(line)?),
"#EXT-X-DISCONTINUITY" => self.pending_discontinuity = true,
"#EXT-X-DISCONTINUITY-SEQUENCE" => {
self.discontinuity_sequence =
Some(parse_tag_value(line, tag)?.parse_numeric("invalid_discontinuity_sequence")?);
}
"#EXT-X-MAP" => self.current_map_ref = Some(self.parse_map(line)?),
"#EXT-X-BYTERANGE" => self.pending_byte_range = Some(parse_byterange_tag(line)?),
"#EXT-X-PROGRAM-DATE-TIME" => {
self.pending_program_date_time = Some(parse_tag_value(line, tag)?.to_string());
}
"#EXT-X-DATERANGE" => self.pending_daterange_tags.push(line.to_string()),
"#EXT-X-INDEPENDENT-SEGMENTS" => self.independent_segments = true,
_ => {}
}
Ok(())
}
fn parse_map(&mut self, line: &str) -> Result<usize, OriginManifestTransientReason> {
let Some(captures) = CONSTANTS.re_hls_uri.captures(line) else {
return Err(parser_feature("map_without_uri"));
};
let Some(uri) = captures.get(1).map(|m| m.as_str()) else {
return Err(parser_feature("map_without_uri"));
};
let resolved_origin_uri = resolve_uri(self.final_manifest_url, uri);
let byte_range = parse_attribute(line, "BYTERANGE").map(parse_map_byterange).transpose()?;
let map_id = self.maps.len();
self.maps.push(ParsedOriginMap { map_id, resolved_origin_uri, byte_range });
Ok(map_id)
}
fn parse_segment_uri(&mut self, line: &str) -> Result<(), OriginManifestTransientReason> {
let Some(duration_ms) = self.pending_extinf_duration_ms.take() else {
return Err(parser_feature("segment_uri_without_extinf"));
};
let resolved_origin_url = resolve_uri(self.final_manifest_url, line);
let origin_byte_range = self.resolve_pending_byte_range(&resolved_origin_url)?;
self.segments_without_seq.push(ParsedOriginSegment {
origin_seq: 0,
duration_ms,
resolved_origin_url,
discontinuity_before: std::mem::take(&mut self.pending_discontinuity),
program_date_time: self.pending_program_date_time.take(),
daterange_tags_before: std::mem::take(&mut self.pending_daterange_tags),
origin_byte_range,
map_ref: self.current_map_ref,
});
Ok(())
}
fn resolve_pending_byte_range(
&mut self,
resolved_origin_url: &str,
) -> Result<Option<ParsedByteRange>, OriginManifestTransientReason> {
let Some(pending) = self.pending_byte_range.take() else {
return Ok(None);
};
let byte_range = match pending {
PendingByteRange::WithOffset(byte_range) => byte_range,
PendingByteRange::WithoutOffset(length) => {
let Some(offset) = self.next_byte_range_offset_by_uri.get(resolved_origin_url).copied() else {
return Err(parser_feature("byterange_without_resolvable_offset"));
};
ParsedByteRange { length, offset }
}
};
self.next_byte_range_offset_by_uri.insert(
resolved_origin_url.to_string(),
byte_range
.offset
.checked_add(byte_range.length)
.ok_or_else(|| parser_feature("byterange_offset_overflow"))?,
);
Ok(Some(byte_range))
}
fn finish(mut self) -> Result<ParsedOriginManifest, OriginManifestTransientReason> {
if !self.seen_extm3u {
return Err(parser_feature("missing_extm3u"));
}
if self.pending_extinf_duration_ms.is_some() {
return Err(parser_feature("extinf_without_segment_uri"));
}
let origin_manifest_sequence = self.media_sequence.unwrap_or(0);
for (idx, segment) in self.segments_without_seq.iter_mut().enumerate() {
segment.origin_seq = origin_manifest_sequence
.checked_add(u64::try_from(idx).map_err(|_| parser_feature("too_many_segments"))?)
.ok_or_else(|| parser_feature("origin_sequence_overflow"))?;
}
Ok(ParsedOriginManifest {
origin_manifest_sequence,
origin_manifest_segment_cnt: self.segments_without_seq.len(),
version: self.version,
target_duration: self.target_duration,
discontinuity_sequence: self.discontinuity_sequence,
independent_segments: self.independent_segments,
maps: self.maps,
segments: self.segments_without_seq,
})
}
}
trait ParseNumericTag {
fn parse_numeric<T>(&self, feature: &'static str) -> Result<T, OriginManifestTransientReason>
where
T: std::str::FromStr;
}
impl ParseNumericTag for str {
fn parse_numeric<T>(&self, feature: &'static str) -> Result<T, OriginManifestTransientReason>
where
T: std::str::FromStr,
{
self.trim().parse().map_err(|_| parser_feature(feature))
}
}
fn tag_name(line: &str) -> &str { line.split_once(':').map_or(line, |(tag, _)| tag) }
fn is_allowed_normal_timeline_tag(tag: &str) -> bool {
matches!(
tag,
"#EXTM3U"
| "#EXT-X-VERSION"
| "#EXT-X-TARGETDURATION"
| "#EXT-X-MEDIA-SEQUENCE"
| "#EXTINF"
| "#EXT-X-DISCONTINUITY"
| "#EXT-X-DISCONTINUITY-SEQUENCE"
| "#EXT-X-MAP"
| "#EXT-X-BYTERANGE"
| "#EXT-X-PROGRAM-DATE-TIME"
| "#EXT-X-DATERANGE"
| "#EXT-X-INDEPENDENT-SEGMENTS"
| "#EXT-X-ALLOW-CACHE"
)
}
fn parse_tag_value<'a>(line: &'a str, tag: &str) -> Result<&'a str, OriginManifestTransientReason> {
line.strip_prefix(tag)
.and_then(|value| value.strip_prefix(':'))
.map(str::trim)
.filter(|value| !value.is_empty())
.ok_or_else(|| parser_feature("missing_tag_value"))
}
fn parse_extinf_duration_ms(line: &str) -> Result<u64, OriginManifestTransientReason> {
let value = parse_tag_value(line, "#EXTINF")?
.split_once(',')
.map_or_else(|| parse_tag_value(line, "#EXTINF").unwrap_or_default(), |(duration, _)| duration)
.trim();
parse_decimal_seconds_to_ms(value).ok_or_else(|| parser_feature("invalid_extinf"))
}
fn parse_decimal_seconds_to_ms(value: &str) -> Option<u64> {
let (whole, fraction) = value.split_once('.').unwrap_or((value, ""));
if whole.is_empty() || !whole.bytes().all(|byte| byte.is_ascii_digit()) {
return None;
}
let whole_ms = whole.parse::<u64>().ok()?.checked_mul(1_000)?;
let mut fraction_ms = 0_u64;
let mut multiplier = 100_u64;
for byte in fraction.bytes().take(3) {
if !byte.is_ascii_digit() {
return None;
}
fraction_ms = fraction_ms.checked_add(u64::from(byte - b'0').checked_mul(multiplier)?)?;
multiplier /= 10;
}
if fraction.bytes().skip(3).any(|byte| !byte.is_ascii_digit()) {
return None;
}
whole_ms.checked_add(fraction_ms)
}
fn parse_byterange_tag(line: &str) -> Result<PendingByteRange, OriginManifestTransientReason> {
parse_byterange_value(parse_tag_value(line, "#EXT-X-BYTERANGE")?)
}
fn parse_map_byterange(value: &str) -> Result<ParsedByteRange, OriginManifestTransientReason> {
match parse_byterange_value(value)? {
PendingByteRange::WithOffset(byte_range) => Ok(byte_range),
PendingByteRange::WithoutOffset(_) => Err(parser_feature("map_byterange_without_offset")),
}
}
fn parse_byterange_value(value: &str) -> Result<PendingByteRange, OriginManifestTransientReason> {
let trimmed = value.trim().trim_matches('"');
if let Some((length, offset)) = trimmed.split_once('@') {
return Ok(PendingByteRange::WithOffset(ParsedByteRange {
length: length.parse_numeric("invalid_byterange_length")?,
offset: offset.parse_numeric("invalid_byterange_offset")?,
}));
}
Ok(PendingByteRange::WithoutOffset(trimmed.parse_numeric("invalid_byterange_length")?))
}
fn parse_attribute<'a>(line: &'a str, name: &str) -> Option<&'a str> {
line.split_once(':')?
.1
.split(',')
.filter_map(|part| part.split_once('='))
.find_map(|(key, value)| (key.trim() == name).then(|| value.trim()))
}
fn resolve_uri(base: &str, reference: &str) -> String {
match rewrite_hls_url(base, reference) {
Cow::Borrowed(value) => value.to_string(),
Cow::Owned(value) => value,
}
}
fn parser_feature(feature: &str) -> OriginManifestTransientReason {
OriginManifestTransientReason::ParserUnsupportedFeature { feature: feature.to_string() }
}
#[cfg(test)]
mod tests {
use super::{
OriginManifestParseOutcome, OriginManifestTransientReason, ParsedByteRange, ParsedManifestValiditySource,
parse_manifest_timing, parse_manifest_validity, parse_origin_manifest_timeline, parse_origin_media_manifest,
};
const BASE_URL: &str = "http://origin.example.com/live/final/index.m3u8";
fn normal_manifest(body: &str) -> super::ParsedOriginManifest {
match parse_origin_media_manifest(body, BASE_URL) {
OriginManifestParseOutcome::Normal(manifest) => manifest,
OriginManifestParseOutcome::TransientPassthrough { reason } => {
panic!("expected normal manifest: {reason:?}")
}
}
}
#[test]
fn missing_media_sequence_defaults_to_zero() {
let manifest = normal_manifest("#EXTM3U\n#EXTINF:4.0,\nseg001.ts\n");
assert_eq!(manifest.origin_manifest_sequence, 0);
assert_eq!(manifest.segments[0].origin_seq, 0);
}
#[test]
fn present_media_sequence_is_used() {
let manifest = normal_manifest("#EXTM3U\n#EXT-X-MEDIA-SEQUENCE:322\n#EXTINF:4.0,\nseg001.ts\n");
assert_eq!(manifest.origin_manifest_sequence, 322);
assert_eq!(manifest.segments[0].origin_seq, 322);
}
#[test]
fn timeline_markers_parse_media_sequence_through_transient_tags() {
let timeline = parse_origin_manifest_timeline(
"#EXTM3U\n#EXT-X-MEDIA-SEQUENCE:226\n#EXT-X-KEY:METHOD=AES-128,URI=\"key.bin\"\n#EXTINF:4.0,\nseg001.ts\n#EXTINF:4.0,\nseg002.ts\n",
)
.expect("timeline markers parse");
assert_eq!(timeline.origin_manifest_sequence, 226);
assert_eq!(timeline.origin_manifest_segment_cnt, 2);
assert_eq!(timeline.origin_highwater(), Some(227));
}
#[test]
fn timeline_markers_default_missing_media_sequence_to_zero() {
let timeline = parse_origin_manifest_timeline(
"#EXTM3U\n#EXT-X-KEY:METHOD=AES-128,URI=\"key.bin\"\n#EXTINF:4.0,\nseg.ts\n",
)
.expect("timeline markers parse");
assert_eq!(timeline.origin_manifest_sequence, 0);
assert_eq!(timeline.origin_manifest_segment_cnt, 1);
}
#[test]
fn extinf_and_segment_uri_create_segment_model() {
let manifest = normal_manifest("#EXTM3U\n#EXTINF:4.567,\nmedia/seg001.ts\n");
let segment = &manifest.segments[0];
assert_eq!(manifest.origin_manifest_segment_cnt, 1);
assert_eq!(segment.duration_ms, 4_567);
assert_eq!(segment.resolved_origin_url, "http://origin.example.com/live/final/media/seg001.ts");
}
#[test]
fn ext_x_key_triggers_transient_passthrough() {
let outcome = parse_origin_media_manifest("#EXTM3U\n#EXT-X-KEY:METHOD=AES-128,URI=\"key.bin\"\n", BASE_URL);
assert_eq!(
outcome,
OriginManifestParseOutcome::TransientPassthrough { reason: OriginManifestTransientReason::ExtXKey }
);
}
#[test]
fn unsupported_tag_triggers_transient_passthrough() {
let outcome = parse_origin_media_manifest("#EXTM3U\n#EXT-X-PART:DURATION=1.0,URI=\"part.ts\"\n", BASE_URL);
assert_eq!(
outcome,
OriginManifestParseOutcome::TransientPassthrough {
reason: OriginManifestTransientReason::UnsupportedTag { tag: "#EXT-X-PART".to_string() }
}
);
}
#[test]
fn ext_x_map_is_parsed_with_resolved_uri() {
let manifest = normal_manifest("#EXTM3U\n#EXT-X-MAP:URI=\"init.mp4\"\n#EXTINF:4.0,\nseg001.m4s\n");
assert_eq!(manifest.maps[0].resolved_origin_uri, "http://origin.example.com/live/final/init.mp4");
assert_eq!(manifest.segments[0].map_ref, Some(0));
}
#[test]
fn ext_x_map_relative_uri_is_resolved_against_final_manifest_url_after_redirect() {
let final_manifest_url = "https://cdn.example.net/live/redirected/playlist.m3u8";
let outcome =
parse_origin_media_manifest("#EXTM3U\n#EXT-X-MAP:URI=\"init.mp4\"\n#EXTINF:4.0,\nseg001.m4s\n", final_manifest_url);
let OriginManifestParseOutcome::Normal(manifest) = outcome else {
panic!("manifest should parse as normal timeline");
};
assert_eq!(manifest.maps[0].resolved_origin_uri, "https://cdn.example.net/live/redirected/init.mp4");
assert_eq!(manifest.segments[0].resolved_origin_url, "https://cdn.example.net/live/redirected/seg001.m4s");
}
#[test]
fn byterange_with_explicit_offset_is_parsed() {
let manifest = normal_manifest("#EXTM3U\n#EXT-X-BYTERANGE:500@1000\n#EXTINF:4.0,\nbig.m4s\n");
assert_eq!(manifest.segments[0].origin_byte_range, Some(ParsedByteRange { length: 500, offset: 1_000 }));
}
#[test]
fn byterange_without_offset_uses_previous_offset_for_same_uri() {
let manifest = normal_manifest(
"#EXTM3U\n#EXT-X-BYTERANGE:500@1000\n#EXTINF:4.0,\nbig.m4s\n#EXT-X-BYTERANGE:250\n#EXTINF:4.0,\nbig.m4s\n",
);
assert_eq!(manifest.segments[1].origin_byte_range, Some(ParsedByteRange { length: 250, offset: 1_500 }));
}
#[test]
fn byterange_without_previous_offset_fails_controlled() {
let outcome = parse_origin_media_manifest("#EXTM3U\n#EXT-X-BYTERANGE:250\n#EXTINF:4.0,\nbig.m4s\n", BASE_URL);
assert_eq!(
outcome,
OriginManifestParseOutcome::TransientPassthrough {
reason: OriginManifestTransientReason::ParserUnsupportedFeature {
feature: "byterange_without_resolvable_offset".to_string()
}
}
);
}
#[test]
fn byterange_offset_overflow_triggers_parser_unsupported_feature() {
let body = "#EXTM3U\n#EXT-X-BYTERANGE:2@18446744073709551614\n#EXTINF:4.0,\nbig.m4s\n";
let outcome = parse_origin_media_manifest(body, BASE_URL);
assert_eq!(
outcome,
OriginManifestParseOutcome::TransientPassthrough {
reason: OriginManifestTransientReason::ParserUnsupportedFeature {
feature: "byterange_offset_overflow".to_string()
}
}
);
}
#[test]
fn discontinuity_and_discontinuity_sequence_are_parsed() {
let manifest = normal_manifest(
"#EXTM3U\n#EXT-X-DISCONTINUITY-SEQUENCE:7\n#EXT-X-DISCONTINUITY\n#EXTINF:4.0,\nseg001.ts\n",
);
assert_eq!(manifest.discontinuity_sequence, Some(7));
assert!(manifest.segments[0].discontinuity_before);
}
#[test]
fn program_date_time_and_daterange_attach_to_next_segment() {
let manifest = normal_manifest(
"#EXTM3U\n#EXT-X-PROGRAM-DATE-TIME:2026-05-27T10:11:12Z\n#EXT-X-DATERANGE:ID=\"ad-1\",START-DATE=\"2026-05-27T10:11:12Z\"\n#EXTINF:4.0,\nseg001.ts\n",
);
let segment = &manifest.segments[0];
assert_eq!(segment.program_date_time.as_deref(), Some("2026-05-27T10:11:12Z"));
assert_eq!(
segment.daterange_tags_before,
vec!["#EXT-X-DATERANGE:ID=\"ad-1\",START-DATE=\"2026-05-27T10:11:12Z\"".to_string()]
);
}
#[test]
fn version_and_target_duration_are_parsed() {
let manifest =
normal_manifest("#EXTM3U\n#EXT-X-VERSION:6\n#EXT-X-TARGETDURATION:10\n#EXTINF:4.0,\nseg001.ts\n");
assert_eq!(manifest.version, Some(6));
assert_eq!(manifest.target_duration, Some(10));
}
#[test]
fn manifest_timing_parses_target_duration_and_last_extinf() {
let timing = parse_manifest_timing(
"#EXTM3U\n#EXT-X-TARGETDURATION:10\n#EXTINF:4.000,\na.ts\n#EXT-X-KEY:METHOD=AES-128,URI=\"k\"\n#EXTINF:6.250,\nb.ts\n",
);
assert_eq!(timing.target_duration_ms, Some(10_000));
assert_eq!(timing.last_segment_duration_ms, Some(6_250));
}
#[test]
fn manifest_timing_normalizes_extinf_fraction_to_milliseconds() {
assert_eq!(
parse_manifest_timing("#EXTM3U\n#EXTINF:9.23,title\nseg.ts\n").last_segment_duration_ms,
Some(9_230)
);
assert_eq!(parse_manifest_timing("#EXTM3U\n#EXTINF:7,\nseg.ts\n").last_segment_duration_ms, Some(7_000));
}
#[test]
fn manifest_validity_prefers_extinf_sum() {
let validity = parse_manifest_validity(
"#EXTM3U\n#EXT-X-TARGETDURATION:10\n#EXTINF:4.000,\na.ts\n#EXTINF:6.250,\nb.ts\n",
)
.expect("validity");
assert_eq!(validity.playlist_duration_ms, 10_250);
assert_eq!(validity.segment_count, 2);
assert_eq!(validity.source, ParsedManifestValiditySource::ExtInfSum);
}
#[test]
fn manifest_validity_falls_back_to_target_duration() {
let validity = parse_manifest_validity("#EXTM3U\n#EXT-X-TARGETDURATION:12\n#EXTINF:x,\na.ts\n#EXTINF:y,\nb.ts\n")
.expect("validity");
assert_eq!(validity.playlist_duration_ms, 24_000);
assert_eq!(validity.segment_count, 2);
assert_eq!(validity.source, ParsedManifestValiditySource::TargetDuration);
}
#[test]
fn manifest_validity_falls_back_to_segment_count_times_15_seconds() {
let validity = parse_manifest_validity("#EXTM3U\n#EXTINF:x,\na.ts\n#EXTINF:y,\nb.ts\n").expect("validity");
assert_eq!(validity.playlist_duration_ms, 30_000);
assert_eq!(validity.segment_count, 2);
assert_eq!(validity.source, ParsedManifestValiditySource::Fallback);
}
}
@@ -0,0 +1,911 @@
use super::rewrite_hls_url;
use crate::api::model::{
ProxySessionId, TransientResourceId, TransientResourceKind, TransientResourceRef,
HLS_ACCESS_LEASE_ID_PLACEHOLDER,
};
use crate::model::StripConfig;
use shared::model::HlsStripMode;
use shared::utils::CONSTANTS;
use std::{collections::HashMap, time::Duration};
use url::Url;
pub(super) const MIN_HLS_INITIAL_VISIBLE_SEGMENTS: usize = 3;
/// Result of a transient passthrough manifest rewrite.
#[derive(Debug, Clone)]
pub struct TransientRewriteResult {
pub body: String,
pub resources: Vec<TransientResourceRef>,
}
#[derive(Debug, Clone, Copy, Default, Eq, PartialEq)]
pub struct TransientRewriteOptions {
pub handoff_discontinuity_sequence: Option<u64>,
}
/// Rewrites only HLS URI surfaces to transient live-HLS proxy resources.
pub struct TransientManifestRewriter;
impl TransientManifestRewriter {
/// Rewrites HLS URI surfaces to transient proxy resources.
///
/// `final_manifest_url` must be the concrete URL of the manifest that was actually fetched. For provider-url
/// failover and HTTP redirects this can be the selected mirror or final CDN/origin host. Relative segment, MAP and
/// key URIs are resolved against this URL so transient resource fetches keep working after redirects.
pub fn rewrite(
body: &str,
final_manifest_url: &str,
proxy_session_id: &ProxySessionId,
reverse_proxy_rewrite_secret: &[u8],
now_ms: u64,
ttl_ms: u64,
) -> TransientRewriteResult {
Self::rewrite_with_options(
body,
final_manifest_url,
proxy_session_id,
reverse_proxy_rewrite_secret,
now_ms,
ttl_ms,
TransientRewriteOptions::default(),
)
}
/// Rewrites HLS URI surfaces to transient proxy resources with extra rendering options.
///
/// Keep `final_manifest_url` aligned with the actual fetched manifest URL. Passing the original `provider://` input
/// URL or a pre-redirect URL would make relative transient resources point at the wrong fetch target.
pub fn rewrite_with_options(
body: &str,
final_manifest_url: &str,
proxy_session_id: &ProxySessionId,
reverse_proxy_rewrite_secret: &[u8],
now_ms: u64,
ttl_ms: u64,
options: TransientRewriteOptions,
) -> TransientRewriteResult {
let mut rewritten_body = String::with_capacity(body.len());
let mut resources = HashMap::<TransientResourceId, TransientResourceRef>::new();
for part in body.split_inclusive('\n') {
let (line, line_ending) = split_line_ending(part);
let rewritten_line = rewrite_line(
line,
final_manifest_url,
proxy_session_id,
reverse_proxy_rewrite_secret,
now_ms,
ttl_ms,
&mut resources,
);
rewritten_body.push_str(&rewritten_line);
rewritten_body.push_str(line_ending);
}
if body.is_empty() {
return TransientRewriteResult {
body: rewritten_body,
resources: Vec::new(),
};
}
if let Some(discontinuity_sequence) = options.handoff_discontinuity_sequence {
rewritten_body = apply_handoff_discontinuity_boundary(&rewritten_body, discontinuity_sequence);
}
TransientRewriteResult {
body: rewritten_body,
resources: resources.into_values().collect(),
}
}
}
pub fn materialize_transient_provisioning_handoff_view(
origin_body: &str,
previous_provisioning_body: Option<&str>,
strip: &StripConfig,
provisioning_segment_duration_ms: u64,
) -> Option<String> {
let previous_provisioning_body = previous_provisioning_body?;
let origin_lines = manifest_lines(origin_body);
let origin_units = media_segment_units(&origin_lines);
if origin_units.is_empty() {
return None;
}
let origin_window_segments = configured_strip_segments(strip, &origin_lines, &origin_units)
.saturating_add(MIN_HLS_INITIAL_VISIBLE_SEGMENTS)
.min(origin_units.len());
let origin_window_start = origin_units.len().saturating_sub(origin_window_segments);
let origin_batch_segments = origin_window_segments.min(MIN_HLS_INITIAL_VISIBLE_SEGMENTS);
if origin_batch_segments == 0 {
return None;
}
let selected_origin_units = &origin_units[origin_window_start..origin_window_start + origin_batch_segments];
let previous_lines = manifest_lines(previous_provisioning_body);
let previous_units = media_segment_units(&previous_lines);
if previous_units.is_empty() {
return None;
}
let provisioning_tail_segments =
TARGET_TRANSIENT_HANDOFF_SEGMENTS.saturating_sub(origin_batch_segments).saturating_sub(1);
let selected_previous_units = previous_units
.len()
.checked_sub(provisioning_tail_segments)
.map_or(previous_units.as_slice(), |start| &previous_units[start..]);
let gap_unit = selected_previous_units.last().copied().or_else(|| previous_units.last().copied())?;
let first_origin_unit = origin_units[0];
let media_sequence = handoff_provisioning_media_sequence(
&previous_lines,
&previous_units,
previous_units.len().saturating_sub(selected_previous_units.len()),
);
let mut rewritten = String::with_capacity(
origin_body
.len()
.saturating_add(previous_provisioning_body.len())
.saturating_add(64),
);
for line in &origin_lines[..first_origin_unit.start] {
if line.line.trim().starts_with("#EXT-X-MEDIA-SEQUENCE:") {
if let Some(media_sequence) = media_sequence {
let _ = std::fmt::Write::write_fmt(
&mut rewritten,
format_args!("#EXT-X-MEDIA-SEQUENCE:{media_sequence}{}", line.ending),
);
continue;
}
}
rewritten.push_str(line.line);
rewritten.push_str(line.ending);
}
append_manifest_block_separator(&mut rewritten);
for unit in selected_previous_units {
append_media_unit_with_duration_override(
&mut rewritten,
&previous_lines,
*unit,
Some(provisioning_segment_duration_ms),
);
}
append_manifest_block_separator(&mut rewritten);
append_gap_media_unit(&mut rewritten, &previous_lines, gap_unit, provisioning_segment_duration_ms);
append_manifest_block_separator(&mut rewritten);
for (index, unit) in selected_origin_units.iter().enumerate() {
if index == 0 && !media_unit_has_discontinuity(&origin_lines, *unit) {
rewritten.push_str("#EXT-X-DISCONTINUITY\n");
}
append_media_unit_with_duration_override(&mut rewritten, &origin_lines, *unit, None);
}
Some(rewritten)
}
pub fn apply_transient_discontinuity_sequence(body: &str, discontinuity_sequence: u64) -> String {
let lines = manifest_lines(body);
let existing_sequence_index = lines.iter().position(|line| is_discontinuity_sequence_tag(line.line.trim()));
let insert_sequence_index = existing_sequence_index.unwrap_or_else(|| discontinuity_sequence_insert_index(&lines));
let mut rewritten = String::with_capacity(body.len().saturating_add(40));
for index in 0..=lines.len() {
if existing_sequence_index.is_none() && index == insert_sequence_index {
let _ = std::fmt::Write::write_fmt(
&mut rewritten,
format_args!("#EXT-X-DISCONTINUITY-SEQUENCE:{discontinuity_sequence}\n"),
);
}
if index == lines.len() {
break;
}
if existing_sequence_index == Some(index) {
let _ = std::fmt::Write::write_fmt(
&mut rewritten,
format_args!("#EXT-X-DISCONTINUITY-SEQUENCE:{}{}", discontinuity_sequence, lines[index].ending),
);
} else {
rewritten.push_str(lines[index].line);
rewritten.push_str(lines[index].ending);
}
}
rewritten
}
pub fn transient_discontinuity_sequence(body: &str) -> Option<u64> {
manifest_lines(body)
.iter()
.find_map(|line| parse_discontinuity_sequence(line.line))
}
pub fn transient_visible_discontinuity_count(body: &str) -> u64 {
u64::try_from(
manifest_lines(body)
.iter()
.filter(|line| is_discontinuity_tag(line.line.trim()))
.count(),
)
.unwrap_or(u64::MAX)
}
fn rewrite_line(
line: &str,
final_manifest_url: &str,
proxy_session_id: &ProxySessionId,
reverse_proxy_rewrite_secret: &[u8],
now_ms: u64,
ttl_ms: u64,
resources: &mut HashMap<TransientResourceId, TransientResourceRef>,
) -> String {
let trimmed = line.trim();
if trimmed.is_empty() {
return line.to_string();
}
if trimmed.starts_with("#EXT-X-KEY:") {
return rewrite_uri_attribute(
line,
final_manifest_url,
proxy_session_id,
reverse_proxy_rewrite_secret,
now_ms,
ttl_ms,
resources,
TransientResourceKind::Key,
);
}
if trimmed.starts_with("#EXT-X-MAP:") {
return rewrite_uri_attribute(
line,
final_manifest_url,
proxy_session_id,
reverse_proxy_rewrite_secret,
now_ms,
ttl_ms,
resources,
TransientResourceKind::Map,
);
}
if trimmed.starts_with('#') && CONSTANTS.re_hls_uri.is_match(line) {
return rewrite_uri_attribute(
line,
final_manifest_url,
proxy_session_id,
reverse_proxy_rewrite_secret,
now_ms,
ttl_ms,
resources,
TransientResourceKind::Other,
);
}
if trimmed.starts_with('#') {
return line.to_string();
}
rewrite_resource_uri(
trimmed,
final_manifest_url,
proxy_session_id,
reverse_proxy_rewrite_secret,
now_ms,
ttl_ms,
resources,
TransientResourceKind::Segment,
)
.0
}
#[allow(clippy::too_many_arguments)]
fn rewrite_uri_attribute(
line: &str,
final_manifest_url: &str,
proxy_session_id: &ProxySessionId,
reverse_proxy_rewrite_secret: &[u8],
now_ms: u64,
ttl_ms: u64,
resources: &mut HashMap<TransientResourceId, TransientResourceRef>,
kind: TransientResourceKind,
) -> String {
let Some(caps) = CONSTANTS.re_hls_uri.captures(line) else {
return line.to_string();
};
let uri = &caps[1];
let (proxy_uri, _) = rewrite_resource_uri(
uri,
final_manifest_url,
proxy_session_id,
reverse_proxy_rewrite_secret,
now_ms,
ttl_ms,
resources,
kind,
);
CONSTANTS
.re_hls_uri
.replace(line, format!(r#"URI="{proxy_uri}""#))
.to_string()
}
#[allow(clippy::too_many_arguments)]
fn rewrite_resource_uri(
uri: &str,
final_manifest_url: &str,
proxy_session_id: &ProxySessionId,
reverse_proxy_rewrite_secret: &[u8],
now_ms: u64,
ttl_ms: u64,
resources: &mut HashMap<TransientResourceId, TransientResourceRef>,
kind: TransientResourceKind,
) -> (String, TransientResourceId) {
// Resolve against the final fetched manifest URL, not the configured provider:// entry or pre-redirect URL. Relative
// resources commonly live under the redirect/CDN host returned by the manifest request.
let resolved_origin_uri = rewrite_hls_url(final_manifest_url, uri).into_owned();
let extension = extension_for_resource(&resolved_origin_uri, kind);
let resource = TransientResourceRef::new(
kind,
resolved_origin_uri,
reverse_proxy_rewrite_secret,
now_ms,
ttl_ms,
Some(extension.clone()),
);
let resource_id = resource.id.clone();
resources
.entry(resource_id.clone())
.and_modify(|existing| *existing = resource.clone())
.or_insert(resource);
(
format!(
"/hls/shared/live/{}/{}/r/{}.{}",
proxy_session_id.0, HLS_ACCESS_LEASE_ID_PLACEHOLDER, resource_id.0, extension
),
resource_id,
)
}
fn extension_for_resource(resolved_origin_uri: &str, kind: TransientResourceKind) -> String {
extract_extension(resolved_origin_uri)
.filter(|extension| extension.bytes().all(|byte| byte.is_ascii_alphanumeric()))
.map_or_else(|| fallback_extension(kind).to_string(), |extension| extension.to_ascii_lowercase())
}
fn extract_extension(resolved_origin_uri: &str) -> Option<String> {
if let Ok(url) = Url::parse(resolved_origin_uri) {
return url
.path_segments()
.and_then(Iterator::last)
.and_then(|file_name| file_name.rsplit_once('.').map(|(_, extension)| extension.to_string()))
.filter(|extension| !extension.is_empty());
}
let without_query = resolved_origin_uri
.split_once('?')
.map_or(resolved_origin_uri, |(path, _)| path);
let without_fragment = without_query
.split_once('#')
.map_or(without_query, |(path, _)| path);
without_fragment
.rsplit_once('.')
.map(|(_, extension)| extension.to_string())
.filter(|extension| !extension.is_empty())
}
fn fallback_extension(kind: TransientResourceKind) -> &'static str {
match kind {
TransientResourceKind::Key => "key",
TransientResourceKind::Map => "mp4",
TransientResourceKind::Segment | TransientResourceKind::Other => "bin",
}
}
fn split_line_ending(part: &str) -> (&str, &str) {
if let Some(line) = part.strip_suffix("\r\n") {
(line, "\r\n")
} else if let Some(line) = part.strip_suffix('\n') {
(line, "\n")
} else {
(part, "")
}
}
#[derive(Debug, Clone, Copy)]
pub(super) struct ManifestLine<'a> {
pub(super) line: &'a str,
pub(super) ending: &'a str,
}
#[derive(Debug, Clone, Copy)]
pub(super) struct MediaSegmentUnit {
pub(super) start: usize,
pub(super) end: usize,
}
impl MediaSegmentUnit {
pub(super) const fn contains(self, index: usize) -> bool { self.start <= index && index <= self.end }
}
pub(super) fn manifest_lines(body: &str) -> Vec<ManifestLine<'_>> {
body.split_inclusive('\n')
.map(|part| {
let (line, ending) = split_line_ending(part);
ManifestLine { line, ending }
})
.collect()
}
pub(super) fn media_segment_units(lines: &[ManifestLine<'_>]) -> Vec<MediaSegmentUnit> {
let mut units = Vec::new();
let mut unit_start = None;
for (index, line) in lines.iter().enumerate() {
let trimmed = line.line.trim();
if is_segment_unit_tag(trimmed) {
unit_start.get_or_insert(index);
continue;
}
if trimmed.is_empty() {
continue;
}
if trimmed.starts_with('#') {
unit_start = None;
continue;
}
let start = unit_start.take().unwrap_or(index);
units.push(MediaSegmentUnit { start, end: index });
}
units
}
pub(super) fn configured_strip_segments(
strip: &StripConfig,
lines: &[ManifestLine<'_>],
units: &[MediaSegmentUnit],
) -> usize {
match strip.mode {
HlsStripMode::Segments => usize::try_from(strip.value).unwrap_or(usize::MAX),
HlsStripMode::Seconds => {
let target_ms = strip.value.saturating_mul(1_000);
let mut accumulated_ms = 0_u64;
let mut strip_segments = 0_usize;
for unit in units.iter().rev() {
strip_segments = strip_segments.saturating_add(1);
accumulated_ms = accumulated_ms.saturating_add(segment_duration_ms(lines, *unit));
if accumulated_ms >= target_ms {
break;
}
}
strip_segments
}
}
}
fn segment_duration_ms(lines: &[ManifestLine<'_>], unit: MediaSegmentUnit) -> u64 {
lines[unit.start..=unit.end]
.iter()
.find_map(|line| {
let trimmed = line.line.trim();
let extinf_value = trimmed.strip_prefix("#EXTINF:")?;
let duration = extinf_value.split_once(',').map_or(extinf_value, |(value, _)| value);
duration_ms_from_extinf(duration)
})
.unwrap_or(0)
}
const TARGET_TRANSIENT_HANDOFF_SEGMENTS: usize = 6;
fn append_media_unit_with_duration_override(
output: &mut String,
lines: &[ManifestLine<'_>],
unit: MediaSegmentUnit,
duration_ms: Option<u64>,
) {
for line in &lines[unit.start..=unit.end] {
if line.line.trim().starts_with("#EXTINF:") {
if let Some(duration_ms) = duration_ms {
let _ = std::fmt::Write::write_fmt(
output,
format_args!("#EXTINF:{},{}", format_duration_ms(duration_ms), line.ending),
);
continue;
}
}
output.push_str(line.line);
output.push_str(line.ending);
}
}
fn append_gap_media_unit(
output: &mut String,
lines: &[ManifestLine<'_>],
unit: MediaSegmentUnit,
duration_ms: u64,
) {
output.push_str("#EXT-X-GAP\n");
let uri_index = lines[unit.start..=unit.end]
.iter()
.rposition(|line| {
let trimmed = line.line.trim();
!trimmed.is_empty() && !trimmed.starts_with('#')
})
.map(|relative_index| unit.start + relative_index);
if let Some(uri_index) = uri_index {
let _ = std::fmt::Write::write_fmt(
output,
format_args!("#EXTINF:{},{}", format_duration_ms(duration_ms), lines[uri_index].ending),
);
output.push_str(lines[uri_index].line);
output.push_str(lines[uri_index].ending);
}
}
fn append_manifest_block_separator(output: &mut String) {
if output.is_empty() || output.ends_with("\n\n") {
return;
}
if !output.ends_with('\n') {
output.push('\n');
}
output.push('\n');
}
fn media_unit_has_discontinuity(lines: &[ManifestLine<'_>], unit: MediaSegmentUnit) -> bool {
lines[unit.start..=unit.end]
.iter()
.any(|line| is_discontinuity_tag(line.line.trim()))
}
fn handoff_provisioning_media_sequence(
lines: &[ManifestLine<'_>],
units: &[MediaSegmentUnit],
selected_unit_start: usize,
) -> Option<u64> {
let media_sequence = lines.iter().find_map(|line| parse_media_sequence(line.line))?;
if selected_unit_start >= units.len() {
return None;
}
Some(media_sequence.saturating_add(u64::try_from(selected_unit_start).ok()?))
}
fn parse_media_sequence(line: &str) -> Option<u64> {
line.trim().strip_prefix("#EXT-X-MEDIA-SEQUENCE:")?.trim().parse().ok()
}
fn format_duration_ms(duration_ms: u64) -> String { format!("{}.{:03}", duration_ms / 1_000, duration_ms % 1_000) }
fn apply_handoff_discontinuity_boundary(body: &str, handoff_discontinuity_sequence: u64) -> String {
let lines = manifest_lines(body);
let origin_discontinuity_sequence = lines
.iter()
.find_map(|line| parse_discontinuity_sequence(line.line))
.unwrap_or(0);
let effective_discontinuity_sequence =
origin_discontinuity_sequence.saturating_add(handoff_discontinuity_sequence);
let existing_sequence_index = lines.iter().position(|line| is_discontinuity_sequence_tag(line.line.trim()));
let insert_sequence_index = existing_sequence_index.unwrap_or_else(|| discontinuity_sequence_insert_index(&lines));
let first_segment = first_media_segment_boundary(&lines);
let mut rewritten = String::with_capacity(body.len().saturating_add(64));
for index in 0..=lines.len() {
if existing_sequence_index.is_none() && index == insert_sequence_index {
let _ = std::fmt::Write::write_fmt(
&mut rewritten,
format_args!("#EXT-X-DISCONTINUITY-SEQUENCE:{effective_discontinuity_sequence}\n"),
);
}
if let Some((boundary_index, has_discontinuity)) = first_segment {
if !has_discontinuity && index == boundary_index {
rewritten.push_str("#EXT-X-DISCONTINUITY\n");
}
}
if index == lines.len() {
break;
}
if existing_sequence_index == Some(index) {
let _ = std::fmt::Write::write_fmt(
&mut rewritten,
format_args!(
"#EXT-X-DISCONTINUITY-SEQUENCE:{}{}",
effective_discontinuity_sequence, lines[index].ending
),
);
} else {
rewritten.push_str(lines[index].line);
rewritten.push_str(lines[index].ending);
}
}
rewritten
}
fn discontinuity_sequence_insert_index(lines: &[ManifestLine<'_>]) -> usize {
lines
.iter()
.position(|line| line.line.trim().starts_with("#EXT-X-MEDIA-SEQUENCE:"))
.or_else(|| lines.iter().position(|line| line.line.trim().starts_with("#EXT-X-TARGETDURATION:")))
.or_else(|| lines.iter().position(|line| line.line.trim() == "#EXTM3U"))
.map_or(0, |index| index + 1)
}
fn first_media_segment_boundary(lines: &[ManifestLine<'_>]) -> Option<(usize, bool)> {
let mut unit_start = None;
let mut unit_has_discontinuity = false;
for (index, line) in lines.iter().enumerate() {
let trimmed = line.line.trim();
if trimmed.is_empty() {
continue;
}
if is_media_segment_unit_tag(trimmed) {
unit_start.get_or_insert(index);
if is_discontinuity_tag(trimmed) {
unit_has_discontinuity = true;
}
continue;
}
if trimmed.starts_with('#') {
unit_start = None;
unit_has_discontinuity = false;
continue;
}
return Some((unit_start.unwrap_or(index), unit_has_discontinuity));
}
None
}
fn parse_discontinuity_sequence(line: &str) -> Option<u64> {
line.trim().strip_prefix("#EXT-X-DISCONTINUITY-SEQUENCE:")?.trim().parse().ok()
}
fn is_discontinuity_sequence_tag(line: &str) -> bool {
line.starts_with("#EXT-X-DISCONTINUITY-SEQUENCE:")
}
fn is_discontinuity_tag(line: &str) -> bool { line == "#EXT-X-DISCONTINUITY" }
fn duration_ms_from_extinf(value: &str) -> Option<u64> {
let seconds = value.trim().parse::<f64>().ok()?;
if !seconds.is_finite() || seconds.is_sign_negative() {
return None;
}
u64::try_from(Duration::from_secs_f64(seconds).as_millis()).ok()
}
fn is_segment_unit_tag(line: &str) -> bool {
is_media_segment_unit_tag(line)
}
fn is_media_segment_unit_tag(line: &str) -> bool {
line.starts_with("#EXTINF:")
|| line.starts_with("#EXT-X-BYTERANGE:")
|| line.starts_with("#EXT-X-PROGRAM-DATE-TIME:")
|| is_discontinuity_tag(line)
}
pub(super) const fn strip_mode_log_value(mode: HlsStripMode) -> &'static str {
match mode {
HlsStripMode::Segments => "segments",
HlsStripMode::Seconds => "seconds",
}
}
#[cfg(test)]
mod tests {
use super::{
apply_transient_discontinuity_sequence, materialize_transient_provisioning_handoff_view,
transient_discontinuity_sequence, TransientManifestRewriter, TransientRewriteOptions,
};
use crate::api::model::{build_transient_resource_id, ProxySessionId, TransientResourceKind};
use crate::model::StripConfig;
use shared::model::HlsStripMode;
const BASE_URL: &str = "http://origin.example.com/live/final/index.m3u8";
fn rewrite(body: &str) -> crate::processing::parser::hls::transient_manifest::TransientRewriteResult {
TransientManifestRewriter::rewrite(body, BASE_URL, &ProxySessionId("proxy-id".to_string()), b"secret", 100, 1_000)
}
fn rewrite_with_handoff(body: &str) -> crate::processing::parser::hls::transient_manifest::TransientRewriteResult {
TransientManifestRewriter::rewrite_with_options(
body,
BASE_URL,
&ProxySessionId("proxy-id".to_string()),
b"secret",
100,
1_000,
TransientRewriteOptions {
handoff_discontinuity_sequence: Some(0),
},
)
}
#[test]
fn ext_x_key_uri_is_rewritten_without_changing_other_attributes() {
let result = rewrite("#EXTM3U\n#EXT-X-KEY:METHOD=AES-128,URI=\"keys/key.bin\",IV=0x1\n#EXTINF:4.0,\nseg.ts\n");
assert!(result.body.contains("#EXT-X-KEY:METHOD=AES-128,URI=\"/hls/shared/live/proxy-id/__hls_access_lease_id__/r/"));
assert!(result.body.contains(".bin\",IV=0x1"));
assert!(!result.body.contains("keys/key.bin"));
}
#[test]
fn segment_uri_lines_are_rewritten_to_transient_resources() {
let result = rewrite("#EXTM3U\n#EXTINF:4.0,\nmedia/seg001.ts\n");
assert!(result.body.contains("/hls/shared/live/proxy-id/__hls_access_lease_id__/r/"));
assert!(result.body.contains(".ts"));
assert!(!result.body.contains("media/seg001.ts"));
}
#[test]
fn ext_x_map_uri_is_rewritten_without_changing_other_attributes() {
let result = rewrite("#EXTM3U\n#EXT-X-MAP:URI=\"init.mp4\",BYTERANGE=\"10@5\"\n#EXTINF:4.0,\nseg.m4s\n");
assert!(result.body.contains("#EXT-X-MAP:URI=\"/hls/shared/live/proxy-id/__hls_access_lease_id__/r/"));
assert!(result.body.contains(".mp4\",BYTERANGE=\"10@5\""));
assert!(!result.body.contains("init.mp4"));
}
#[test]
fn byterange_and_media_sequence_remain_unchanged() {
let result = rewrite("#EXTM3U\n#EXT-X-MEDIA-SEQUENCE:42\n#EXT-X-BYTERANGE:100@200\n#EXTINF:4.0,\nseg.ts\n");
assert!(result.body.contains("#EXT-X-MEDIA-SEQUENCE:42"));
assert!(result.body.contains("#EXT-X-BYTERANGE:100@200"));
}
#[test]
fn transient_output_never_uses_legacy_hls_token_route() {
let result = rewrite(
"#EXTM3U\n#EXT-X-KEY:METHOD=AES-128,URI=\"key.bin\"\n#EXT-X-MAP:URI=\"init.mp4\"\n#EXTINF:4.0,\nseg.ts\n",
);
assert!(!result.body.contains("/hls/user/pass/"));
assert!(!result.body.contains("key.bin"));
assert!(!result.body.contains("init.mp4"));
assert!(!result.body.contains("seg.ts"));
assert_eq!(result.body.matches("/hls/shared/live/proxy-id/__hls_access_lease_id__/r/").count(), 3);
}
#[test]
fn unsupported_uri_attributes_are_rewritten_to_transient_resources() {
let result = rewrite("#EXTM3U\n#EXT-X-PART:DURATION=1.0,URI=\"parts/part001.m4s\"\n");
assert!(result.body.contains("#EXT-X-PART:DURATION=1.0,URI=\"/hls/shared/live/proxy-id/__hls_access_lease_id__/r/"));
assert!(result.body.contains(".m4s\""));
assert!(!result.body.contains("parts/part001.m4s"));
}
#[test]
fn same_origin_uri_uses_same_transient_resource_id() {
let result = rewrite("#EXTM3U\n#EXTINF:4.0,\nseg.ts\n#EXTINF:4.0,\n./seg.ts\n");
let expected = build_transient_resource_id("http://origin.example.com/live/final/seg.ts", b"secret");
assert_eq!(result.resources.len(), 1);
assert_eq!(result.resources[0].id, expected);
assert_eq!(result.body.matches(&expected.0).count(), 2);
}
#[test]
fn relative_resources_use_final_manifest_url_after_http_redirect() {
let final_manifest_url = "https://cdn-final.example.net/redirected/live/index.m3u8";
let result = TransientManifestRewriter::rewrite(
"#EXTM3U\n#EXT-X-KEY:METHOD=AES-128,URI=\"keys/key.bin\"\n#EXT-X-MAP:URI=\"init/init.mp4\"\n#EXTINF:4.0,\nmedia/seg001.ts\n",
final_manifest_url,
&ProxySessionId("proxy-id".to_string()),
b"secret",
100,
1_000,
);
let key_uri = "https://cdn-final.example.net/redirected/live/keys/key.bin";
let map_uri = "https://cdn-final.example.net/redirected/live/init/init.mp4";
let segment_uri = "https://cdn-final.example.net/redirected/live/media/seg001.ts";
let expected_segment_id = build_transient_resource_id(segment_uri, b"secret");
assert_eq!(result.resources.len(), 3);
assert!(result.resources.iter().any(|resource| {
resource.kind == TransientResourceKind::Key && resource.resolved_origin_uri == key_uri
}));
assert!(result.resources.iter().any(|resource| {
resource.kind == TransientResourceKind::Map && resource.resolved_origin_uri == map_uri
}));
let segment = result
.resources
.iter()
.find(|resource| resource.kind == TransientResourceKind::Segment)
.expect("segment resource");
assert_eq!(segment.resolved_origin_uri, segment_uri);
assert_eq!(segment.id, expected_segment_id);
assert!(result.body.contains(&format!("/r/{}.ts", expected_segment_id.0)));
}
#[test]
fn same_relative_resource_on_different_final_hosts_uses_distinct_transient_ids() {
let first = TransientManifestRewriter::rewrite(
"#EXTM3U\n#EXTINF:4.0,\nseg.ts\n",
"https://cdn-a.example.net/live/index.m3u8",
&ProxySessionId("proxy-id".to_string()),
b"secret",
100,
1_000,
);
let second = TransientManifestRewriter::rewrite(
"#EXTM3U\n#EXTINF:4.0,\nseg.ts\n",
"https://cdn-b.example.net/live/index.m3u8",
&ProxySessionId("proxy-id".to_string()),
b"secret",
100,
1_000,
);
let first_resource = first.resources.first().expect("first resource");
let second_resource = second.resources.first().expect("second resource");
assert_eq!(first_resource.resolved_origin_uri, "https://cdn-a.example.net/live/seg.ts");
assert_eq!(second_resource.resolved_origin_uri, "https://cdn-b.example.net/live/seg.ts");
assert_ne!(first_resource.id, second_resource.id);
}
#[test]
fn handoff_boundary_sets_discontinuity_sequence_and_first_segment_discontinuity() {
let result = rewrite_with_handoff(
"#EXTM3U\n#EXT-X-DISCONTINUITY-SEQUENCE:7\n#EXT-X-MEDIA-SEQUENCE:10\n#EXTINF:4.0,\nseg.ts\n",
);
assert!(result.body.contains("#EXT-X-DISCONTINUITY-SEQUENCE:7\n"));
assert!(result.body.contains("#EXT-X-DISCONTINUITY\n#EXTINF:4.0,\n"));
assert_eq!(result.body.matches("#EXT-X-DISCONTINUITY\n").count(), 1);
}
#[test]
fn handoff_boundary_does_not_duplicate_existing_first_segment_discontinuity() {
let result = rewrite_with_handoff(
"#EXTM3U\n#EXT-X-DISCONTINUITY-SEQUENCE:7\n#EXT-X-MEDIA-SEQUENCE:10\n#EXT-X-DISCONTINUITY\n#EXTINF:4.0,\nseg.ts\n",
);
assert!(result.body.contains("#EXT-X-DISCONTINUITY-SEQUENCE:7\n"));
assert_eq!(result.body.matches("#EXT-X-DISCONTINUITY\n").count(), 1);
}
#[test]
fn transient_provisioning_handoff_view_keeps_tail_gap_and_origin_head() {
let previous = "#EXTM3U\n#EXT-X-VERSION:7\n#EXT-X-INDEPENDENT-SEGMENTS\n#EXT-X-TARGETDURATION:3\n#EXT-X-MEDIA-SEQUENCE:0\n#EXT-X-DISCONTINUITY-SEQUENCE:0\n#EXTINF:2.000,\n/hls/shared/live/proxy-id/__hls_access_lease_id__/000000.ts?pseq=0\n#EXTINF:2.000,\n/hls/shared/live/proxy-id/__hls_access_lease_id__/000001.ts?pseq=1\n#EXTINF:2.000,\n/hls/shared/live/proxy-id/__hls_access_lease_id__/000002.ts?pseq=2\n#EXTINF:2.000,\n/hls/shared/live/proxy-id/__hls_access_lease_id__/000003.ts?pseq=3\n";
let rewritten_origin = rewrite_with_handoff(
"#EXTM3U\n#EXT-X-TARGETDURATION:2\n#EXT-X-MEDIA-SEQUENCE:2455\n#EXTINF:0.64,\na.ts\n#EXTINF:1.92,\nb.ts\n#EXTINF:0.84,\nc.ts\n#EXTINF:1.08,\nd.ts\n#EXTINF:0.56,\ne.ts\n#EXTINF:1.36,\nf.ts\n",
);
let body = materialize_transient_provisioning_handoff_view(
&rewritten_origin.body,
Some(previous),
&strip_segments(2),
2_000,
)
.expect("handoff view should render");
assert!(body.contains("#EXT-X-MEDIA-SEQUENCE:2\n"));
assert!(body.contains("#EXT-X-TARGETDURATION:2\n"));
assert!(!body.contains("#EXT-X-TARGETDURATION:3\n"));
assert!(body.contains("#EXT-X-DISCONTINUITY-SEQUENCE:0\n\n#EXTINF:2.000,"));
assert_eq!(body.matches("#EXTINF:").count(), 6);
assert!(body.contains("#EXTINF:2.000,\n/hls/shared/live/proxy-id/__hls_access_lease_id__/000002.ts?pseq=2"));
assert!(body.contains("\n#EXT-X-GAP\n#EXTINF:2.000,\n/hls/shared/live/proxy-id/__hls_access_lease_id__/000003.ts?pseq=3"));
assert!(body.contains("\n#EXT-X-DISCONTINUITY\n#EXTINF:1.92,"));
let gap = body.find("\n#EXT-X-GAP\n").expect("gap tag");
let discontinuity = body.find("\n#EXT-X-DISCONTINUITY\n#EXTINF:1.92,").expect("handoff discontinuity");
assert!(gap < discontinuity);
}
#[test]
fn transient_discontinuity_sequence_is_kept_after_handoff() {
let body = apply_transient_discontinuity_sequence(
"#EXTM3U\n#EXT-X-VERSION:3\n#EXT-X-MEDIA-SEQUENCE:6560\n#EXT-X-TARGETDURATION:12\n#EXTINF:9.6,\na.ts\n",
1,
);
assert_eq!(transient_discontinuity_sequence(&body), Some(1));
assert!(body.contains("#EXT-X-MEDIA-SEQUENCE:6560\n#EXT-X-DISCONTINUITY-SEQUENCE:1\n"));
}
fn strip_segments(value: u64) -> StripConfig { StripConfig { mode: HlsStripMode::Segments, value } }
}
+2 -1
View File
@@ -4,7 +4,8 @@ use shared::model::{
CatchupAttribute, CatchupProperties, LiveStreamProperties, PlaylistGroup, PlaylistItem, PlaylistItemHeader,
PlaylistItemType, StreamProperties, XtreamCluster,
};
use shared::utils::{default_supported_video_extensions, extract_id_from_url, extract_numeric_id_from_url, Internable};
use shared::utils::{extract_id_from_url, extract_numeric_id_from_url, Internable};
use shared::defaults::{default_supported_video_extensions};
use std::borrow::BorrowMut;
use std::sync::Arc;
use tokio::io::AsyncBufReadExt;
+2 -2
View File
@@ -67,8 +67,8 @@ macro_rules! create_resolve_options_function_for_xtream_target {
false,
false,
false,
shared::utils::default_resolve_delay_secs(),
shared::utils::default_resolve_background(),
shared::defaults::default_resolve_delay_secs(),
shared::defaults::default_resolve_background(),
)
};
+4 -1
View File
@@ -48,9 +48,12 @@ use shared::{
SourceStats, StreamProperties, TargetStats, UUIDType, XtreamCluster,
},
utils::{
create_alias_uuid, default_as_default, default_probe_delay_secs, default_probe_live_interval, interner_gc,
create_alias_uuid, interner_gc,
Internable,
},
defaults::{
default_as_default, default_probe_delay_secs, default_probe_live_interval,
}
};
use std::{
collections::{HashMap, HashSet},
@@ -49,7 +49,8 @@ mod tests {
use arc_swap::{ArcSwap, ArcSwapOption};
use shared::{
model::{ConfigPaths, InputFetchMethod, InputType},
utils::{default_probe_user_priority, Internable},
utils::{Internable},
defaults::{default_probe_user_priority},
};
use std::{collections::HashMap, sync::Arc};
@@ -23,7 +23,7 @@ impl Default for ResolveOptions {
fn default() -> Self {
Self {
flags: ResolveOptionsFlags::Background.into(),
resolve_delay: shared::utils::default_resolve_delay_secs(),
resolve_delay: shared::defaults::default_resolve_delay_secs(),
}
}
}
@@ -36,7 +36,7 @@ use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc;
use std::time::{Duration, Instant};
use shared::foundation::ValueProvider;
use shared::utils::default_probe_user_priority;
use shared::defaults::default_probe_user_priority;
create_resolve_options_function_for_xtream_target!(series);
@@ -36,7 +36,7 @@ use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc;
use std::time::{Duration, Instant};
use shared::foundation::ValueProvider;
use shared::utils::default_probe_user_priority;
use shared::defaults::default_probe_user_priority;
create_resolve_options_function_for_xtream_target!(vod);