From 5d3da8b0067abb98b945031d7419af3133084e2a Mon Sep 17 00:00:00 2001 From: Mollusk Date: Sat, 15 Aug 2026 15:30:43 -0400 Subject: [PATCH] fix(host): contain capture owners and fail closed Bound the libpipewire router shutdown without detaching its OS handle, contain GStreamer and pactl children in parent-bound process groups, and make ownership failures terminal through the capture supervisor.\n\nAdd focused lifecycle tests plus a serialized live router teardown gate. --- src/common/contained.rs | 188 ++++++++++++++++++++++++++ src/common/mod.rs | 1 + src/host/audio.rs | 284 ++++++++++++++++++++++++++++++++++----- src/host/capture.rs | 8 +- src/host/health.rs | 83 ++++++++++++ src/host/mod.rs | 100 +++++++++++++- src/host/owned_thread.rs | 170 +++++++++++++++++++++++ src/host/pipeline.rs | 223 ++++++++++++++++++++++++++---- src/host/serve.rs | 58 +++++++- src/host/wayland.rs | 8 +- src/host/x11.rs | 9 +- 11 files changed, 1060 insertions(+), 72 deletions(-) create mode 100644 src/common/contained.rs create mode 100644 src/host/health.rs create mode 100644 src/host/owned_thread.rs diff --git a/src/common/contained.rs b/src/common/contained.rs new file mode 100644 index 0000000..bf936a4 --- /dev/null +++ b/src/common/contained.rs @@ -0,0 +1,188 @@ +//! Linux child-process containment for capture-side helpers. +//! +//! PixelPass owns `gst-launch-1.0` and the short-lived `pactl load-module` +//! workers. They must not outlive a host that is killed or fail-stops: GStreamer +//! can retain a portal/PipeWire capture resource, and a late pactl mutation can +//! race teardown. Each child therefore gets both: +//! +//! - `PR_SET_PDEATHSIG(SIGKILL)`, with the standard parent-race check; and +//! - its own process group, so explicit teardown reaches descendants as well as +//! the direct child. +//! +//! This is intentionally the inverse of [`super::process`], whose viewer +//! players are user-facing detached processes and are meant to survive their +//! launcher. + +use nix::libc; +use nix::sys::signal::{Signal, killpg}; +use nix::unistd::Pid; +use std::io; +use std::os::unix::process::CommandExt; +use std::process::{Child, Command}; + +/// Install parent-death and process-group containment on `command`. +/// +/// The closure runs between `fork` and `exec`, so it contains only direct +/// async-signal-safe syscalls. A failure aborts the spawn rather than launching +/// an uncontained graph-mutating child. +fn configure(command: &mut Command) { + let expected_parent = std::process::id() as libc::pid_t; + // SAFETY: `setpgid`, `prctl`, `getppid`, and `_exit` are direct syscalls and + // are async-signal-safe in the post-fork child. No allocation, logging, or + // locking occurs in the closure. + unsafe { + command.pre_exec(move || { + if libc::setpgid(0, 0) == -1 { + return Err(io::Error::last_os_error()); + } + if libc::prctl(libc::PR_SET_PDEATHSIG, libc::SIGKILL) == -1 { + return Err(io::Error::last_os_error()); + } + + // The parent may have died after fork but before PR_SET_PDEATHSIG + // was installed. Checking after the prctl closes that window: a + // later death delivers SIGKILL, an earlier one is handled here. + if libc::getppid() != expected_parent { + libc::_exit(127); + } + Ok(()) + }); + } +} + +/// Spawn a contained blocking child. +pub fn spawn(command: &mut Command) -> io::Result { + configure(command); + command.spawn() +} + +/// Spawn a contained Tokio child. +pub fn spawn_tokio(command: &mut tokio::process::Command) -> io::Result { + configure(command.as_std_mut()); + command.spawn() +} + +/// Signal the process group created by [`configure`]. +pub fn signal_group(leader_pid: u32, signal: Signal) -> nix::Result<()> { + killpg(Pid::from_raw(leader_pid as i32), signal) +} + +#[cfg(test)] +mod tests { + use super::*; + use std::io::Write; + use std::process::Stdio; + use std::time::{Duration, Instant}; + + const PDEATH_HELPER: &str = "PIXELPASS_TEST_PDEATH_HELPER"; + + fn process_is_running(pid: u32) -> bool { + let Ok(stat) = std::fs::read_to_string(format!("/proc/{pid}/stat")) else { + return false; + }; + // comm is parenthesized and may contain spaces; the state byte follows + // the final `) `. A zombie holds no resources and only awaits init's + // reap, so it is not a surviving capture child for this gate. + stat.rsplit_once(") ") + .and_then(|(_, rest)| rest.as_bytes().first().copied()) + .is_some_and(|state| state != b'Z' && state != b'X') + } + + #[test] + fn contained_child_has_its_own_process_group() { + let mut command = Command::new("sleep"); + command.arg("30"); + let mut child = spawn(&mut command).expect("spawn contained child"); + let pid = child.id(); + + // SAFETY: getpgid is a read-only syscall for the live child we own. + let pgid = unsafe { libc::getpgid(pid as libc::pid_t) }; + assert_eq!(pgid, pid as libc::pid_t); + + signal_group(pid, Signal::SIGKILL).expect("kill contained group"); + child.wait().expect("reap contained child"); + } + + #[test] + fn process_group_signal_reaches_descendants() { + let mut command = Command::new("sh"); + command + .args(["-c", "sleep 30 & child=$!; echo $child; wait"]) + .stdout(Stdio::piped()); + let mut leader = spawn(&mut command).expect("spawn group leader"); + let mut line = String::new(); + use std::io::BufRead; + std::io::BufReader::new(leader.stdout.take().expect("piped stdout")) + .read_line(&mut line) + .expect("read descendant pid"); + let descendant: u32 = line.trim().parse().expect("numeric descendant pid"); + assert!(process_is_running(descendant)); + + signal_group(leader.id(), Signal::SIGKILL).expect("kill whole group"); + leader.wait().expect("reap group leader"); + let deadline = Instant::now() + Duration::from_secs(2); + while process_is_running(descendant) && Instant::now() < deadline { + std::thread::sleep(Duration::from_millis(10)); + } + assert!( + !process_is_running(descendant), + "a descendant must not survive explicit group teardown" + ); + } + + /// Subprocess-only half of the parent-death gate. The outer test launches + /// this exact test in a disposable harness process; when that process exits, + /// the contained sleep must receive SIGKILL. + #[test] + fn pdeathsig_helper() { + if std::env::var_os(PDEATH_HELPER).is_none() { + return; + } + let mut command = Command::new("sleep"); + command + .arg("30") + .stdout(Stdio::null()) + .stderr(Stdio::null()); + let child = spawn(&mut command).expect("spawn pdeath child"); + println!("PIXELPASS_CONTAINED_PID={}", child.id()); + std::io::stdout().flush().expect("flush child pid"); + // std::process::Child has no kill-on-drop behavior. Returning lets the + // helper harness exit while the contained process is still live. + drop(child); + } + + #[test] + fn parent_death_kills_the_contained_child() { + let helper = std::env::current_exe().expect("test executable path"); + let output = Command::new(helper) + .args([ + "--exact", + "common::contained::tests::pdeathsig_helper", + "--nocapture", + ]) + .env(PDEATH_HELPER, "1") + .output() + .expect("run pdeath helper harness"); + assert!( + output.status.success(), + "helper failed: {}", + String::from_utf8_lossy(&output.stderr) + ); + let stdout = String::from_utf8_lossy(&output.stdout); + let pid: u32 = stdout + .lines() + .find_map(|line| line.strip_prefix("PIXELPASS_CONTAINED_PID=")) + .expect("helper printed contained pid") + .parse() + .expect("numeric contained pid"); + + let deadline = Instant::now() + Duration::from_secs(2); + while process_is_running(pid) && Instant::now() < deadline { + std::thread::sleep(Duration::from_millis(10)); + } + assert!( + !process_is_running(pid), + "PR_SET_PDEATHSIG must stop the child when its PixelPass parent exits" + ); + } +} diff --git a/src/common/mod.rs b/src/common/mod.rs index 1335ddc..b4691c6 100644 --- a/src/common/mod.rs +++ b/src/common/mod.rs @@ -1,6 +1,7 @@ pub mod alpn; pub mod bandwidth; pub mod config; +pub mod contained; // The friends stack (persistent identity + control plane) is GUI-only — a // headless CLI host runs no presence service — so it's gated with the feature // that pulls the rest of the GUI, keeping the headless build lean. diff --git a/src/host/audio.rs b/src/host/audio.rs index 3b4c1e0..488d899 100644 --- a/src/host/audio.rs +++ b/src/host/audio.rs @@ -37,11 +37,14 @@ use std::io::{self, Read}; use std::process::{Child, Command, ExitStatus, Stdio}; use std::rc::Rc; use std::sync::Arc; -use std::thread::JoinHandle; +use std::sync::atomic::{AtomicBool, AtomicU8, Ordering}; use std::time::{Duration, Instant}; use crate::cli::HostOpts; +use crate::common::contained; +use crate::host::health; use crate::host::ledger::{self, LedgerError, ModuleLedger, UnloadOutcome}; +use crate::host::owned_thread::OwnedThread; use crate::repair::plan::{self as repair_plan, Fingerprint, Shape}; /// How long a `pactl load-module` worker may run before it is killed and reaped. @@ -54,6 +57,8 @@ use crate::repair::plan::{self as repair_plan, Fingerprint, Shape}; /// connect/list/unload requests instead of a second pactl connection. const PACTL_BUDGET: Duration = Duration::from_secs(5); const PACTL_REAP_BUDGET: Duration = Duration::from_secs(1); +const ROUTER_RUNNING_STOP_BUDGET: Duration = Duration::from_secs(2); +const ROUTER_STARTING_STOP_BUDGET: Duration = Duration::from_secs(5); /// Owns the pactl-loaded modules plus, when filtering is active, the /// libpipewire stream-router thread. Drop unloads modules as a backstop; @@ -71,12 +76,13 @@ pub struct Routing { sink_name: String, stream_router: Option, event_task: Option>, + health: health::Reporter, } impl Routing { /// Create the per-PID null-sink + loopback. If `opts.app` is set, /// also spawn the libpipewire thread that reroutes matching streams. - pub async fn start(opts: &HostOpts) -> Result { + pub(super) async fn start(opts: &HostOpts, health: health::Reporter) -> Result { let pid = std::process::id(); let sink_name = repair_plan::sink_name_for(pid); let ledger = ModuleLedger::new(); @@ -90,6 +96,7 @@ impl Routing { sink_name: sink_name.clone(), stream_router: None, event_task: None, + health: health.clone(), }; // Every module this host loads carries an ownership token, minted per @@ -125,7 +132,8 @@ impl Routing { ); if let Some(app) = &opts.app { - let (router, mut event_rx) = StreamRouter::spawn(app.clone(), sink_name.clone())?; + let (router, mut event_rx) = + StreamRouter::spawn(app.clone(), sink_name.clone(), health.clone())?; let ledger_for_task = Arc::clone(&ledger); let strict = opts.strict_audio; let event_task = tokio::spawn(async move { @@ -235,27 +243,44 @@ impl Routing { // task cannot register another mutation even if it receives one last // router event while shutdown is in progress. self.ledger.close(); - if let Some(router) = self.stream_router.take() { - // ⚠️ Still an unbounded join: a wedged PipeWire thread parks this - // task indefinitely. That is the pre-existing defect S3b exists for. - // Nothing here makes it worse, and the ledger is what will make - // bounding it safe when it lands. - router.shutdown(); - } + let router_stopped = if let Some(router) = self.stream_router.take() { + router.shutdown().await + } else { + true + }; if let Some(mut task) = self.event_task.take() { - // The router's exit drops the event senders, so the task normally - // ends by itself. Abort is the fallback, and it is awaited through - // `&mut JoinHandle` so the future is genuinely dropped — and with it - // any in-flight permit — before reconciliation reads the ledger. - // Dropping the handle instead would *detach* the task, which is how a - // load could still land after teardown believed it was finished. - if tokio::time::timeout(PACTL_BUDGET, &mut task).await.is_err() { - tracing::warn!( - "audio routing: the event task did not finish within {PACTL_BUDGET:?}; \ - cancelling it" - ); + if !router_stopped { + // A quarantined router still owns its event sender, so this task + // cannot finish naturally. Cancel and await it before ledger + // reconciliation; the router timeout already poisoned the host. task.abort(); let _ = task.await; + } else { + // The router's exit drops the event senders, so the task normally + // ends by itself. Abort is the fallback, and it is awaited through + // `&mut JoinHandle` so the future is genuinely dropped — and with + // it any in-flight permit — before reconciliation reads the + // ledger. Dropping the handle instead would *detach* the task, + // which is how a load could still land after teardown believed it + // was finished. + match tokio::time::timeout(PACTL_BUDGET, &mut task).await { + Ok(Ok(())) => {} + Ok(Err(e)) => { + self.health + .poison(format!("audio routing event task failed: {e}")); + } + Err(_) => { + tracing::warn!( + "audio routing: the event task did not finish within {PACTL_BUDGET:?}; \ + cancelling it" + ); + self.health.poison(format!( + "audio routing event task did not stop within {PACTL_BUDGET:?}" + )); + task.abort(); + let _ = task.await; + } + } } } @@ -269,6 +294,8 @@ impl Routing { .await { tracing::warn!("audio routing: module-operation wait task failed: {e}"); + self.health + .poison(format!("audio module-operation wait task failed: {e}")); } cleanup_modules(&self.ledger).await; @@ -278,6 +305,9 @@ impl Routing { "audio routing: some audio modules could not be removed safely; \ `pixelpass --repair` will clean up anything left behind" ); + self.health.poison( + "audio routing teardown left module ownership unresolved; repair is required", + ); } } } @@ -292,7 +322,7 @@ impl Drop for Routing { fn drop(&mut self) { self.ledger.close(); if let Some(router) = self.stream_router.take() { - router.shutdown(); + drop(router); } if let Some(task) = self.event_task.take() { task.abort(); @@ -305,6 +335,8 @@ impl Drop for Routing { "audio routing: torn down with modules that could not be removed safely; \ run `pixelpass --repair` to clean up anything left behind" ); + self.health + .poison("audio routing Drop left module ownership unresolved; repair is required"); } } } @@ -635,7 +667,7 @@ impl Drop for ReapedChild { if self.reaped { return; } - let _ = self.child.kill(); + self.kill_group(); match self.reap_within(PACTL_REAP_BUDGET) { Ok(Some(_)) => {} Ok(None) => tracing::warn!( @@ -647,6 +679,12 @@ impl Drop for ReapedChild { } impl ReapedChild { + fn kill_group(&mut self) { + let _ = contained::signal_group(self.child.id(), nix::sys::signal::Signal::SIGKILL); + // Backstop in case the group disappeared between lookup and signal. + let _ = self.child.kill(); + } + fn reap_within(&mut self, budget: Duration) -> io::Result> { let deadline = Instant::now() + budget; loop { @@ -665,7 +703,7 @@ impl ReapedChild { fn bounded_output(command: &mut Command, budget: Duration) -> io::Result { command.stdout(Stdio::piped()).stderr(Stdio::piped()); let mut child = ReapedChild { - child: command.spawn()?, + child: contained::spawn(command)?, reaped: false, }; let stdout = child @@ -700,7 +738,7 @@ fn bounded_output(command: &mut Command, budget: Duration) -> io::Result= deadline { - let _ = child.child.kill(); + child.kill_group(); let Some(status) = child.reap_within(PACTL_REAP_BUDGET)? else { return Err(io::Error::new( io::ErrorKind::TimedOut, @@ -807,24 +845,53 @@ enum Event { /// Handle to the libpipewire stream-router thread. pub struct StreamRouter { cmd_tx: pipewire::channel::Sender, - thread: Option>, + thread: OwnedThread, + phase: Arc, } +const ROUTER_STARTING: u8 = 0; +const ROUTER_RUNNING: u8 = 1; +const ROUTER_EXITED: u8 = 2; + impl StreamRouter { /// Spawn the libpipewire thread. Returns the router handle and the /// event receiver tokio side polls. fn spawn( filter_name: String, sink_name: String, + health: health::Reporter, ) -> Result<(Self, tokio::sync::mpsc::UnboundedReceiver)> { let (cmd_tx, cmd_rx) = pipewire::channel::channel::(); let (event_tx, event_rx) = tokio::sync::mpsc::unbounded_channel::(); + let phase = Arc::new(AtomicU8::new(ROUTER_STARTING)); + let phase_for_thread = Arc::clone(&phase); + let shutdown_observed = Arc::new(AtomicBool::new(false)); + let shutdown_for_thread = Arc::clone(&shutdown_observed); + let health_for_thread = health.clone(); let thread = std::thread::Builder::new() .name("pixelpass-pw-router".to_string()) .spawn(move || { - if let Err(e) = run_router(filter_name, sink_name, cmd_rx, event_tx) { - tracing::warn!("audio routing: libpipewire thread exited with error: {e:#}"); + let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| { + run_router( + filter_name, + sink_name, + cmd_rx, + event_tx, + Arc::clone(&phase_for_thread), + Arc::clone(&shutdown_for_thread), + ) + })); + phase_for_thread.store(ROUTER_EXITED, Ordering::Release); + match result { + Ok(result) => report_router_exit( + &health_for_thread, + shutdown_for_thread.load(Ordering::Acquire), + result, + ), + Err(_) => { + health_for_thread.poison("libpipewire router thread panicked"); + } } }) .context("failed to spawn libpipewire router thread")?; @@ -832,19 +899,46 @@ impl StreamRouter { Ok(( Self { cmd_tx, - thread: Some(thread), + thread: OwnedThread::new("libpipewire router thread", thread, health), + phase, }, event_rx, )) } - fn shutdown(mut self) { + async fn shutdown(mut self) -> bool { // Best-effort: if the send fails the thread is already gone. let _ = self.cmd_tx.send(Cmd::Shutdown); - if let Some(t) = self.thread.take() - && let Err(e) = t.join() - { - tracing::warn!("audio routing: pw thread join failed: {e:?}"); + let budget = router_shutdown_budget(self.phase.load(Ordering::Acquire)); + self.thread.join_within(budget).await + } +} + +impl Drop for StreamRouter { + fn drop(&mut self) { + // If async shutdown is cancelled, wake the MainLoop before OwnedThread's + // Drop poisons/quarantines the still-owned handle. + let _ = self.cmd_tx.send(Cmd::Shutdown); + } +} + +fn router_shutdown_budget(phase: u8) -> Duration { + if phase == ROUTER_STARTING { + ROUTER_STARTING_STOP_BUDGET + } else { + ROUTER_RUNNING_STOP_BUDGET + } +} + +fn report_router_exit(health: &health::Reporter, shutdown_observed: bool, result: Result<()>) { + match result { + Ok(()) if shutdown_observed => {} + Ok(()) => { + health.poison("libpipewire router thread exited without a shutdown command"); + } + Err(e) => { + tracing::warn!("audio routing: libpipewire thread exited with error: {e:#}"); + health.poison(format!("libpipewire router thread failed: {e:#}")); } } } @@ -856,6 +950,8 @@ fn run_router( sink_name: String, cmd_rx: pipewire::channel::Receiver, event_tx: tokio::sync::mpsc::UnboundedSender, + phase: Arc, + shutdown_observed: Arc, ) -> Result<()> { use pipewire::{self as pw, types::ObjectType}; @@ -878,8 +974,10 @@ fn run_router( // Cmd handler: clear metadata for routed streams, then quit. let main_loop_for_cmd = main_loop.clone(); let state_for_cmd = Rc::clone(&state); + let shutdown_for_cmd = Arc::clone(&shutdown_observed); let _cmd_recv = cmd_rx.attach(main_loop.loop_(), move |cmd| match cmd { Cmd::Shutdown => { + shutdown_for_cmd.store(true, Ordering::Release); let s = state_for_cmd.borrow(); if let Some(meta) = &s.default_metadata { for &nid in &s.routed_node_ids { @@ -978,6 +1076,7 @@ fn run_router( .register(); tracing::info!(filter = %filter_name, "audio routing: pw thread running"); + phase.store(ROUTER_RUNNING, Ordering::Release); main_loop.run(); tracing::info!("audio routing: pw thread exiting"); Ok(()) @@ -1074,6 +1173,7 @@ mod tests { use super::*; use crate::host::ledger::SlotState; use crate::repair::plan::{ModuleObservation, classify}; + use std::sync::mpsc; /// Whole-desktop routing: no app filter, so no PipeWire thread and no event /// task — just the null-sink and its default-sink loopback. @@ -1094,6 +1194,86 @@ mod tests { } } + #[test] + fn router_exit_is_healthy_only_after_the_thread_observed_shutdown() { + let (clean, _) = health::channel(); + report_router_exit(&clean, true, Ok(())); + assert!(clean.fault().is_none()); + + let (unexpected, _) = health::channel(); + report_router_exit(&unexpected, false, Ok(())); + assert_eq!( + unexpected.fault().as_deref(), + Some("libpipewire router thread exited without a shutdown command") + ); + + let (failed, _) = health::channel(); + report_router_exit(&failed, false, Err(anyhow::anyhow!("fixture failure"))); + assert!( + failed + .fault() + .as_deref() + .is_some_and(|reason| reason.contains("fixture failure")) + ); + } + + #[test] + fn router_shutdown_has_distinct_startup_and_running_budgets() { + assert_eq!( + router_shutdown_budget(ROUTER_STARTING), + ROUTER_STARTING_STOP_BUDGET + ); + assert_eq!( + router_shutdown_budget(ROUTER_RUNNING), + ROUTER_RUNNING_STOP_BUDGET + ); + assert!(ROUTER_STARTING_STOP_BUDGET > ROUTER_RUNNING_STOP_BUDGET); + } + + #[tokio::test] + async fn cancelling_router_shutdown_keeps_the_thread_owned_and_poisons() { + let (cmd_tx, _cmd_rx) = pipewire::channel::channel::(); + let (release_tx, release_rx) = mpsc::channel(); + let (health, _) = health::channel(); + let thread = std::thread::spawn(move || { + let _ = release_rx.recv(); + }); + let router = StreamRouter { + cmd_tx, + thread: OwnedThread::new("cancellation fixture", thread, health.clone()), + phase: Arc::new(AtomicU8::new(ROUTER_STARTING)), + }; + + assert!( + tokio::time::timeout(Duration::from_millis(20), router.shutdown()) + .await + .is_err(), + "the outer timeout must cancel shutdown before its policy deadline" + ); + assert!( + health.fault().is_some(), + "cancellation must poison instead of detaching the OS handle" + ); + release_tx.send(()).expect("release quarantined fixture"); + } + + #[test] + fn bounded_module_worker_uses_the_contained_spawn_path() { + let mut command = Command::new("sh"); + command.args([ + "-c", + "read pid comm state ppid pgrp rest < /proc/self/stat; printf '%s %s' \"$pid\" \"$pgrp\"", + ]); + let output = bounded_output(&mut command, Duration::from_secs(1)) + .expect("run contained module-worker fixture"); + assert!(output.status.success()); + let ids = String::from_utf8(output.stdout).expect("ascii pid/pgid"); + let mut ids = ids.split_whitespace(); + let pid = ids.next().expect("child pid"); + let pgid = ids.next().expect("child process group"); + assert_eq!(pid, pgid, "module worker must lead its own process group"); + } + /// The module table exactly as `--repair` observes it. fn module_snapshot() -> Vec<(u32, String, String)> { let mut session = @@ -1148,7 +1328,8 @@ mod tests { #[ignore = "loads real Pulse modules; run with --ignored --test-threads=1"] async fn live_teardown_leaves_the_module_table_as_it_found_it() { let before = module_snapshot(); - let routing = Routing::start(&whole_desktop_opts()) + let (health, _) = health::channel(); + let routing = Routing::start(&whole_desktop_opts(), health) .await .expect("routing starts"); @@ -1180,6 +1361,39 @@ mod tests { ); } + /// Exercise the real libpipewire mainloop command path, not only the + /// whole-desktop module path above. The deliberately unmatched app filter + /// is enough to start the router without moving an unrelated live stream. + #[tokio::test] + #[ignore = "uses the real Pulse/PipeWire graph; run with --ignored --test-threads=1"] + async fn live_stream_router_stops_within_its_policy_budget() { + let before = module_snapshot(); + let mut opts = whole_desktop_opts(); + opts.app = Some("__pixelpass_s3b_no_matching_application__".to_string()); + opts.strict_audio = true; + let (health, _) = health::channel(); + let routing = Routing::start(&opts, health.clone()) + .await + .expect("per-app routing starts"); + + // Let the OS thread reach its normal running phase so this covers the + // tighter steady-state budget rather than only startup containment. + tokio::time::sleep(Duration::from_millis(100)).await; + tokio::time::timeout(Duration::from_secs(10), routing.shutdown()) + .await + .expect("router and graph teardown stay globally bounded"); + + assert!( + health.fault().is_none(), + "an observed shutdown and successful join must remain healthy" + ); + assert_eq!( + module_snapshot(), + before, + "per-app teardown must leave the module table byte-identical" + ); + } + /// The orphan race, staged against a real server: a load cancelled while /// `pactl` is in flight must still be findable and removable. /// diff --git a/src/host/capture.rs b/src/host/capture.rs index 4635826..1257bf8 100644 --- a/src/host/capture.rs +++ b/src/host/capture.rs @@ -8,18 +8,20 @@ use anyhow::Result; use crate::cli::HostOpts; use crate::common::display::DisplayServer; +use crate::host::health; use crate::host::pipeline::CaptureHandle; use crate::host::quality::EffectiveQuality; use crate::host::{wayland, x11}; -pub async fn spawn( +pub(super) async fn spawn( display: DisplayServer, opts: &HostOpts, quality: &EffectiveQuality, + health: health::Reporter, ) -> Result { match display { - DisplayServer::Wayland => wayland::start(opts, quality).await, - DisplayServer::X11 => x11::start(opts, quality).await, + DisplayServer::Wayland => wayland::start(opts, quality, health).await, + DisplayServer::X11 => x11::start(opts, quality, health).await, DisplayServer::Unknown => unreachable!("caller guarantees display != Unknown"), } } diff --git a/src/host/health.rs b/src/host/health.rs new file mode 100644 index 0000000..fd230f2 --- /dev/null +++ b/src/host/health.rs @@ -0,0 +1,83 @@ +//! Terminal health shared by capture components and the host supervisor. +//! +//! A capture-side ownership failure is not recoverable inside the same host +//! process: a wedged PipeWire owner or an unaccounted Pulse module can collide +//! with the next capture. Health is therefore one-way (`Healthy -> Poisoned`) +//! and first-fault-wins so a later, less precise teardown symptom cannot erase +//! the original cause. + +use std::sync::Arc; +use tokio::sync::watch; + +#[derive(Clone, Debug, PartialEq, Eq)] +pub(super) enum State { + Healthy, + Poisoned(Arc), +} + +#[derive(Clone)] +pub(super) struct Reporter { + tx: watch::Sender, +} + +pub(super) struct Monitor { + rx: watch::Receiver, +} + +pub(super) fn channel() -> (Reporter, Monitor) { + let (tx, rx) = watch::channel(State::Healthy); + (Reporter { tx }, Monitor { rx }) +} + +impl Reporter { + /// Poison the host exactly once. Returns true for the first fault. + pub(super) fn poison(&self, reason: impl Into>) -> bool { + let reason = reason.into(); + self.tx.send_if_modified(move |state| { + if matches!(state, State::Healthy) { + *state = State::Poisoned(reason); + true + } else { + false + } + }) + } + + #[cfg(test)] + pub(super) fn fault(&self) -> Option> { + match &*self.tx.borrow() { + State::Healthy => None, + State::Poisoned(reason) => Some(Arc::clone(reason)), + } + } +} + +impl Monitor { + pub(super) fn fault(&self) -> Option> { + match &*self.rx.borrow() { + State::Healthy => None, + State::Poisoned(reason) => Some(Arc::clone(reason)), + } + } + + pub(super) async fn changed(&mut self) { + // The supervisor owns a Reporter for the whole run, so closure is not a + // normal state. Treat it like a wake and let `fault()` decide. + let _ = self.rx.changed().await; + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[tokio::test] + async fn poison_is_terminal_and_first_fault_wins() { + let (reporter, mut monitor) = channel(); + assert!(reporter.poison("first")); + monitor.changed().await; + assert_eq!(monitor.fault().as_deref(), Some("first")); + assert!(!reporter.poison("second")); + assert_eq!(reporter.fault().as_deref(), Some("first")); + } +} diff --git a/src/host/mod.rs b/src/host/mod.rs index 944c43f..087ee06 100644 --- a/src/host/mod.rs +++ b/src/host/mod.rs @@ -2,8 +2,10 @@ pub mod aec; pub mod audio; pub mod audit; mod capture; +mod health; pub mod ledger; mod observer; +mod owned_thread; mod pipeline; mod quality; mod serve; @@ -127,12 +129,15 @@ pub async fn run(opts: HostOpts) -> Result<()> { }); let (sup_tx, sup_rx) = mpsc::channel::(16); + let (capture_health, capture_health_monitor) = health::channel(); let supervisor = tokio::spawn(supervise( opts.clone(), quality, display, resolution.value, sup_rx, + (capture_health, capture_health_monitor), + cancel.clone(), )); // Command channel for the GUI front-end: read `kick ` lines @@ -289,14 +294,46 @@ async fn supervise( display: DisplayServer, max_viewers: u32, mut rx: mpsc::Receiver, + capture_health: (health::Reporter, health::Monitor), + host_cancel: CancellationToken, ) { + let (capture_health, mut capture_health_monitor) = capture_health; let mut handle: Option = None; // Active viewers, keyed by endpoint id, holding each one's kill switch. // The count is just `viewers.len()`. (A given endpoint connecting twice is // a non-case here: each viewer process uses a fresh ephemeral identity.) let mut viewers: HashMap = HashMap::new(); - while let Some(msg) = rx.recv().await { + loop { + // Poison is terminal for this host process. A surviving wedged owner or + // an unaccounted graph mutation can collide with a later capture, so do + // not detach it and keep advertising the ticket. Cancelling the host + // closes the endpoint; PeerSpeak's S2 EOF path then clears presence. + if let Some(reason) = capture_health_monitor.fault() { + tracing::error!(%reason, "capture supervisor poisoned — stopping host"); + host_cancel.cancel(); + for cancel in viewers.values() { + cancel.cancel(); + } + if let Some(h) = handle.take() { + h.shutdown().await; + output::emit(output::Event::Capture { + state: output::CaptureState::Stopped, + }); + } + break; + } + + let msg = tokio::select! { + // Health wins a simultaneous race with another viewer request: a + // poisoned host must not begin one more capture. + biased; + _ = capture_health_monitor.changed() => continue, + msg = rx.recv() => msg, + }; + let Some(msg) = msg else { + break; + }; match msg { SupervisorMsg::AddViewer { id, cancel, reply } => { let count = viewers.len() as u32; @@ -310,8 +347,37 @@ async fn supervise( if handle.is_none() { tracing::info!("first viewer arriving — spawning capture"); - match capture::spawn(display, &opts, &quality).await { + let spawned = tokio::select! { + // A subsystem can fail while the portal/GStreamer setup + // future is still running. Do not wait for setup to + // return and then admit one viewer to an already-poisoned + // capture. + biased; + _ = capture_health_monitor.changed() => { + let reason = capture_health_monitor + .fault() + .unwrap_or_else(|| "capture health channel changed unexpectedly".into()); + let _ = reply.send(Err(format!( + "capture ownership failed during startup: {reason}" + ))); + continue; + } + result = capture::spawn( + display, + &opts, + &quality, + capture_health.clone(), + ) => result, + }; + match spawned { Ok(h) => { + if let Some(reason) = capture_health_monitor.fault() { + h.shutdown().await; + let _ = reply.send(Err(format!( + "capture ownership failed during startup: {reason}" + ))); + continue; + } handle = Some(h); output::emit(output::Event::Capture { state: output::CaptureState::Started, @@ -568,4 +634,34 @@ mod tests { assert_eq!(initial_app_audio_state(&opts(None, true)), None); assert_eq!(initial_app_audio_state(&opts(None, false)), None); } + + #[tokio::test] + async fn supervisor_health_poison_cancels_the_host_with_command_channel_open() { + let opts = opts(None, false); + let quality = quality::resolve(&opts, 1); + let (tx, rx) = mpsc::channel(1); + let (health, monitor) = health::channel(); + let cancel = CancellationToken::new(); + + let supervisor = tokio::spawn(supervise( + opts, + quality, + DisplayServer::Unknown, + 1, + rx, + (health.clone(), monitor), + cancel.clone(), + )); + assert!(health.poison("test ownership fault")); + tokio::time::timeout(Duration::from_secs(1), supervisor) + .await + .expect("poisoned supervisor must stop promptly") + .expect("supervisor task must not panic"); + assert!(cancel.is_cancelled()); + + // The sender deliberately stayed open until the supervisor exited. If + // the health arm were removed, the task above would still be blocked on + // `rx.recv()` and the timeout would fail. + drop(tx); + } } diff --git a/src/host/owned_thread.rs b/src/host/owned_thread.rs new file mode 100644 index 0000000..72ae856 --- /dev/null +++ b/src/host/owned_thread.rs @@ -0,0 +1,170 @@ +//! Cancellation-safe ownership for blocking subsystem threads. +//! +//! `std::thread::JoinHandle` has no timed join. Moving it into +//! `spawn_blocking` only moves the problem: cancelling the async waiter detaches +//! the blocking task and loses the only handle. Instead this owner polls +//! `is_finished()` while retaining the handle, joins only after completion, and +//! quarantines an unfinished handle on timeout or Drop. Quarantine is +//! process-lifetime ownership, not recovery; the shared health channel makes +//! the supervisor terminate the poisoned host. + +use super::health::Reporter; +use std::sync::{Mutex, OnceLock}; +use std::thread::JoinHandle; +use std::time::Duration; + +static QUARANTINED: OnceLock>>> = OnceLock::new(); + +fn quarantine(handle: JoinHandle<()>) { + let mut handles = QUARANTINED + .get_or_init(|| Mutex::new(Vec::new())) + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()); + + // Reap anything that happened to finish since the previous quarantine; + // every still-running handle remains owned until process exit. + let old = std::mem::take(&mut *handles); + for old_handle in old { + if old_handle.is_finished() { + let _ = old_handle.join(); + } else { + handles.push(old_handle); + } + } + handles.push(handle); +} + +pub(super) struct OwnedThread { + name: &'static str, + handle: Option>, + health: Reporter, +} + +impl OwnedThread { + pub(super) fn new(name: &'static str, handle: JoinHandle<()>, health: Reporter) -> Self { + Self { + name, + handle: Some(handle), + health, + } + } + + /// Wait up to `budget`, retaining the OS handle across every await. + /// + /// Returns true only when the thread was joined successfully. Timeout and + /// panic poison the process; a timeout also moves the still-running handle + /// into process-lifetime quarantine. + pub(super) async fn join_within(&mut self, budget: Duration) -> bool { + let deadline = tokio::time::Instant::now() + budget; + loop { + let Some(handle) = self.handle.as_ref() else { + return true; + }; + if handle.is_finished() { + let handle = self.handle.take().expect("checked as present"); + return match handle.join() { + Ok(()) => true, + Err(_) => { + self.health + .poison(format!("{} panicked during shutdown", self.name)); + false + } + }; + } + if tokio::time::Instant::now() >= deadline { + self.health.poison(format!( + "{} did not stop within {:?}; its thread handle is quarantined", + self.name, budget + )); + quarantine(self.handle.take().expect("checked as present")); + return false; + } + tokio::time::sleep(Duration::from_millis(10)).await; + } + } +} + +impl Drop for OwnedThread { + fn drop(&mut self) { + let Some(handle) = self.handle.take() else { + return; + }; + if handle.is_finished() { + if handle.join().is_err() { + self.health + .poison(format!("{} panicked before it was joined", self.name)); + } + return; + } + + self.health.poison(format!( + "{} was dropped before it stopped; its thread handle is quarantined", + self.name + )); + quarantine(handle); + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::host::health; + use std::sync::mpsc; + + fn reap_finished_quarantine() { + let Some(handles) = QUARANTINED.get() else { + return; + }; + let mut handles = handles + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()); + let old = std::mem::take(&mut *handles); + for handle in old { + if handle.is_finished() { + let _ = handle.join(); + } else { + handles.push(handle); + } + } + } + + #[tokio::test] + async fn completed_thread_is_joined_without_poison() { + let (health, _) = health::channel(); + let handle = std::thread::spawn(|| {}); + let mut owner = OwnedThread::new("test worker", handle, health.clone()); + assert!(owner.join_within(Duration::from_secs(1)).await); + assert!(health.fault().is_none()); + } + + #[tokio::test] + async fn timeout_poisons_and_quarantines_instead_of_detaching() { + let (release_tx, release_rx) = mpsc::channel(); + let (health, _) = health::channel(); + let handle = std::thread::spawn(move || { + let _ = release_rx.recv(); + }); + let mut owner = OwnedThread::new("wedged worker", handle, health.clone()); + + assert!(!owner.join_within(Duration::from_millis(20)).await); + assert!(health.fault().is_some()); + drop(owner); + release_tx.send(()).expect("release quarantined worker"); + std::thread::sleep(Duration::from_millis(20)); + reap_finished_quarantine(); + } + + #[test] + fn dropping_an_unfinished_owner_poisons_and_quarantines() { + let (release_tx, release_rx) = mpsc::channel(); + let (health, _) = health::channel(); + let handle = std::thread::spawn(move || { + let _ = release_rx.recv(); + }); + drop(OwnedThread::new("cancelled worker", handle, health.clone())); + assert!(health.fault().is_some()); + release_tx.send(()).expect("release quarantined worker"); + std::thread::sleep(Duration::from_millis(20)); + reap_finished_quarantine(); + } +} diff --git a/src/host/pipeline.rs b/src/host/pipeline.rs index fda7694..61cc163 100644 --- a/src/host/pipeline.rs +++ b/src/host/pipeline.rs @@ -6,22 +6,122 @@ //! and lives here. Backends call [`spawn`] with just their source-element args. use anyhow::{Context, Result, bail}; -use nix::sys::signal::{Signal, kill}; -use nix::unistd::Pid; +use nix::sys::signal::Signal; use std::process::Stdio; +use std::sync::Arc; +use std::sync::atomic::{AtomicBool, Ordering}; use std::time::Duration; use tokio::process::{Child, Command}; use tokio::time::timeout; use super::audio::Routing; +use super::health; use super::quality::EffectiveQuality; use super::serve::Serve; use crate::cli::HostOpts; +use crate::common::contained; -pub struct CaptureHandle { - gst: Option, +const GST_TERM_BUDGET: Duration = Duration::from_secs(1); +const GST_KILL_BUDGET: Duration = Duration::from_secs(1); + +/// Owns the contained GStreamer process from spawn through confirmed reap. +/// +/// Keeping this guard alive during `Serve::bind` closes the old constructor +/// leak: any error after spawn kills the whole process group, not merely the +/// direct child. An unconfirmed Drop poisons the host so the supervisor cannot +/// start another capture on top of a possibly-live portal/PipeWire owner. +struct CaptureProcess { + child: Child, + leader_pid: u32, + reaped: bool, + health: health::Reporter, +} + +impl CaptureProcess { + fn new(child: Child, health: health::Reporter) -> Result { + let leader_pid = child + .id() + .context("gst-launch-1.0 exited before its process id was recorded")?; + Ok(Self { + child, + leader_pid, + reaped: false, + health, + }) + } + + fn take_stdout(&mut self) -> Option { + self.child.stdout.take() + } + + async fn shutdown(&mut self) { + // Reap an already-dead child before addressing its process group. A + // zombie still reserves its pid, but after `try_wait` succeeds that pid + // may be reused; returning here avoids ever signalling a new group that + // inherited the old numeric id. + match self.child.try_wait() { + Ok(Some(_)) => { + self.reaped = true; + self.health + .poison("gst-launch-1.0 exited before PixelPass began capture shutdown"); + return; + } + Ok(None) => {} + Err(e) => tracing::warn!( + "capture: could not inspect gst before SIGTERM ({e}); continuing teardown" + ), + } + + let _ = contained::signal_group(self.leader_pid, Signal::SIGTERM); + match timeout(GST_TERM_BUDGET, self.child.wait()).await { + Ok(Ok(_)) => { + self.reaped = true; + return; + } + Ok(Err(e)) => tracing::warn!( + "capture: gst wait after SIGTERM failed ({e}); escalating to SIGKILL" + ), + Err(_) => tracing::warn!( + "capture: gst did not exit within {GST_TERM_BUDGET:?}; escalating to SIGKILL" + ), + } + + let _ = contained::signal_group(self.leader_pid, Signal::SIGKILL); + let _ = self.child.start_kill(); + match timeout(GST_KILL_BUDGET, self.child.wait()).await { + Ok(Ok(_)) => self.reaped = true, + Ok(Err(e)) => { + self.health.poison(format!( + "gst-launch-1.0 could not be reaped after SIGKILL: {e}" + )); + } + Err(_) => { + self.health.poison(format!( + "gst-launch-1.0 did not exit within {GST_KILL_BUDGET:?} after SIGKILL" + )); + } + } + } +} + +impl Drop for CaptureProcess { + fn drop(&mut self) { + if self.reaped { + return; + } + let _ = contained::signal_group(self.leader_pid, Signal::SIGKILL); + let _ = self.child.start_kill(); + self.health.poison( + "gst-launch-1.0 was dropped before a confirmed reap; its process group was killed", + ); + } +} + +pub(super) struct CaptureHandle { + gst: Option, audio: Option, serve: Option, + stopping: Arc, } impl CaptureHandle { @@ -32,19 +132,15 @@ impl CaptureHandle { .local_port() } - /// Graceful teardown: SIGTERM gst, give it ~1s to exit, then SIGKILL, + /// Graceful teardown: SIGTERM the gst process group, give it ~1s to exit, + /// then SIGKILL and a second bounded reap, /// unload audio routing (if any), then tear down the serve layer. /// The serve reader will see EOF on gst stdout and exit on its own; /// serve.shutdown() is the backstop. pub async fn shutdown(mut self) { - if let Some(child) = self.gst.as_mut() - && let Some(pid) = child.id() - { - let _ = kill(Pid::from_raw(pid as i32), Signal::SIGTERM); - } - if let Some(child) = self.gst.as_mut() { - let _ = timeout(Duration::from_millis(1000), child.wait()).await; - let _ = child.start_kill(); + self.stopping.store(true, Ordering::Release); + if let Some(mut gst) = self.gst.take() { + gst.shutdown().await; } if let Some(audio) = self.audio.take() { audio.shutdown().await; @@ -57,10 +153,9 @@ impl CaptureHandle { impl Drop for CaptureHandle { fn drop(&mut self) { - if let Some(child) = self.gst.as_mut() { - let _ = child.start_kill(); - } - // Routing's and Serve's own Drop impls handle the rest. + self.stopping.store(true, Ordering::Release); + // CaptureProcess kills the whole process group and poisons the host; + // Routing's and Serve's own Drop impls handle their respective layers. } } @@ -73,14 +168,15 @@ impl Drop for CaptureHandle { /// `after_spawn` runs once, immediately after the gst child is launched — /// Wayland uses it to `close` the pipewire fd it leaked into the child; X11 /// passes a no-op. -pub async fn spawn( +pub(super) async fn spawn( opts: &HostOpts, quality: &EffectiveQuality, source_dims: Option<(u32, u32)>, source_args: Vec, + health: health::Reporter, after_spawn: impl FnOnce(), ) -> Result { - let (audio_routing, audio_device) = setup_audio(opts).await?; + let (audio_routing, audio_device) = setup_audio(opts, health.clone()).await?; let args = build_args(&source_args, &audio_device, opts, quality, source_dims); let mut gst_cmd = Command::new("gst-launch-1.0"); @@ -88,29 +184,32 @@ pub async fn spawn( .args(&args) .stdin(Stdio::null()) .stdout(Stdio::piped()) - .stderr(Stdio::inherit()); + .stderr(Stdio::inherit()) + .kill_on_drop(true); if std::env::var_os("PIXELPASS_GST_DEBUG").is_some() { gst_cmd.env("GST_DEBUG", "3"); } - let mut gst = gst_cmd.spawn().context("failed to spawn gst-launch-1.0")?; + let gst = contained::spawn_tokio(&mut gst_cmd).context("failed to spawn gst-launch-1.0")?; + let mut gst = CaptureProcess::new(gst, health.clone())?; // Backend-specific post-spawn cleanup (Wayland closes its leaked pw fd here, // once gst has inherited its own copy). after_spawn(); let gst_stdout = gst - .stdout - .take() + .take_stdout() .context("gst-launch-1.0 stdout pipe unavailable")?; // Hand stdout to the serve layer, which binds the localhost HTTP listener // and runs the broadcast fanout. No demux/remux, no codec assumptions. - let serve = Serve::bind(gst_stdout).await?; + let stopping = Arc::new(AtomicBool::new(false)); + let serve = Serve::bind(gst_stdout, health.clone(), Arc::clone(&stopping)).await?; Ok(CaptureHandle { gst: Some(gst), audio: audio_routing, serve: Some(serve), + stopping, }) } @@ -120,12 +219,15 @@ pub async fn spawn( /// is set (no app filter — captures everything via the null-sink, used for /// dogfooding the loopback path). Otherwise we capture the default sink's /// monitor (system audio out), not the default source (the mic). -async fn setup_audio(opts: &HostOpts) -> Result<(Option, String)> { +async fn setup_audio( + opts: &HostOpts, + health: health::Reporter, +) -> Result<(Option, String)> { let routing_requested = opts.app.is_some() || std::env::var_os("PIXELPASS_AUDIO_VIA_NULL_SINK").is_some(); let audio_routing = if routing_requested { Some( - Routing::start(opts) + Routing::start(opts, health) .await .context("audio routing setup failed")?, ) @@ -349,3 +451,72 @@ async fn default_audio_monitor() -> Result { } Ok(format!("{sink}.monitor")) } + +#[cfg(test)] +mod tests { + use super::*; + use std::time::{Duration, Instant}; + + fn process_is_running(pid: u32) -> bool { + let Ok(stat) = std::fs::read_to_string(format!("/proc/{pid}/stat")) else { + return false; + }; + stat.rsplit_once(") ") + .and_then(|(_, rest)| rest.as_bytes().first().copied()) + .is_some_and(|state| state != b'Z' && state != b'X') + } + + fn sleeping_capture(health: health::Reporter) -> CaptureProcess { + let mut command = Command::new("sleep"); + command.arg("30").kill_on_drop(true); + let child = contained::spawn_tokio(&mut command).expect("spawn contained fixture"); + CaptureProcess::new(child, health).expect("capture process guard") + } + + #[tokio::test] + async fn graceful_capture_shutdown_confirms_reap_without_poison() { + let (health, _) = health::channel(); + let mut capture = sleeping_capture(health.clone()); + capture.shutdown().await; + assert!(health.fault().is_none()); + } + + #[tokio::test] + async fn an_already_exited_capture_is_not_misreported_as_a_clean_shutdown() { + let (health, _) = health::channel(); + let mut command = Command::new("true"); + command.kill_on_drop(true); + let child = contained::spawn_tokio(&mut command).expect("spawn short contained fixture"); + let mut capture = CaptureProcess::new(child, health.clone()).expect("capture guard"); + + let deadline = Instant::now() + Duration::from_secs(1); + while process_is_running(capture.leader_pid) && Instant::now() < deadline { + tokio::task::yield_now().await; + } + assert!( + !process_is_running(capture.leader_pid), + "short fixture must exit before shutdown begins" + ); + + capture.shutdown().await; + assert_eq!( + health.fault().as_deref(), + Some("gst-launch-1.0 exited before PixelPass began capture shutdown") + ); + } + + #[tokio::test] + async fn dropping_live_capture_kills_its_group_and_poisons() { + let (health, _) = health::channel(); + let capture = sleeping_capture(health.clone()); + let pid = capture.leader_pid; + drop(capture); + assert!(health.fault().is_some()); + + let deadline = Instant::now() + Duration::from_secs(2); + while process_is_running(pid) && Instant::now() < deadline { + tokio::time::sleep(Duration::from_millis(10)).await; + } + assert!(!process_is_running(pid)); + } +} diff --git a/src/host/serve.rs b/src/host/serve.rs index 38bfef3..53c9057 100644 --- a/src/host/serve.rs +++ b/src/host/serve.rs @@ -10,6 +10,7 @@ use anyhow::{Context, Result, bail}; use std::sync::Arc; +use std::sync::atomic::{AtomicBool, Ordering}; use std::time::Duration; use tokio::io::{AsyncReadExt, AsyncWriteExt}; use tokio::net::{TcpListener, TcpStream}; @@ -18,6 +19,8 @@ use tokio::sync::broadcast; use tokio::task::JoinHandle; use tokio::time::{Instant, sleep}; +use super::health; + /// Broadcast-channel capacity in chunks. Each chunk is up to 64 KiB from /// the capture child's stdout, so 16 chunks ≈ 1 MiB ≈ ~2 s of buffered /// jitter at typical bitrates. A viewer that falls behind by more than @@ -40,14 +43,18 @@ impl Serve { /// Bind a localhost listener on a random port, set up the broadcast /// fanout, and spawn the reader + accept-loop tasks. The provided /// `stdout` is assumed to produce MPEG-TS bytes. - pub async fn bind(stdout: ChildStdout) -> Result { + pub(super) async fn bind( + stdout: ChildStdout, + health: health::Reporter, + stopping: Arc, + ) -> Result { let listener = TcpListener::bind("127.0.0.1:0") .await .context("could not bind local capture HTTP listener")?; let port = listener.local_addr()?.port(); let (tx, _) = broadcast::channel::>>(FANOUT_CAPACITY); - let reader = tokio::spawn(pump_to_broadcast(stdout, tx.clone())); + let reader = tokio::spawn(pump_to_broadcast(stdout, tx.clone(), health, stopping)); let server = tokio::spawn(run_accept_loop(listener, tx)); Ok(Self { @@ -105,12 +112,20 @@ pub async fn connect_to_capture(port: u16, max_wait: Duration) -> Result>>) { +async fn pump_to_broadcast( + mut stdout: impl tokio::io::AsyncRead + Unpin, + tx: broadcast::Sender>>, + health: health::Reporter, + stopping: Arc, +) { let mut buf = vec![0u8; READ_CHUNK]; loop { match stdout.read(&mut buf).await { Ok(0) => { tracing::info!("capture stdout EOF — fanout reader exiting"); + if !stopping.load(Ordering::Acquire) { + health.poison("GStreamer capture stdout closed unexpectedly"); + } return; } Ok(n) => { @@ -119,6 +134,9 @@ async fn pump_to_broadcast(mut stdout: ChildStdout, tx: broadcast::Sender { tracing::warn!("capture stdout read error: {e}"); + if !stopping.load(Ordering::Acquire) { + health.poison(format!("GStreamer capture stdout failed: {e}")); + } return; } } @@ -192,3 +210,37 @@ async fn drain_http_request(sock: &mut TcpStream) -> bool { } } } + +#[cfg(test)] +mod tests { + use super::*; + + #[tokio::test] + async fn unexpected_capture_eof_poisons_the_host() { + let (reader, writer) = tokio::io::duplex(16); + let (tx, _) = broadcast::channel(FANOUT_CAPACITY); + let (health, _) = health::channel(); + let stopping = Arc::new(AtomicBool::new(false)); + + drop(writer); + pump_to_broadcast(reader, tx, health.clone(), stopping).await; + + assert_eq!( + health.fault().as_deref(), + Some("GStreamer capture stdout closed unexpectedly") + ); + } + + #[tokio::test] + async fn expected_capture_eof_during_shutdown_stays_healthy() { + let (reader, writer) = tokio::io::duplex(16); + let (tx, _) = broadcast::channel(FANOUT_CAPACITY); + let (health, _) = health::channel(); + let stopping = Arc::new(AtomicBool::new(true)); + + drop(writer); + pump_to_broadcast(reader, tx, health.clone(), stopping).await; + + assert!(health.fault().is_none()); + } +} diff --git a/src/host/wayland.rs b/src/host/wayland.rs index f487a11..f5dcec4 100644 --- a/src/host/wayland.rs +++ b/src/host/wayland.rs @@ -14,11 +14,16 @@ use ashpd::{ use nix::fcntl::{FcntlArg, FdFlag, fcntl}; use std::os::fd::{AsFd, AsRawFd, OwnedFd, RawFd}; +use super::health; use super::pipeline::{self, CaptureHandle}; use super::quality::EffectiveQuality; use crate::cli::HostOpts; -pub async fn start(opts: &HostOpts, quality: &EffectiveQuality) -> Result { +pub(super) async fn start( + opts: &HostOpts, + quality: &EffectiveQuality, + health: health::Reporter, +) -> Result { // 1. Negotiate the screencast session with the portal. let proxy = Screencast::new() .await @@ -81,6 +86,7 @@ pub async fn start(opts: &HostOpts, quality: &EffectiveQuality) -> Result Result { +pub(super) async fn start( + opts: &HostOpts, + quality: &EffectiveQuality, + health: health::Reporter, +) -> Result { let xid = if opts.window { Some(pick_window().await?) } else { @@ -60,7 +65,7 @@ pub async fn start(opts: &HostOpts, quality: &EffectiveQuality) -> Result