873 lines
32 KiB
Rust
873 lines
32 KiB
Rust
use crate::content_hash;
|
||
use crate::database::PreviewDao;
|
||
use crate::libraries::Library;
|
||
use crate::otel::global_tracer;
|
||
use crate::video::ffmpeg::{generate_preview_clip, get_duration_seconds_blocking};
|
||
use crate::video::hls_paths;
|
||
use actix::prelude::*;
|
||
use log::{debug, error, info, warn};
|
||
use opentelemetry::KeyValue;
|
||
use opentelemetry::trace::{Span, Status, Tracer};
|
||
use std::io::Result;
|
||
use std::path::{Path, PathBuf};
|
||
use std::process::{Command, Stdio};
|
||
use std::sync::{Arc, Mutex};
|
||
use tokio::sync::Semaphore;
|
||
// ffmpeg -i test.mp4 -c:v h264 -flags +cgop -g 30 -hls_time 3 out.m3u8
|
||
// ffmpeg -i "filename.mp4" -preset veryfast -c:v libx264 -f hls -hls_list_size 100 -hls_time 2 -crf 24 -vf scale=1080:-2,setsar=1:1 attempt/vid_out.m3u8
|
||
|
||
pub struct StreamActor;
|
||
|
||
impl Actor for StreamActor {
|
||
type Context = Context<Self>;
|
||
}
|
||
|
||
/// A video paired with its content hash, ready to be queued for HLS
|
||
/// playlist generation. Hash is required because all output paths are
|
||
/// keyed on it; callers that lack a hash (rows mid-backfill) must skip
|
||
/// the video rather than fabricate one.
|
||
#[derive(Debug, Clone)]
|
||
pub struct VideoToQueue {
|
||
pub video_path: PathBuf,
|
||
pub content_hash: String,
|
||
}
|
||
|
||
pub fn generate_video_thumbnail(path: &Path, destination: &Path) -> std::io::Result<()> {
|
||
// Probe duration up front and seek to ~50% — gives a more
|
||
// representative frame than a fixed offset (skipping title cards on
|
||
// long videos, landing inside the clip on 1–2s Snapchat MP4s) and
|
||
// sidesteps the seek-past-EOF class of bug entirely. When duration
|
||
// probing fails (LRV files, fragmented MP4s, ffprobe missing) fall
|
||
// back to the first frame: ugly but reliable.
|
||
//
|
||
// -vf scale + -c:v mjpeg mirrors `generate_image_thumbnail_ffmpeg`. The
|
||
// filter chain matters as much as the scale does: without it, ffmpeg
|
||
// hands the decoded frame straight to the mjpeg encoder, which rejects
|
||
// any non-yuvj420p source ("Non full-range YUV is non-standard"). The
|
||
// filter chain lets ffmpeg auto-insert the pix_fmt converter the
|
||
// encoder needs, which is how the image-thumbnail path already handles
|
||
// the same class of source.
|
||
let seek = get_duration_seconds_blocking(path).map(|d| format!("{:.3}", d / 2.0));
|
||
|
||
let mut cmd = Command::new("ffmpeg");
|
||
cmd.arg("-y");
|
||
if let Some(s) = &seek {
|
||
cmd.arg("-ss").arg(s);
|
||
}
|
||
let output = cmd
|
||
.arg("-i")
|
||
.arg(path)
|
||
.arg("-vframes")
|
||
.arg("1")
|
||
.arg("-vf")
|
||
.arg("scale=200:-1")
|
||
.arg("-f")
|
||
.arg("image2")
|
||
.arg("-c:v")
|
||
.arg("mjpeg")
|
||
.arg(destination)
|
||
.output()?;
|
||
|
||
if !output.status.success() {
|
||
return Err(std::io::Error::other(format!(
|
||
"ffmpeg failed ({}): {}",
|
||
output.status,
|
||
String::from_utf8_lossy(&output.stderr).trim()
|
||
)));
|
||
}
|
||
// ffmpeg can exit 0 without writing a frame for malformed files where
|
||
// the probe duration lies. Confirm a non-empty file actually landed —
|
||
// returning Err makes the caller write the `.unsupported` sentinel so
|
||
// we stop re-detecting on every scan.
|
||
let wrote = std::fs::metadata(destination)
|
||
.map(|m| m.len() > 0)
|
||
.unwrap_or(false);
|
||
if !wrote {
|
||
return Err(std::io::Error::other(
|
||
"ffmpeg exited successfully but produced no thumbnail output",
|
||
));
|
||
}
|
||
Ok(())
|
||
}
|
||
|
||
/// Use ffmpeg to extract a 200px-wide thumbnail from formats the `image` crate
|
||
/// can't decode (RAW: NEF/ARW, HEIC/HEIF). Writes JPEG bytes to `destination`
|
||
/// regardless of its extension.
|
||
pub fn generate_image_thumbnail_ffmpeg(path: &Path, destination: &Path) -> std::io::Result<()> {
|
||
let output = Command::new("ffmpeg")
|
||
.arg("-y")
|
||
.arg("-i")
|
||
.arg(path)
|
||
.arg("-vframes")
|
||
.arg("1")
|
||
.arg("-vf")
|
||
.arg("scale=200:-1")
|
||
.arg("-f")
|
||
.arg("image2")
|
||
.arg("-c:v")
|
||
.arg("mjpeg")
|
||
.arg(destination)
|
||
.output()?;
|
||
|
||
if !output.status.success() {
|
||
return Err(std::io::Error::other(format!(
|
||
"ffmpeg failed ({}): {}",
|
||
output.status,
|
||
String::from_utf8_lossy(&output.stderr).trim()
|
||
)));
|
||
}
|
||
Ok(())
|
||
}
|
||
|
||
/// Video stream metadata needed to pick HLS encode settings. Populated by
|
||
/// a single ffprobe call to avoid spawning multiple subprocesses per video.
|
||
#[derive(Debug, Default)]
|
||
pub struct VideoStreamMeta {
|
||
pub is_h264: bool,
|
||
/// Rotation in degrees (0/90/180/270). Checks both the legacy `rotate`
|
||
/// stream tag and the modern display-matrix side data.
|
||
pub rotation: i32,
|
||
/// Frames per second. Prefers `avg_frame_rate` (handles VFR better than
|
||
/// `r_frame_rate`, which lies on variable-framerate sources). `None`
|
||
/// when ffprobe couldn't parse either field — caller picks a fallback.
|
||
pub frame_rate: Option<f32>,
|
||
}
|
||
|
||
/// Parse ffprobe's rational frame-rate strings (`"30000/1001"`,
|
||
/// `"60/1"`, `"0/0"`). Rejects 0/0 (ffprobe's "unknown" sentinel),
|
||
/// non-positive results, and anything wildly out of range so a malformed
|
||
/// probe can't poison the scrubber's step size.
|
||
fn parse_ffprobe_rational(s: &str) -> Option<f32> {
|
||
let (num, den) = s.split_once('/')?;
|
||
let num: f32 = num.parse().ok()?;
|
||
let den: f32 = den.parse().ok()?;
|
||
if den.abs() < f32::EPSILON {
|
||
return None;
|
||
}
|
||
let v = num / den;
|
||
(v.is_finite() && v > 0.0 && v < 1000.0).then_some(v)
|
||
}
|
||
|
||
/// Probe video stream metadata in one ffprobe call. Returns default (codec
|
||
/// unknown, rotation 0, fps None) on any failure — callers fall back to
|
||
/// transcoding / a default framerate.
|
||
pub async fn probe_video_stream_meta(video_path: &str) -> VideoStreamMeta {
|
||
let output = tokio::process::Command::new("ffprobe")
|
||
.arg("-v")
|
||
.arg("error")
|
||
.arg("-select_streams")
|
||
.arg("v:0")
|
||
.arg("-print_format")
|
||
.arg("json")
|
||
.arg("-show_entries")
|
||
.arg("stream=codec_name,r_frame_rate,avg_frame_rate:stream_tags=rotate:side_data_list")
|
||
.arg(video_path)
|
||
.output()
|
||
.await;
|
||
|
||
let Ok(output) = output else {
|
||
warn!("Failed to run ffprobe for {}", video_path);
|
||
return VideoStreamMeta::default();
|
||
};
|
||
if !output.status.success() {
|
||
warn!(
|
||
"ffprobe failed for {}: {}",
|
||
video_path,
|
||
String::from_utf8_lossy(&output.stderr).trim()
|
||
);
|
||
return VideoStreamMeta::default();
|
||
}
|
||
|
||
let Ok(json) = serde_json::from_slice::<serde_json::Value>(&output.stdout) else {
|
||
warn!("ffprobe returned non-JSON for {}", video_path);
|
||
return VideoStreamMeta::default();
|
||
};
|
||
|
||
let stream = &json["streams"][0];
|
||
let is_h264 = stream
|
||
.get("codec_name")
|
||
.and_then(|v| v.as_str())
|
||
.map(|s| s == "h264")
|
||
.unwrap_or(false);
|
||
|
||
// Prefer legacy `tags.rotate` (older containers); fall back to the
|
||
// display-matrix side data (iPhone and other modern recorders).
|
||
let rotation = stream
|
||
.get("tags")
|
||
.and_then(|t| t.get("rotate"))
|
||
.and_then(|r| r.as_str())
|
||
.and_then(|s| s.parse::<i32>().ok())
|
||
.filter(|r| *r != 0)
|
||
.or_else(|| {
|
||
stream
|
||
.get("side_data_list")
|
||
.and_then(|l| l.as_array())
|
||
.and_then(|arr| {
|
||
arr.iter()
|
||
.find_map(|sd| sd.get("rotation").and_then(|r| r.as_f64()))
|
||
})
|
||
.map(|f| f.abs() as i32)
|
||
.filter(|r| *r != 0)
|
||
})
|
||
.unwrap_or(0);
|
||
|
||
// ffprobe reports frame rates as rational strings like "30000/1001".
|
||
// Prefer avg_frame_rate (handles VFR) and fall back to r_frame_rate.
|
||
let frame_rate = stream
|
||
.get("avg_frame_rate")
|
||
.and_then(|v| v.as_str())
|
||
.and_then(parse_ffprobe_rational)
|
||
.or_else(|| {
|
||
stream
|
||
.get("r_frame_rate")
|
||
.and_then(|v| v.as_str())
|
||
.and_then(parse_ffprobe_rational)
|
||
});
|
||
|
||
debug!(
|
||
"Probed {}: codec_h264={}, rotation={}°, fps={:?}",
|
||
video_path, is_h264, rotation, frame_rate
|
||
);
|
||
|
||
VideoStreamMeta {
|
||
is_h264,
|
||
rotation,
|
||
frame_rate,
|
||
}
|
||
}
|
||
|
||
/// Probe the max keyframe interval (GOP) in the first ~30s of a video.
|
||
/// Returns `None` on probe failure or if we couldn't see at least two keyframes.
|
||
///
|
||
/// Used to decide between stream-copy and transcode: HLS needs segments to
|
||
/// start on keyframes, so if the source GOP exceeds `hls_time`, copying
|
||
/// produces oversized/glitchy segments and we need to re-encode.
|
||
async fn get_max_gop_seconds(video_path: &str) -> Option<f64> {
|
||
let output = tokio::process::Command::new("ffprobe")
|
||
.arg("-v")
|
||
.arg("error")
|
||
.arg("-select_streams")
|
||
.arg("v:0")
|
||
.arg("-skip_frame")
|
||
.arg("nokey")
|
||
.arg("-show_entries")
|
||
.arg("frame=pts_time")
|
||
.arg("-of")
|
||
.arg("csv=p=0")
|
||
.arg("-read_intervals")
|
||
.arg("%+30")
|
||
.arg(video_path)
|
||
.output()
|
||
.await
|
||
.ok()?;
|
||
|
||
if !output.status.success() {
|
||
warn!(
|
||
"ffprobe GOP check failed for {}: {}",
|
||
video_path,
|
||
String::from_utf8_lossy(&output.stderr).trim()
|
||
);
|
||
return None;
|
||
}
|
||
|
||
let times: Vec<f64> = String::from_utf8_lossy(&output.stdout)
|
||
.lines()
|
||
.filter_map(|l| l.trim().parse::<f64>().ok())
|
||
.collect();
|
||
|
||
if times.len() < 2 {
|
||
return None;
|
||
}
|
||
|
||
let max_gop = times
|
||
.windows(2)
|
||
.map(|w| w[1] - w[0])
|
||
.fold(0.0_f64, f64::max);
|
||
debug!(
|
||
"Max GOP in first {} keyframes of {}: {:.2}s",
|
||
times.len(),
|
||
video_path,
|
||
max_gop
|
||
);
|
||
Some(max_gop)
|
||
}
|
||
|
||
pub struct VideoPlaylistManager {
|
||
video_dir: PathBuf,
|
||
playlist_generator: Addr<PlaylistGenerator>,
|
||
}
|
||
|
||
impl VideoPlaylistManager {
|
||
pub fn new<P: Into<PathBuf>>(
|
||
video_dir: P,
|
||
playlist_generator: Addr<PlaylistGenerator>,
|
||
) -> Self {
|
||
Self {
|
||
video_dir: video_dir.into(),
|
||
playlist_generator,
|
||
}
|
||
}
|
||
}
|
||
|
||
impl Actor for VideoPlaylistManager {
|
||
type Context = Context<Self>;
|
||
}
|
||
|
||
impl Handler<QueueVideosMessage> for VideoPlaylistManager {
|
||
type Result = ();
|
||
|
||
fn handle(&mut self, msg: QueueVideosMessage, _ctx: &mut Self::Context) -> Self::Result {
|
||
if msg.videos.is_empty() {
|
||
return;
|
||
}
|
||
|
||
let video_dir = self.video_dir.clone();
|
||
let playlist_generator = self.playlist_generator.clone();
|
||
|
||
let mut queued = 0usize;
|
||
let mut already_present = 0usize;
|
||
for VideoToQueue {
|
||
video_path,
|
||
content_hash,
|
||
} in msg.videos
|
||
{
|
||
let playlist = hls_paths::playlist_for_hash(&video_dir, &content_hash);
|
||
let sentinel = hls_paths::sentinel_for_hash(&video_dir, &content_hash);
|
||
if playlist.exists() || sentinel.exists() {
|
||
already_present += 1;
|
||
continue;
|
||
}
|
||
debug!(
|
||
"Queueing playlist generation for {} (hash={})",
|
||
video_path.display(),
|
||
short_hash(&content_hash)
|
||
);
|
||
playlist_generator.do_send(GeneratePlaylistMessage {
|
||
video_path,
|
||
content_hash,
|
||
});
|
||
queued += 1;
|
||
}
|
||
info!(
|
||
"Queue tick: {} queued, {} skipped (playlist or sentinel already on disk)",
|
||
queued, already_present
|
||
);
|
||
}
|
||
}
|
||
|
||
#[derive(Message)]
|
||
#[rtype(result = "()")]
|
||
pub struct QueueVideosMessage {
|
||
pub videos: Vec<VideoToQueue>,
|
||
}
|
||
|
||
#[derive(Message)]
|
||
#[rtype(result = "Result<()>")]
|
||
pub struct GeneratePlaylistMessage {
|
||
pub video_path: PathBuf,
|
||
pub content_hash: String,
|
||
}
|
||
|
||
pub struct PlaylistGenerator {
|
||
semaphore: Arc<Semaphore>,
|
||
video_dir: PathBuf,
|
||
}
|
||
|
||
impl PlaylistGenerator {
|
||
pub(crate) fn new<P: Into<PathBuf>>(video_dir: P) -> Self {
|
||
// Concurrency is tunable via HLS_CONCURRENCY so operators can dial
|
||
// it to their hardware: 1 on weak Synology boxes to avoid thermal
|
||
// throttling, higher on desktops with spare cores.
|
||
let concurrency = std::env::var("HLS_CONCURRENCY")
|
||
.ok()
|
||
.and_then(|v| v.parse::<usize>().ok())
|
||
.filter(|&n| n > 0)
|
||
.unwrap_or(2);
|
||
info!("PlaylistGenerator: concurrency={}", concurrency);
|
||
PlaylistGenerator {
|
||
semaphore: Arc::new(Semaphore::new(concurrency)),
|
||
video_dir: video_dir.into(),
|
||
}
|
||
}
|
||
}
|
||
|
||
impl Actor for PlaylistGenerator {
|
||
type Context = Context<Self>;
|
||
}
|
||
|
||
impl Handler<GeneratePlaylistMessage> for PlaylistGenerator {
|
||
type Result = ResponseFuture<Result<()>>;
|
||
|
||
fn handle(&mut self, msg: GeneratePlaylistMessage, _ctx: &mut Self::Context) -> Self::Result {
|
||
let video_file = msg.video_path.to_str().unwrap().to_owned();
|
||
let content_hash_str = msg.content_hash.clone();
|
||
let semaphore = self.semaphore.clone();
|
||
let video_dir = self.video_dir.clone();
|
||
|
||
let hash_dir = content_hash::hls_dir(&video_dir, &content_hash_str);
|
||
let playlist_path = hls_paths::playlist_for_hash(&video_dir, &content_hash_str);
|
||
let sentinel_path = hls_paths::sentinel_for_hash(&video_dir, &content_hash_str);
|
||
let segment_template = hls_paths::segment_template_for_hash(&video_dir, &content_hash_str);
|
||
let playlist_file = playlist_path.to_string_lossy().to_string();
|
||
let segment_pattern = segment_template.to_string_lossy().to_string();
|
||
|
||
let tracer = global_tracer();
|
||
let mut span = tracer
|
||
.span_builder("playlistgenerator.generate_playlist")
|
||
.with_attributes(vec![
|
||
KeyValue::new("video_file", video_file.clone()),
|
||
KeyValue::new("content_hash", content_hash_str.clone()),
|
||
KeyValue::new("playlist_file", playlist_file.clone()),
|
||
])
|
||
.start(&tracer);
|
||
|
||
Box::pin(async move {
|
||
let wait_start = std::time::Instant::now();
|
||
let permit = semaphore
|
||
.acquire_owned()
|
||
.await
|
||
.expect("Unable to acquire semaphore");
|
||
|
||
debug!(
|
||
"Waited for {:?} before starting ffmpeg",
|
||
wait_start.elapsed()
|
||
);
|
||
span.add_event(
|
||
"Waited for FFMPEG semaphore",
|
||
vec![KeyValue::new(
|
||
"wait_time",
|
||
wait_start.elapsed().as_secs_f64(),
|
||
)],
|
||
);
|
||
|
||
if playlist_path.exists() {
|
||
debug!("Playlist already exists: {}", playlist_file);
|
||
span.set_status(Status::error(format!(
|
||
"Playlist already exists: {}",
|
||
playlist_file
|
||
)));
|
||
return Err(std::io::Error::from(std::io::ErrorKind::AlreadyExists));
|
||
}
|
||
|
||
// Ensure the shard + hash directory exist. Idempotent — the
|
||
// dir may already be present from a prior attempt that wrote
|
||
// a sentinel before being cleared for retry.
|
||
if let Err(e) = tokio::fs::create_dir_all(&hash_dir).await {
|
||
error!(
|
||
"Failed to create HLS hash dir {}: {}",
|
||
hash_dir.display(),
|
||
e
|
||
);
|
||
span.set_status(Status::error(format!("mkdir failed: {}", e)));
|
||
return Err(e);
|
||
}
|
||
|
||
// One ffprobe call for codec + rotation metadata.
|
||
let stream_meta = probe_video_stream_meta(&video_file).await;
|
||
let is_h264 = stream_meta.is_h264;
|
||
let rotation = stream_meta.rotation;
|
||
let has_rotation = rotation != 0;
|
||
|
||
// Stream-copy is only safe when the source GOP fits inside a
|
||
// single HLS segment. Otherwise ffmpeg has to extend segments
|
||
// past hls_time to land on a keyframe, producing uneven
|
||
// segments and seeking glitches.
|
||
const HLS_SEGMENT_SECONDS: f64 = 3.0;
|
||
let gop_ok = if is_h264 && !has_rotation {
|
||
match get_max_gop_seconds(&video_file).await {
|
||
Some(g) if g > HLS_SEGMENT_SECONDS => {
|
||
info!(
|
||
"Video {} has long GOP ({:.1}s > {}s), transcoding for segment alignment",
|
||
video_file, g, HLS_SEGMENT_SECONDS
|
||
);
|
||
false
|
||
}
|
||
Some(_) => true,
|
||
None => {
|
||
// Probe failed — be conservative and transcode rather
|
||
// than risk broken segments from a mystery source.
|
||
debug!(
|
||
"GOP probe failed for {}, transcoding to be safe",
|
||
video_file
|
||
);
|
||
false
|
||
}
|
||
}
|
||
} else {
|
||
false
|
||
};
|
||
|
||
let use_copy = is_h264 && !has_rotation && gop_ok;
|
||
|
||
if has_rotation {
|
||
info!(
|
||
"Video {} has rotation metadata ({}°), transcoding to apply rotation",
|
||
video_file, rotation
|
||
);
|
||
span.add_event(
|
||
"Transcoding due to rotation",
|
||
vec![KeyValue::new("rotation_degrees", rotation as i64)],
|
||
);
|
||
} else if use_copy {
|
||
info!("Video {} is already h264, using stream copy", video_file);
|
||
span.add_event("Using stream copy (h264 detected)", vec![]);
|
||
} else if is_h264 {
|
||
info!(
|
||
"Video {} is h264 but needs transcoding for GOP alignment",
|
||
video_file
|
||
);
|
||
span.add_event("Transcoding for GOP alignment", vec![]);
|
||
} else {
|
||
info!("Video {} needs transcoding to h264", video_file);
|
||
span.add_event("Transcoding to h264", vec![]);
|
||
}
|
||
|
||
// Encode to a .tmp playlist alongside the final inside the
|
||
// hash dir, so a concurrent scan never sees a half-written
|
||
// .m3u8 as "done". Segments use the hash-keyed template;
|
||
// ffmpeg writes them next to the playlist (relative refs).
|
||
let playlist_tmp = format!("{}.tmp", playlist_file);
|
||
|
||
let mut cmd = tokio::process::Command::new("ffmpeg");
|
||
cmd.arg("-y").arg("-i").arg(&video_file);
|
||
|
||
if use_copy {
|
||
cmd.arg("-c:v").arg("copy");
|
||
cmd.arg("-c:a").arg("aac");
|
||
} else {
|
||
let nvenc = crate::video::ffmpeg::is_nvenc_available().await;
|
||
if nvenc {
|
||
// NVENC: no CRF, use VBR + target CQ. p1 = fastest
|
||
// preset — prioritizes encoder throughput over bitrate
|
||
// efficiency. CQ 23 roughly matches libx264 crf 21
|
||
// visually; NVENC has slightly lower compression
|
||
// efficiency per quality.
|
||
cmd.arg("-c:v").arg("h264_nvenc");
|
||
cmd.arg("-preset").arg("p1");
|
||
cmd.arg("-rc").arg("vbr");
|
||
cmd.arg("-cq").arg("23");
|
||
cmd.arg("-pix_fmt").arg("yuv420p");
|
||
} else {
|
||
cmd.arg("-c:v").arg("h264");
|
||
cmd.arg("-crf").arg("21");
|
||
cmd.arg("-preset").arg("veryfast");
|
||
}
|
||
cmd.arg("-vf").arg("scale='min(1080,iw)':-2,setsar=1:1");
|
||
cmd.arg("-c:a").arg("aac");
|
||
// Force an IDR frame every hls_time seconds so each HLS
|
||
// segment starts on a keyframe — accurate seeking without
|
||
// players having to decode from a prior segment.
|
||
cmd.arg("-force_key_frames").arg("expr:gte(t,n_forced*3)");
|
||
}
|
||
|
||
// -f hls is required because the playlist is written to a .tmp
|
||
// path during encoding — ffmpeg normally infers the muxer from
|
||
// the output extension and doesn't recognize ".m3u8.tmp".
|
||
cmd.arg("-f").arg("hls");
|
||
cmd.arg("-hls_time").arg("3");
|
||
cmd.arg("-hls_list_size").arg("0");
|
||
cmd.arg("-hls_playlist_type").arg("vod");
|
||
// independent_segments advertises that each segment can be
|
||
// decoded without reference to any other — the matching guarantee
|
||
// for the forced keyframes above.
|
||
cmd.arg("-hls_flags").arg("independent_segments");
|
||
cmd.arg("-hls_segment_filename").arg(&segment_pattern);
|
||
cmd.arg(&playlist_tmp);
|
||
cmd.stdout(Stdio::null());
|
||
cmd.stderr(Stdio::piped());
|
||
cmd.kill_on_drop(true);
|
||
|
||
// Spawn + wait under a timeout so a hung ffmpeg (corrupt source,
|
||
// NFS stall, etc.) doesn't permanently hold a semaphore slot.
|
||
// Default is generous — a long 4K transcode on CPU can take hours.
|
||
let timeout_secs = std::env::var("HLS_TIMEOUT_SECONDS")
|
||
.ok()
|
||
.and_then(|v| v.parse::<u64>().ok())
|
||
.unwrap_or(7200);
|
||
|
||
let ffmpeg_result = match cmd.spawn() {
|
||
Ok(child) => {
|
||
match tokio::time::timeout(
|
||
std::time::Duration::from_secs(timeout_secs),
|
||
child.wait_with_output(),
|
||
)
|
||
.await
|
||
{
|
||
Ok(res) => res
|
||
.inspect_err(|e| {
|
||
error!("Failed to wait on ffmpeg child process: {}", e)
|
||
})
|
||
.map_err(|e| std::io::Error::other(e.to_string())),
|
||
Err(_) => Err(std::io::Error::other(format!(
|
||
"ffmpeg exceeded {}s timeout",
|
||
timeout_secs
|
||
))),
|
||
}
|
||
}
|
||
Err(e) => {
|
||
error!("Failed to spawn ffmpeg: {}", e);
|
||
Err(std::io::Error::other(e.to_string()))
|
||
}
|
||
};
|
||
|
||
drop(permit);
|
||
|
||
let success = matches!(&ffmpeg_result, Ok(out) if out.status.success());
|
||
|
||
if success {
|
||
if let Err(e) = tokio::fs::rename(&playlist_tmp, &playlist_path).await {
|
||
error!(
|
||
"ffmpeg succeeded but rename {} -> {} failed: {}",
|
||
playlist_tmp, playlist_file, e
|
||
);
|
||
cleanup_partial_hls(&hash_dir).await;
|
||
span.set_status(Status::error(format!("rename failed: {}", e)));
|
||
return Err(e);
|
||
}
|
||
debug!("Playlist complete: {}", playlist_file);
|
||
span.set_status(Status::Ok);
|
||
Ok(())
|
||
} else {
|
||
let detail = match &ffmpeg_result {
|
||
Ok(out) => format!(
|
||
"exit {}: {}",
|
||
out.status,
|
||
String::from_utf8_lossy(&out.stderr).trim()
|
||
),
|
||
Err(e) => format!("ffmpeg failed: {}", e),
|
||
};
|
||
error!("ffmpeg failed for {}: {}", video_file, detail);
|
||
cleanup_partial_hls(&hash_dir).await;
|
||
if let Err(se) = tokio::fs::write(&sentinel_path, b"").await {
|
||
warn!(
|
||
"Failed to write playlist sentinel {}: {}",
|
||
sentinel_path.display(),
|
||
se
|
||
);
|
||
} else {
|
||
info!(
|
||
"Wrote playlist sentinel {} so future scans skip {}",
|
||
sentinel_path.display(),
|
||
video_file
|
||
);
|
||
}
|
||
span.set_status(Status::error(detail.clone()));
|
||
Err(std::io::Error::other(detail))
|
||
}
|
||
})
|
||
}
|
||
}
|
||
|
||
/// Delete the partial playlist (.tmp) and any segment files left behind by
|
||
/// a failed ffmpeg run. Wipes every non-sentinel file in the hash dir;
|
||
/// retains the sentinel if one has already been written by an earlier
|
||
/// caller in the same path (today there is none, but kept defensively so
|
||
/// the function is safe to call after sentinel write too).
|
||
async fn cleanup_partial_hls(hash_dir: &Path) {
|
||
let Ok(mut entries) = tokio::fs::read_dir(hash_dir).await else {
|
||
return;
|
||
};
|
||
while let Ok(Some(entry)) = entries.next_entry().await {
|
||
let path = entry.path();
|
||
let is_sentinel = path
|
||
.file_name()
|
||
.and_then(|n| n.to_str())
|
||
.map(|n| n == hls_paths::UNSUPPORTED_SENTINEL_FILENAME)
|
||
.unwrap_or(false);
|
||
if is_sentinel {
|
||
continue;
|
||
}
|
||
if let Err(e) = tokio::fs::remove_file(&path).await {
|
||
warn!(
|
||
"Failed to remove partial HLS file {}: {}",
|
||
path.display(),
|
||
e
|
||
);
|
||
}
|
||
}
|
||
}
|
||
|
||
/// First 16 chars of a content hash for log lines. Short enough to keep
|
||
/// log volume sane, long enough that distinct hashes don't collide in
|
||
/// practice.
|
||
fn short_hash(hash: &str) -> &str {
|
||
let end = hash
|
||
.char_indices()
|
||
.nth(16)
|
||
.map(|(i, _)| i)
|
||
.unwrap_or(hash.len());
|
||
&hash[..end]
|
||
}
|
||
|
||
#[derive(Message)]
|
||
#[rtype(result = "()")]
|
||
pub struct GeneratePreviewClipMessage {
|
||
pub video_path: String,
|
||
}
|
||
|
||
pub struct PreviewClipGenerator {
|
||
semaphore: Arc<Semaphore>,
|
||
preview_clips_dir: String,
|
||
libraries: Vec<Library>,
|
||
preview_dao: Arc<Mutex<Box<dyn PreviewDao>>>,
|
||
}
|
||
|
||
impl PreviewClipGenerator {
|
||
pub fn new(
|
||
preview_clips_dir: String,
|
||
libraries: Vec<Library>,
|
||
preview_dao: Arc<Mutex<Box<dyn PreviewDao>>>,
|
||
) -> Self {
|
||
PreviewClipGenerator {
|
||
semaphore: Arc::new(Semaphore::new(2)),
|
||
preview_clips_dir,
|
||
libraries,
|
||
preview_dao,
|
||
}
|
||
}
|
||
|
||
/// Strip whichever library root actually contains `video_path`.
|
||
/// Falls back to the first library if none match, so we never
|
||
/// accidentally emit the absolute input path as the output path
|
||
/// (which ffmpeg rejects as "cannot edit existing files in place").
|
||
fn relativize(&self, video_path: &str) -> String {
|
||
for lib in &self.libraries {
|
||
if let Some(stripped) = video_path.strip_prefix(&lib.root_path) {
|
||
return stripped.trim_start_matches(['/', '\\']).replace('\\', "/");
|
||
}
|
||
}
|
||
video_path
|
||
.trim_start_matches(['/', '\\'])
|
||
.replace('\\', "/")
|
||
}
|
||
}
|
||
|
||
impl Actor for PreviewClipGenerator {
|
||
type Context = Context<Self>;
|
||
}
|
||
|
||
impl Handler<GeneratePreviewClipMessage> for PreviewClipGenerator {
|
||
type Result = ResponseFuture<()>;
|
||
|
||
fn handle(
|
||
&mut self,
|
||
msg: GeneratePreviewClipMessage,
|
||
_ctx: &mut Self::Context,
|
||
) -> Self::Result {
|
||
let semaphore = self.semaphore.clone();
|
||
let preview_clips_dir = self.preview_clips_dir.clone();
|
||
let preview_dao = self.preview_dao.clone();
|
||
let video_path = msg.video_path;
|
||
// Resolve against whichever library actually owns this video.
|
||
let relative_path = self.relativize(&video_path);
|
||
|
||
Box::pin(async move {
|
||
let permit = semaphore
|
||
.acquire_owned()
|
||
.await
|
||
.expect("Unable to acquire preview semaphore");
|
||
|
||
// Update status to processing
|
||
{
|
||
let otel_ctx = opentelemetry::Context::current();
|
||
let mut dao = preview_dao.lock().expect("Unable to lock PreviewDao");
|
||
let _ =
|
||
dao.update_status(&otel_ctx, &relative_path, "processing", None, None, None);
|
||
}
|
||
|
||
// Compute output path: join preview_clips_dir with relative path, change ext to .mp4
|
||
let output_path = PathBuf::from(&preview_clips_dir)
|
||
.join(&relative_path)
|
||
.with_extension("mp4");
|
||
let output_str = output_path.to_string_lossy().to_string();
|
||
|
||
let video_path_owned = video_path.clone();
|
||
let relative_path_owned = relative_path.clone();
|
||
tokio::spawn(async move {
|
||
match generate_preview_clip(&video_path_owned, &output_str).await {
|
||
Ok((duration, size)) => {
|
||
info!(
|
||
"Preview clip complete for '{}' ({:.1}s, {} bytes)",
|
||
relative_path_owned, duration, size
|
||
);
|
||
let otel_ctx = opentelemetry::Context::current();
|
||
let mut dao = preview_dao.lock().expect("Unable to lock PreviewDao");
|
||
let _ = dao.update_status(
|
||
&otel_ctx,
|
||
&relative_path_owned,
|
||
"complete",
|
||
Some(duration as f32),
|
||
Some(size as i32),
|
||
None,
|
||
);
|
||
}
|
||
Err(e) => {
|
||
error!(
|
||
"Failed to generate preview clip for '{}': {}",
|
||
relative_path_owned, e
|
||
);
|
||
let otel_ctx = opentelemetry::Context::current();
|
||
let mut dao = preview_dao.lock().expect("Unable to lock PreviewDao");
|
||
let _ = dao.update_status(
|
||
&otel_ctx,
|
||
&relative_path_owned,
|
||
"failed",
|
||
None,
|
||
None,
|
||
Some(&e.to_string()),
|
||
);
|
||
}
|
||
}
|
||
drop(permit);
|
||
});
|
||
})
|
||
}
|
||
}
|
||
|
||
#[cfg(test)]
|
||
mod tests {
|
||
use super::parse_ffprobe_rational;
|
||
|
||
#[test]
|
||
fn parses_common_rational_framerates() {
|
||
// NTSC 29.97 fps
|
||
assert!((parse_ffprobe_rational("30000/1001").unwrap() - 29.970_03).abs() < 1e-3);
|
||
// Plain integer fps
|
||
assert!((parse_ffprobe_rational("30/1").unwrap() - 30.0).abs() < 1e-6);
|
||
assert!((parse_ffprobe_rational("60/1").unwrap() - 60.0).abs() < 1e-6);
|
||
// iPhone slow-mo
|
||
assert!((parse_ffprobe_rational("240/1").unwrap() - 240.0).abs() < 1e-6);
|
||
}
|
||
|
||
#[test]
|
||
fn rejects_ffprobe_unknown_sentinel() {
|
||
// 0/0 is ffprobe's way of saying "I don't know" — must not be
|
||
// interpreted as 0 fps.
|
||
assert_eq!(parse_ffprobe_rational("0/0"), None);
|
||
}
|
||
|
||
#[test]
|
||
fn rejects_malformed_input() {
|
||
assert_eq!(parse_ffprobe_rational(""), None);
|
||
assert_eq!(parse_ffprobe_rational("30"), None);
|
||
assert_eq!(parse_ffprobe_rational("/1"), None);
|
||
assert_eq!(parse_ffprobe_rational("30/"), None);
|
||
assert_eq!(parse_ffprobe_rational("abc/def"), None);
|
||
}
|
||
|
||
#[test]
|
||
fn rejects_non_positive_results() {
|
||
// Negative numerator -> negative fps; meaningless.
|
||
assert_eq!(parse_ffprobe_rational("-30/1"), None);
|
||
// Zero numerator -> zero fps; also meaningless for frame stepping.
|
||
assert_eq!(parse_ffprobe_rational("0/1"), None);
|
||
}
|
||
|
||
#[test]
|
||
fn rejects_out_of_range() {
|
||
// Anything > 1000 fps is almost certainly garbage probe output,
|
||
// not a real source. (Real high-speed capture maxes near 1 kHz.)
|
||
assert_eq!(parse_ffprobe_rational("999999/1"), None);
|
||
}
|
||
}
|