use std::fs::File;
use std::io::{Read, Seek, SeekFrom};
use std::path::{Path, PathBuf};
use bytes::Bytes;
use hls_runtime::server::ClosedSegment;
use crate::dvr::{DvrConfig, IndexEntry};
const NANOS_PER_SEC_U64: u64 = 1_000_000_000;
const NANOS_PER_SEC: f64 = NANOS_PER_SEC_U64 as f64;
#[derive(Debug, Clone, Copy, PartialEq)]
pub(crate) struct ArchivedSegment {
pub seq: u32,
pub start_pts_ns: u64,
pub duration_secs: f64,
pub discontinuous: bool,
pub period_num: u32,
pub byte_offset: u64,
pub byte_len: u64,
}
#[derive(Debug, Clone, Copy, PartialEq)]
pub(crate) struct CatchupSegment {
pub seq: u32,
pub start_pts_ns: u64,
pub duration_secs: f64,
pub discontinuous: bool,
}
pub(crate) fn archive_dir(dvr: &DvrConfig, route_name: &str) -> PathBuf {
Path::new(&dvr.archive_root).join(route_name)
}
pub(crate) fn list_period_nums(dir: &Path) -> Vec<u32> {
let Ok(entries) = std::fs::read_dir(dir) else {
return Vec::new();
};
let mut nums: Vec<u32> = entries
.filter_map(|e| e.ok())
.filter_map(|e| {
let name = e.file_name();
let name = name.to_str()?.to_string();
name.strip_prefix('p')?
.strip_suffix(".idx")?
.parse::<u32>()
.ok()
})
.collect();
nums.sort_unstable();
nums
}
pub(crate) fn read_period_segments(dir: &Path, period_num: u32) -> Vec<ArchivedSegment> {
let path = dir.join(format!("p{period_num}.idx"));
let data = match std::fs::read(&path) {
Ok(d) => d,
Err(e) => {
tracing::warn!(
path = %path.display(),
error = %e,
"catch-up: could not read period index"
);
return Vec::new();
}
};
let entries: Vec<IndexEntry> = match serde_json::from_slice(&data) {
Ok(e) => e,
Err(e) => {
tracing::warn!(
path = %path.display(),
error = %e,
"catch-up: could not parse period index"
);
return Vec::new();
}
};
entries
.into_iter()
.map(|e| ArchivedSegment {
seq: e.seq,
start_pts_ns: e.start_pts_ns,
duration_secs: e.duration_ns as f64 / NANOS_PER_SEC,
discontinuous: e.discontinuous,
period_num,
byte_offset: e.byte_offset,
byte_len: e.byte_len,
})
.collect()
}
pub(crate) fn scan_archive(dir: &Path) -> Vec<ArchivedSegment> {
list_period_nums(dir)
.into_iter()
.flat_map(|n| read_period_segments(dir, n))
.collect()
}
pub(crate) fn find_archived_segment(dir: &Path, seq: u32) -> Option<ArchivedSegment> {
list_period_nums(dir).into_iter().find_map(|n| {
read_period_segments(dir, n)
.into_iter()
.find(|s| s.seq == seq)
})
}
pub(crate) fn read_archived_bytes(
dir: &Path,
ext: &str,
period_num: u32,
byte_offset: u64,
byte_len: u64,
) -> Result<Bytes, String> {
let path = dir.join(format!("p{period_num}.{ext}"));
let mut file = File::open(&path).map_err(|e| format!("opening {}: {e}", path.display()))?;
let file_len = file
.metadata()
.map_err(|e| format!("stat {}: {e}", path.display()))?
.len();
let end = byte_offset.checked_add(byte_len).ok_or_else(|| {
format!(
"index entry for {} overflows u64: offset {byte_offset} + len {byte_len}",
path.display()
)
})?;
if end > file_len {
return Err(format!(
"index entry for {} claims range [{byte_offset}, {end}) but the file is only \
{file_len} bytes — corrupt or truncated sidecar",
path.display()
));
}
file.seek(SeekFrom::Start(byte_offset))
.map_err(|e| format!("seeking {}: {e}", path.display()))?;
let mut buf = vec![0u8; byte_len as usize];
file.read_exact(&mut buf)
.map_err(|e| format!("reading {}: {e}", path.display()))?;
Ok(Bytes::from(buf))
}
pub(crate) fn merge_segments(
archived: &[ArchivedSegment],
live: &[ClosedSegment],
) -> Vec<CatchupSegment> {
let archive_max_seq = archived.last().map(|s| s.seq);
let mut combined: Vec<CatchupSegment> = archived
.iter()
.map(|s| CatchupSegment {
seq: s.seq,
start_pts_ns: s.start_pts_ns,
duration_secs: s.duration_secs,
discontinuous: s.discontinuous,
})
.collect();
combined.extend(live.iter().filter_map(|s| {
let is_tail = match archive_max_seq {
Some(max) => s.sequence_number > max,
None => true,
};
is_tail.then_some(CatchupSegment {
seq: s.sequence_number,
start_pts_ns: s.start_ns,
duration_secs: s.duration_secs,
discontinuous: s.discontinuous,
})
}));
combined
}
pub(crate) fn apply_window(
combined: &[CatchupSegment],
window_secs: Option<u64>,
) -> Vec<CatchupSegment> {
let Some(window_secs) = window_secs.filter(|&w| w > 0) else {
return combined.to_vec();
};
let Some(edge_ns) = combined.last().map(|s| s.start_pts_ns) else {
return Vec::new();
};
let window_ns = window_secs.saturating_mul(NANOS_PER_SEC_U64);
let floor_ns = edge_ns.saturating_sub(window_ns);
combined
.iter()
.copied()
.filter(|s| s.start_pts_ns >= floor_ns)
.collect()
}
const MIN_TARGET_DURATION_SECS: u32 = 1;
pub(crate) fn render_playlist(
segments: &[CatchupSegment],
ext: &str,
map_uri: Option<&str>,
playlist_type: broadcast_hls::PlaylistType,
endlist: bool,
) -> String {
let target_duration = segments
.iter()
.map(|s| s.duration_secs)
.fold(0.0_f64, f64::max)
.ceil()
.max(f64::from(MIN_TARGET_DURATION_SECS)) as u32;
let media_sequence = segments
.first()
.map(|s| u64::from(s.seq))
.unwrap_or(u64::from(MIN_TARGET_DURATION_SECS));
let hls_segments: Vec<broadcast_hls::MediaSegment> = segments
.iter()
.map(|s| broadcast_hls::MediaSegment {
uri: format!("catchup/seg-{}.{ext}", s.seq),
duration: s.duration_secs,
discontinuous: s.discontinuous,
..Default::default()
})
.collect();
let extra_tags = match map_uri {
Some(uri) => vec![format!("#EXT-X-MAP:URI=\"{uri}\"")],
None => Vec::new(),
};
let playlist = broadcast_hls::MediaPlaylist {
target_duration,
media_sequence,
segments: hls_segments,
endlist,
extra_tags,
playlist_type: Some(playlist_type),
..Default::default()
};
playlist.to_m3u8()
}
#[cfg(test)]
mod tests {
use super::*;
use crate::dvr::{ArchiveOverrunSerde, DvrRecorder};
use media_plane::trunk::{SegmentEntry, Trunk, TrunkConfig};
use std::num::NonZeroUsize;
use std::time::Duration;
fn nz(n: usize) -> NonZeroUsize {
NonZeroUsize::new(n).expect("test capacity must be non-zero")
}
fn temp_dir() -> PathBuf {
use std::sync::atomic::{AtomicU64, Ordering};
static COUNTER: AtomicU64 = AtomicU64::new(0);
let n = COUNTER.fetch_add(1, Ordering::SeqCst);
let dir =
std::env::temp_dir().join(format!("multimux-catchup-{}-{}", std::process::id(), n));
let _ = std::fs::create_dir_all(&dir);
dir
}
fn cleanup(dir: &Path) {
let _ = std::fs::remove_dir_all(dir);
}
fn closed(seq: u32, start_ns: u64, duration_secs: f64, discontinuous: bool) -> ClosedSegment {
ClosedSegment::new(seq, start_ns, duration_secs, discontinuous)
}
#[test]
fn merge_segments_excludes_archived_segments_from_live_tail() {
let archived = vec![
ArchivedSegment {
seq: 1,
start_pts_ns: 0,
duration_secs: 2.0,
discontinuous: false,
period_num: 0,
byte_offset: 0,
byte_len: 10,
},
ArchivedSegment {
seq: 2,
start_pts_ns: 2_000_000_000,
duration_secs: 2.0,
discontinuous: false,
period_num: 0,
byte_offset: 10,
byte_len: 10,
},
ArchivedSegment {
seq: 3,
start_pts_ns: 4_000_000_000,
duration_secs: 2.0,
discontinuous: false,
period_num: 0,
byte_offset: 20,
byte_len: 10,
},
];
let live = vec![
closed(1, 0, 2.0, false),
closed(2, 2_000_000_000, 2.0, false),
closed(3, 4_000_000_000, 2.0, false),
closed(4, 6_000_000_000, 2.0, false),
];
let combined = merge_segments(&archived, &live);
let seqs: Vec<u32> = combined.iter().map(|s| s.seq).collect();
assert_eq!(
seqs,
vec![1, 2, 3, 4],
"archive + live must merge into one continuous sequence, no duplicates"
);
}
#[test]
fn merge_segments_empty_archive_uses_every_live_segment() {
let live = vec![
closed(5, 0, 1.0, false),
closed(6, 1_000_000_000, 1.0, true),
];
let combined = merge_segments(&[], &live);
let seqs: Vec<u32> = combined.iter().map(|s| s.seq).collect();
assert_eq!(seqs, vec![5, 6]);
assert!(combined[1].discontinuous);
}
#[test]
fn apply_window_keeps_only_the_trailing_seconds() {
let combined = vec![
CatchupSegment {
seq: 1,
start_pts_ns: 0,
duration_secs: 2.0,
discontinuous: false,
},
CatchupSegment {
seq: 2,
start_pts_ns: 10_000_000_000,
duration_secs: 2.0,
discontinuous: false,
},
CatchupSegment {
seq: 3,
start_pts_ns: 20_000_000_000,
duration_secs: 2.0,
discontinuous: false,
},
];
let windowed = apply_window(&combined, Some(5));
let seqs: Vec<u32> = windowed.iter().map(|s| s.seq).collect();
assert_eq!(
seqs,
vec![3],
"only the trailing 5s window must survive: {seqs:?}"
);
}
#[test]
fn apply_window_none_or_zero_returns_everything() {
let combined = vec![CatchupSegment {
seq: 1,
start_pts_ns: 0,
duration_secs: 2.0,
discontinuous: false,
}];
assert_eq!(apply_window(&combined, None).len(), 1);
assert_eq!(apply_window(&combined, Some(0)).len(), 1);
}
fn dummy_segment(seq: u32, byte: u8) -> SegmentEntry {
SegmentEntry::new(
bytes::Bytes::from(vec![byte; 24]),
seq,
Duration::from_secs(3),
broadcast_common::Timestamp::from_nanos(u64::from(seq) * 3_000_000_000),
transmux::SegmentMeta {
discontinuous: seq == 2,
},
)
}
fn recorder_cfg(tmp: &Path) -> DvrConfig {
DvrConfig {
enabled: true,
archive_root: tmp.to_string_lossy().to_string(),
retention_periods: 10,
retention_bytes: 0,
period_duration_secs: 3600,
overrun: ArchiveOverrunSerde::Gap,
dvb_service_id: None,
}
}
#[test]
fn scan_archive_recovers_real_recorder_output_byte_exact() {
let tmp = temp_dir();
let trunk = Trunk::new(TrunkConfig::new(nz(4), nz(4), nz(8), nz(4), nz(4)));
let writer = trunk.segment_writer().expect("segment writer");
let mut recorder =
DvrRecorder::new("straddle".to_string(), recorder_cfg(&tmp), ".m4s", &trunk)
.expect("recorder");
let init = b"REAL_INIT";
recorder.poll_and_persist(Some(init)).expect("poll init");
for (seq, byte) in [(1u32, 0xAAu8), (2, 0xBB), (3, 0xCC)] {
writer.publish_segment(dummy_segment(seq, byte));
}
recorder.poll_and_persist(Some(init)).expect("persist");
let dir = archive_dir(&recorder_cfg(&tmp), "straddle");
let archived = scan_archive(&dir);
assert_eq!(archived.len(), 3);
for (i, seg) in archived.iter().enumerate() {
let seq = i as u32 + 1;
assert_eq!(seg.seq, seq);
assert_eq!(
seg.duration_secs, 3.0,
"seq {seq} duration must be real, not a shape"
);
assert_eq!(
seg.discontinuous,
seq == 2,
"seq {seq} discontinuous bit must match what was published"
);
let expected_byte = match seq {
1 => 0xAAu8,
2 => 0xBB,
3 => 0xCC,
_ => unreachable!(),
};
let bytes =
read_archived_bytes(&dir, "m4s", seg.period_num, seg.byte_offset, seg.byte_len)
.expect("read archived bytes");
assert_eq!(
bytes.as_ref(),
vec![expected_byte; 24].as_slice(),
"seq {seq} bytes must be byte-exact with what DvrRecorder wrote"
);
}
cleanup(&tmp);
}
#[test]
fn find_archived_segment_locates_the_right_period_and_range() {
let tmp = temp_dir();
let trunk = Trunk::new(TrunkConfig::new(nz(4), nz(4), nz(8), nz(4), nz(4)));
let writer = trunk.segment_writer().expect("segment writer");
let mut recorder = DvrRecorder::new("find".to_string(), recorder_cfg(&tmp), ".m4s", &trunk)
.expect("recorder");
let init = b"INIT";
recorder.poll_and_persist(Some(init)).expect("poll init");
writer.publish_segment(dummy_segment(1, 0x11));
recorder.poll_and_persist(Some(init)).expect("persist");
let dir = archive_dir(&recorder_cfg(&tmp), "find");
let found = find_archived_segment(&dir, 1).expect("segment 1 must be found");
assert_eq!(found.period_num, 0);
let bytes = read_archived_bytes(
&dir,
"m4s",
found.period_num,
found.byte_offset,
found.byte_len,
)
.expect("read bytes");
assert_eq!(bytes.as_ref(), vec![0x11u8; 24].as_slice());
assert!(find_archived_segment(&dir, 99).is_none());
cleanup(&tmp);
}
#[test]
fn read_archived_bytes_rejects_a_byte_len_the_file_cannot_back() {
let tmp = temp_dir();
let trunk = Trunk::new(TrunkConfig::new(nz(4), nz(4), nz(8), nz(4), nz(4)));
let writer = trunk.segment_writer().expect("segment writer");
let mut recorder =
DvrRecorder::new("corrupt".to_string(), recorder_cfg(&tmp), ".m4s", &trunk)
.expect("recorder");
let init = b"INIT";
recorder.poll_and_persist(Some(init)).expect("poll init");
writer.publish_segment(dummy_segment(1, 0x11));
recorder.poll_and_persist(Some(init)).expect("persist");
let dir = archive_dir(&recorder_cfg(&tmp), "corrupt");
let found = find_archived_segment(&dir, 1).expect("segment 1 must be found");
let implausible_len = 9_100_000_000_000u64; let err = read_archived_bytes(
&dir,
"m4s",
found.period_num,
found.byte_offset,
implausible_len,
)
.expect_err("a byte_len the file cannot back must be rejected, not allocated");
assert!(
err.contains("corrupt or truncated sidecar"),
"expected the bounds-check error, got: {err}"
);
cleanup(&tmp);
}
#[test]
fn render_playlist_renders_real_segment_numbers_and_map() {
let segments = vec![
CatchupSegment {
seq: 5,
start_pts_ns: 0,
duration_secs: 3.4,
discontinuous: false,
},
CatchupSegment {
seq: 6,
start_pts_ns: 3_400_000_000,
duration_secs: 3.4,
discontinuous: true,
},
];
let body = render_playlist(
&segments,
"m4s",
Some("init-1.mp4"),
broadcast_hls::PlaylistType::Event,
false,
);
assert!(body.contains("#EXT-X-MEDIA-SEQUENCE:5"), "body: {body}");
assert!(body.contains("#EXT-X-TARGETDURATION:4"), "body: {body}");
assert!(body.contains("catchup/seg-5.m4s"), "body: {body}");
assert!(body.contains("catchup/seg-6.m4s"), "body: {body}");
assert!(
body.contains("#EXT-X-MAP:URI=\"init-1.mp4\""),
"body: {body}"
);
assert!(body.contains("#EXT-X-DISCONTINUITY\n"), "body: {body}");
assert!(!body.contains("#EXT-X-ENDLIST"), "body: {body}");
assert!(body.contains("#EXT-X-PLAYLIST-TYPE:EVENT"), "body: {body}");
}
#[test]
fn render_playlist_vod_finished_emits_endlist_and_vod_type() {
let segments = vec![CatchupSegment {
seq: 1,
start_pts_ns: 0,
duration_secs: 2.0,
discontinuous: false,
}];
let body = render_playlist(
&segments,
"ts",
None,
broadcast_hls::PlaylistType::Vod,
true,
);
assert!(body.contains("#EXT-X-ENDLIST"), "body: {body}");
assert!(body.contains("#EXT-X-PLAYLIST-TYPE:VOD"), "body: {body}");
assert!(
!body.contains("#EXT-X-MAP"),
"TS container must not advertise a map: {body}"
);
}
#[test]
fn render_playlist_empty_segments_uses_minimum_target_duration() {
let body = render_playlist(&[], "m4s", None, broadcast_hls::PlaylistType::Event, false);
assert!(body.contains("#EXT-X-TARGETDURATION:1"), "body: {body}");
}
}