use crate::plugins::external_plugin_gui_compatible;
use crate::video::{PlaybackBackend, PlaybackPhase, PlaybackStatus, VideoFrameSink};
use eframe::egui;
use guth_media::{PlayerCommand, PlayerEvent};
use std::fs::File;
use std::io::{self, BufReader, Read};
use std::os::fd::{AsFd, AsRawFd};
use std::os::unix::fs::MetadataExt;
use std::os::unix::process::CommandExt;
use std::path::Path;
use std::process::{Child, ChildStdin, Command, Stdio};
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, Mutex};
use std::thread::{self, JoinHandle};
use std::time::{Duration, Instant};
const PACKET_TIMEOUT: Duration = Duration::from_secs(5);
const PLAYBACK_TIMEOUT: Duration = Duration::from_secs(15);
const READ_POLL_TIMEOUT: rustix::event::Timespec = rustix::event::Timespec {
tv_sec: 0,
tv_nsec: 10_000_000,
};
struct PlaybackReader<R> {
input: BufReader<R>,
cancelled: Arc<AtomicBool>,
packet_started: Option<Instant>,
packet_timeout: Duration,
}
impl<R: Read + AsFd> PlaybackReader<R> {
fn new(input: R, cancelled: Arc<AtomicBool>) -> Self {
Self {
input: BufReader::new(input),
cancelled,
packet_started: None,
packet_timeout: PACKET_TIMEOUT,
}
}
fn next_event(&mut self, max_side: usize) -> io::Result<PlayerEvent> {
self.packet_started = None;
guth_media::read_event(self, max_side)
}
}
impl<R: Read + AsFd> Read for PlaybackReader<R> {
fn read(&mut self, bytes: &mut [u8]) -> io::Result<usize> {
if bytes.is_empty() {
return Ok(0);
}
loop {
if self.cancelled.load(Ordering::Acquire) {
return Err(io::Error::new(
io::ErrorKind::ConnectionAborted,
"playback cancelled",
));
}
if self
.packet_started
.is_some_and(|started| started.elapsed() >= self.packet_timeout)
{
return Err(io::Error::new(
io::ErrorKind::TimedOut,
"guth-media did not finish a playback packet in time",
));
}
match self.input.read(bytes) {
Ok(read) => {
if read > 0 {
self.packet_started.get_or_insert_with(Instant::now);
}
return Ok(read);
}
Err(error) if error.kind() == io::ErrorKind::Interrupted => continue,
Err(error) if error.kind() == io::ErrorKind::WouldBlock => {
let mut descriptors = [rustix::event::PollFd::new(
self.input.get_ref(),
rustix::event::PollFlags::IN,
)];
match rustix::event::poll(&mut descriptors, Some(&READ_POLL_TIMEOUT)) {
Ok(_) | Err(rustix::io::Errno::INTR) => {}
Err(error) => return Err(error.into()),
}
}
Err(error) => return Err(error),
}
}
}
}
struct PlaybackSnapshot {
status: PlaybackStatus,
last_event: Instant,
}
impl Default for PlaybackSnapshot {
fn default() -> Self {
Self {
status: PlaybackStatus::default(),
last_event: Instant::now(),
}
}
}
impl PlaybackSnapshot {
fn check_activity(&mut self, now: Instant) {
if matches!(
self.status.phase,
PlaybackPhase::Loading | PlaybackPhase::Playing
) && now.saturating_duration_since(self.last_event) >= PLAYBACK_TIMEOUT
{
self.status.phase = PlaybackPhase::Failed;
self.status.error = Some(
"guth-media stopped responding. Retry playback or disable the plugin in Settings."
.into(),
);
}
}
fn output_closed(&mut self, error: &io::Error, packet_started: bool) {
let clean_end = self.status.phase == PlaybackPhase::Ended
&& error.kind() == io::ErrorKind::UnexpectedEof
&& !packet_started;
if self.status.phase != PlaybackPhase::Failed && !clean_end {
self.status.phase = PlaybackPhase::Failed;
self.status.error = Some(format!(
"guth-media stopped or sent invalid playback data: {error}"
));
}
}
}
#[derive(Default)]
struct ProcessBackend {
child: Option<Child>,
input: Option<ChildStdin>,
reader: Option<JoinHandle<()>>,
status: Arc<Mutex<PlaybackSnapshot>>,
reader_cancelled: Arc<AtomicBool>,
error: Option<String>,
}
pub fn create() -> Result<Box<dyn PlaybackBackend>, String> {
Ok(Box::<ProcessBackend>::default())
}
fn verified_executable(path: &Path) -> io::Result<File> {
let file = File::from(rustix::fs::open(
path,
rustix::fs::OFlags::RDONLY
| rustix::fs::OFlags::NOFOLLOW
| rustix::fs::OFlags::NONBLOCK
| rustix::fs::OFlags::CLOEXEC,
rustix::fs::Mode::empty(),
)?);
let metadata = file.metadata()?;
if !metadata.is_file()
|| (metadata.uid() != rustix::process::geteuid().as_raw() && metadata.uid() != 0)
|| metadata.mode() & 0o7022 != 0
|| metadata.mode() & 0o111 == 0
{
return Err(io::Error::new(
io::ErrorKind::PermissionDenied,
"untrusted guth-media executable",
));
}
Ok(file)
}
impl ProcessBackend {
fn launch(
&mut self,
program: &Path,
path: &Path,
frames: VideoFrameSink,
volume: u8,
) -> Result<(), String> {
self.stop();
self.error = None;
let executable = verified_executable(program).map_err(|e| e.to_string())?;
rustix::io::fcntl_setfd(&executable, rustix::io::FdFlags::empty())
.map_err(|e| e.to_string())?;
let child = Command::new(format!("/proc/self/fd/{}", executable.as_raw_fd()))
.args([guth_media::PROTOCOL_ARGUMENT, guth_media::PROTOCOL_VERSION])
.arg(frames.max_texture_side.to_string())
.arg(volume.min(100).to_string())
.arg(path)
.process_group(0)
.stdin(Stdio::piped())
.stdout(Stdio::piped())
.stderr(Stdio::null())
.spawn()
.map_err(|e| format!("Cannot start guth-media: {e}"))?;
self.child = Some(child);
let child = self.child.as_mut().unwrap();
self.input = child.stdin.take();
let stdout = child
.stdout
.take()
.ok_or("Missing guth-media output pipe")?;
rustix::fs::fcntl_setfl(&stdout, rustix::fs::OFlags::NONBLOCK)
.map_err(|e| e.to_string())?;
let input = self
.input
.as_ref()
.ok_or("Missing guth-media control pipe")?;
rustix::fs::fcntl_setfl(input, rustix::fs::OFlags::NONBLOCK).map_err(|e| e.to_string())?;
self.status = Arc::new(Mutex::new(PlaybackSnapshot {
status: PlaybackStatus {
phase: PlaybackPhase::Loading,
..Default::default()
},
..Default::default()
}));
self.reader_cancelled = Arc::new(AtomicBool::new(false));
let cancelled = Arc::clone(&self.reader_cancelled);
let snapshot = Arc::clone(&self.status);
self.reader = Some(
thread::Builder::new()
.name("guth-media-frames".into())
.spawn(move || {
let mut reader = PlaybackReader::new(stdout, cancelled);
loop {
match reader.next_event(frames.max_texture_side) {
Ok(PlayerEvent::Status(status)) => {
let mut snapshot =
snapshot.lock().unwrap_or_else(|e| e.into_inner());
snapshot.status = status;
snapshot.last_event = Instant::now();
}
Ok(PlayerEvent::Frame(frame)) => {
if !frames.is_current() {
break;
}
frames.publish(egui::ColorImage::from_rgba_unmultiplied(
frame.size,
&frame.rgba,
));
snapshot
.lock()
.unwrap_or_else(|e| e.into_inner())
.last_event = Instant::now();
}
Err(error) => {
let mut snapshot =
snapshot.lock().unwrap_or_else(|e| e.into_inner());
if frames.is_current() && !reader.cancelled.load(Ordering::Acquire)
{
snapshot.output_closed(&error, reader.packet_started.is_some());
}
break;
}
}
}
})
.map_err(|e| e.to_string())?,
);
Ok(())
}
fn send(&mut self, command: PlayerCommand) {
if let Some(input) = &mut self.input {
if let Err(error) = guth_media::write_command(input, command) {
self.error = Some(format!("guth-media is not accepting controls: {error}"));
}
}
}
}
impl PlaybackBackend for ProcessBackend {
fn load(&mut self, path: &Path, frames: VideoFrameSink, volume: u8) -> Result<(), String> {
let enabled = guth_cli::load_enabled_plugins().map_err(|e| e.to_string())?;
let plugins = guth_cli::discover_plugins().map_err(|e| e.to_string())?;
let plugin = plugins
.iter()
.find(|plugin| {
enabled.contains(&plugin.id)
&& external_plugin_gui_compatible(
plugin,
guth_media::PLUGIN_ID,
guth_media::CAPABILITY,
)
})
.ok_or("Install and enable guth-media in Settings → Plugins to play video inline.")?;
self.launch(&plugin.executable, path, frames, volume)
.inspect_err(|_| self.stop())
}
fn status(&mut self) -> PlaybackStatus {
if let Some(error) = &self.error {
return PlaybackStatus {
phase: PlaybackPhase::Failed,
error: Some(error.clone()),
..Default::default()
};
}
let mut snapshot = self.status.lock().unwrap_or_else(|e| e.into_inner());
snapshot.check_activity(Instant::now());
snapshot.status.clone()
}
fn set_paused(&mut self, paused: bool) {
self.send(PlayerCommand::Pause(paused));
}
fn set_volume(&mut self, volume: u8) {
self.send(PlayerCommand::Volume(volume.min(100)));
}
fn seek(&mut self, milliseconds: u64) {
self.send(PlayerCommand::Seek(milliseconds));
}
fn stop(&mut self) {
self.reader_cancelled.store(true, Ordering::Release);
self.input = None;
if let Some(mut child) = self.child.take() {
if let Some(group) = i32::try_from(child.id())
.ok()
.and_then(rustix::process::Pid::from_raw)
{
let _ = rustix::process::kill_process_group(group, rustix::process::Signal::TERM);
let deadline = Instant::now() + Duration::from_millis(250);
while matches!(child.try_wait(), Ok(None)) && Instant::now() < deadline {
thread::sleep(Duration::from_millis(10));
}
let _ = rustix::process::kill_process_group(group, rustix::process::Signal::KILL);
}
let _ = child.kill();
let _ = child.wait();
}
if let Some(reader) = self.reader.take() {
let _ = reader.join();
}
}
}
impl Drop for ProcessBackend {
fn drop(&mut self) {
self.stop();
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::video::VideoPlayer;
use std::io::Write;
fn test_pipe() -> (File, File) {
let (input, output) = rustix::pipe::pipe().unwrap();
rustix::fs::fcntl_setfl(&input, rustix::fs::OFlags::NONBLOCK).unwrap();
(File::from(input), File::from(output))
}
#[test]
fn stopping_interrupts_idle_and_partial_packets_with_an_open_writer() {
for partial in [false, true] {
let (input, mut held_writer) = test_pipe();
if partial {
held_writer.write_all(&[1]).unwrap();
}
let mut backend = ProcessBackend::default();
let cancelled = Arc::clone(&backend.reader_cancelled);
backend.reader = Some(thread::spawn(move || {
let mut reader = PlaybackReader::new(input, cancelled);
let error = reader.next_event(8192).err().expect("cancel read");
assert_eq!(error.kind(), io::ErrorKind::ConnectionAborted);
}));
let (finished, wait) = std::sync::mpsc::channel();
let worker = thread::spawn(move || {
backend.stop();
finished.send(()).unwrap();
});
let stopped = wait.recv_timeout(Duration::from_secs(1)).is_ok();
drop(held_writer);
worker.join().unwrap();
assert!(stopped, "stop depended on the output writer closing");
}
}
#[test]
fn stalled_partial_packet_times_out_without_waiting_for_eof() {
let (input, mut writer) = test_pipe();
writer.write_all(&[1, 2, 0, 0, 0, b'{']).unwrap();
let mut reader = PlaybackReader::new(input, Arc::default());
reader.packet_timeout = Duration::from_millis(30);
let error = reader.next_event(8192).err().expect("incomplete packet");
assert_eq!(error.kind(), io::ErrorKind::TimedOut);
}
#[test]
fn deadline_also_covers_a_prefetched_partial_packet() {
let (input, mut writer) = test_pipe();
let mut bytes = Vec::new();
guth_media::write_status(&mut bytes, &PlaybackStatus::default()).unwrap();
bytes.push(1);
writer.write_all(&bytes).unwrap();
let mut reader = PlaybackReader::new(input, Arc::default());
reader.packet_timeout = Duration::from_millis(30);
assert!(matches!(
reader.next_event(8192),
Ok(PlayerEvent::Status(_))
));
assert_eq!(
reader.next_event(8192).err().unwrap().kind(),
io::ErrorKind::TimedOut
);
}
#[test]
fn quiet_between_packets_does_not_consume_the_packet_deadline() {
let (input, mut writer) = test_pipe();
let writer = thread::spawn(move || {
thread::sleep(Duration::from_millis(50));
guth_media::write_status(
&mut writer,
&PlaybackStatus {
phase: PlaybackPhase::Paused,
..Default::default()
},
)
.unwrap();
});
let mut reader = PlaybackReader::new(input, Arc::default());
reader.packet_timeout = Duration::from_millis(30);
assert!(
matches!(reader.next_event(8192), Ok(PlayerEvent::Status(status)) if status.phase == PlaybackPhase::Paused)
);
writer.join().unwrap();
}
#[test]
fn full_resolution_frame_streams_across_many_pipe_capacity_chunks() {
let (input, mut output) = test_pipe();
let producer = thread::spawn(move || {
guth_media::write_frame(
&mut output,
&guth_media::VideoFrame {
size: [1920, 1080],
rgba: vec![17; 1920 * 1080 * 4],
},
)
});
let mut reader = PlaybackReader::new(input, Arc::default());
let event = reader.next_event(8192);
drop(reader);
let written = producer.join().unwrap();
let PlayerEvent::Frame(frame) = event.expect("read full frame within packet deadline")
else {
panic!("expected frame")
};
written.unwrap();
assert_eq!(frame.size, [1920, 1080]);
assert_eq!(frame.rgba.len(), 1920 * 1080 * 4);
assert!(frame.rgba.iter().all(|byte| *byte == 17));
}
#[test]
fn output_eof_preserves_clean_completion_but_not_truncated_packets() {
let eof = io::Error::from(io::ErrorKind::UnexpectedEof);
let mut snapshot = PlaybackSnapshot::default();
snapshot.status.phase = PlaybackPhase::Ended;
snapshot.output_closed(&eof, false);
assert_eq!(snapshot.status.phase, PlaybackPhase::Ended);
snapshot.output_closed(&eof, true);
assert_eq!(snapshot.status.phase, PlaybackPhase::Failed);
snapshot.status.error = Some("specific codec failure".into());
snapshot.output_closed(&eof, false);
assert_eq!(
snapshot.status.error.as_deref(),
Some("specific codec failure")
);
snapshot.status = PlaybackStatus {
phase: PlaybackPhase::Playing,
..Default::default()
};
snapshot.output_closed(&eof, false);
assert_eq!(snapshot.status.phase, PlaybackPhase::Failed);
}
#[test]
fn activity_watchdog_distinguishes_stalled_playback_from_paused_or_finished() {
let last_event = Instant::now();
for phase in [
PlaybackPhase::Idle,
PlaybackPhase::Loading,
PlaybackPhase::Playing,
PlaybackPhase::Paused,
PlaybackPhase::Ended,
PlaybackPhase::Failed,
] {
let mut snapshot = PlaybackSnapshot {
status: PlaybackStatus {
phase,
..Default::default()
},
last_event,
};
snapshot.check_activity(last_event + PLAYBACK_TIMEOUT - Duration::from_millis(1));
assert_eq!(snapshot.status.phase, phase);
snapshot.check_activity(last_event + PLAYBACK_TIMEOUT);
if matches!(phase, PlaybackPhase::Loading | PlaybackPhase::Playing) {
assert_eq!(snapshot.status.phase, PlaybackPhase::Failed);
assert!(snapshot
.status
.error
.as_deref()
.unwrap()
.contains("stopped responding"));
} else {
assert_eq!(snapshot.status.phase, phase);
}
}
}
#[test]
fn plugin_dispatch_rejects_unsafe_permissions_symlinks_and_special_files() {
use std::os::unix::fs::{symlink, PermissionsExt};
let directory = std::env::temp_dir().join(format!(
"guth-media-dispatch-{}-{}",
std::process::id(),
std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap()
.as_nanos()
));
std::fs::create_dir(&directory).unwrap();
let executable = directory.join("player");
std::fs::write(&executable, b"test executable placeholder").unwrap();
for mode in [0o600, 0o722, 0o4700] {
std::fs::set_permissions(&executable, std::fs::Permissions::from_mode(mode)).unwrap();
assert!(verified_executable(&executable).is_err());
}
std::fs::set_permissions(&executable, std::fs::Permissions::from_mode(0o700)).unwrap();
assert!(verified_executable(&executable).is_ok());
let link = directory.join("link");
symlink(&executable, &link).unwrap();
assert!(verified_executable(&link).is_err());
assert!(verified_executable(&directory).is_err());
let fifo = directory.join("fifo");
rustix::fs::mknodat(
rustix::fs::CWD,
&fifo,
rustix::fs::FileType::Fifo,
rustix::fs::Mode::RWXU,
0,
)
.unwrap();
assert!(verified_executable(&fifo).is_err());
std::fs::remove_file(link).unwrap();
std::fs::remove_file(fifo).unwrap();
std::fs::remove_file(executable).unwrap();
std::fs::remove_dir(directory).unwrap();
}
#[test]
#[ignore = "requires an installed/enabled guth-media plugin, VLC 3 codecs, and GUTH_VIDEO_TEST_FILE (1920x1080, at least 6 seconds)"]
fn native_playback_decodes_pauses_seeks_replays_and_stops() {
let path = std::env::var_os("GUTH_VIDEO_TEST_FILE").expect("set GUTH_VIDEO_TEST_FILE");
let ctx = egui::Context::default();
let mut video = VideoPlayer::with_backend(create);
video.muted = true;
video.start(Path::new(&path), &ctx);
fn wait(
video: &mut VideoPlayer,
ctx: &egui::Context,
predicate: impl Fn(&VideoPlayer) -> bool,
) {
let started = Instant::now();
loop {
video.poll(ctx);
assert_ne!(
video.status.phase,
PlaybackPhase::Failed,
"{:?}",
video.status.error
);
if predicate(video) {
return;
}
assert!(
started.elapsed() < Duration::from_secs(15),
"timed out: {:?}, dimensions {:?}",
video.status,
video.dimensions
);
std::thread::sleep(Duration::from_millis(20));
}
}
wait(&mut video, &ctx, |v| {
v.dimensions == Some([1920, 1080]) && v.status.position_ms > 300
});
assert_eq!(video.texture().unwrap().size(), [1920, 1080]);
assert!(video.status.seekable);
video.toggle(&ctx);
wait(&mut video, &ctx, |v| {
v.status.phase == PlaybackPhase::Paused
});
let paused_at = video.status.position_ms;
std::thread::sleep(Duration::from_millis(200));
video.poll(&ctx);
assert!(video.status.position_ms.abs_diff(paused_at) < 100);
let frames_before_seek = video.presented_frames;
video.seek(3_000);
wait(&mut video, &ctx, |v| {
v.status.position_ms >= 2_900 && v.presented_frames > frames_before_seek
});
assert_eq!(video.status.phase, PlaybackPhase::Paused);
std::thread::sleep(Duration::from_millis(500));
video.poll(&ctx);
assert!(
video.status.position_ms.abs_diff(3_000) < 150,
"Paused seek timecode drifted: {:?}",
video.status
);
video.toggle(&ctx);
wait(&mut video, &ctx, |v| {
v.status.phase == PlaybackPhase::Playing
});
video.seek(video.status.duration_ms.saturating_sub(300));
wait(&mut video, &ctx, |v| v.status.phase == PlaybackPhase::Ended);
video.toggle(&ctx);
wait(&mut video, &ctx, |v| {
v.status.phase == PlaybackPhase::Playing && v.dimensions == Some([1920, 1080])
});
assert!(video.status.position_ms < 2_000);
video.stop();
assert!(!video.has_source());
assert!(video.texture().is_none());
assert_eq!(video.status.phase, PlaybackPhase::Idle);
}
}