diff --git a/src/core/mod.rs b/src/core/mod.rs index b0bc4e7..46b8c9f 100644 --- a/src/core/mod.rs +++ b/src/core/mod.rs @@ -4,6 +4,7 @@ pub mod fetchbudget; pub mod jitter; pub mod messages; mod recovery; +mod teardown; use crate::audio::eq::{Eq, EqSettings}; use crate::audio::{AudioBackend, PlatformAudioBackend}; @@ -677,31 +678,33 @@ struct ActiveSession { recovery_terminal_task: tokio::task::JoinHandle<()>, grace_timers: GraceTimers, transport: Arc, - /// Loaded PipeWire echo-cancel module (if enabled); unloads on drop. - #[cfg(target_os = "linux")] - echo_cancel: Option, - /// Our pixelpass screen-share host child while sharing (`kill_on_drop`, so it - /// also dies if the session is dropped without an explicit stop). - screenshare_host: Option, - /// pixelpass viewer children we spawned to watch peers' shares, each paired - /// with the share ticket it's viewing so a re-watch of the same share can - /// replace (not stack) its player. Killed on session teardown (each also - /// self-exits when its player window closes). - screenshare_viewers: Vec<(String, tokio::process::Child)>, + /// The screen-share children and the echo-cancel module, held together + /// because their **destruction order** is load-bearing: the AEC module must + /// not unload while a pixelpass host is alive and fanning out (design v3.4 + /// §7.1). `teardown` owns that ordering; see `core::teardown`. + teardown: SessionTeardown, } +/// The session's teardown set, with the echo-cancel guard the platform actually +/// has. On non-Linux there is no AEC module, and `Infallible` makes that +/// structural — the `Option` cannot be `Some`. +#[cfg(target_os = "linux")] +type SessionTeardown = teardown::ScreenshareTeardown< + tokio::process::Child, + crate::audio::echo_cancel::EchoCancelGuard, +>; +#[cfg(not(target_os = "linux"))] +type SessionTeardown = + teardown::ScreenshareTeardown; + impl ActiveSession { async fn shutdown(mut self, audio_backend: Arc) { crate::log_msg("ActiveSession::shutdown started"); // Tear down any screen-share children first so the host stops streaming - // promptly (kill_on_drop is the backstop, but kill explicitly so viewers - // see the stream end without waiting on drop ordering). - if let Some(mut host) = self.screenshare_host.take() { - let _ = host.kill().await; - } - for (_, mut viewer) in self.screenshare_viewers.drain(..) { - let _ = viewer.kill().await; - } + // promptly, and so they are dead *and reaped* well before the AEC guard + // unloads at the end of this function (design v3.4 §7.1). Drop ordering + // is the backstop for the unwind path; this is the path we control. + self.teardown.shutdown_children().await; self.datagram_task.abort(); self.mixer_task.abort(); self.event_task.abort(); @@ -726,8 +729,9 @@ impl ActiveSession { // Unload the echo-cancel module now that the audio streams releasing its // virtual nodes have stopped. (Dropping the guard runs `pactl unload`.) - #[cfg(target_os = "linux")] - drop(self.echo_cancel); + // The screen-share children were killed *and reaped* at the top of this + // function, so nothing pixelpass-side is alive to see the module vanish. + drop(self.teardown); crate::log_msg("Leaving room..."); let _ = self.room_state.leave().await; @@ -1511,7 +1515,17 @@ async fn run_core_loop( biased; maybe_cmd = reliable_rx.recv() => match maybe_cmd { Some(cmd) => cmd, - None => break, + // Every `CoreController`/`CoreCommandSender` is gone — the UI has + // dropped the core. Tear the session down explicitly instead of + // letting it drop on the way out of this function: an implicit + // drop unloads the echo-cancel module without first reaping the + // pixelpass host (design v3.4 §7.2, decision D4). + None => { + if let Some(session) = active_session.take() { + session.shutdown(audio_backend.clone()).await; + } + break; + } }, maybe_wake = besteffort_wake_rx.recv() => match maybe_wake { Some(()) => { @@ -1529,6 +1543,18 @@ async fn run_core_loop( None => continue, } } + // ⚠️ UNREACHABLE BY CONSTRUCTION, twice over — do not mistake this + // for a tested teardown path (phase 0b finding, 2026-07-26): + // 1. this function owns `besteffort_wake_tx` (cloned at the + // `CoreController::new` spawn site, used just above for the + // `has_more` re-arm), so the channel can never close while + // this loop is running; + // 2. even without that, every holder of a wake sender — + // `CoreController` and `CoreCommandSender` — holds + // `reliable_tx` too, and the `biased` select polls that one + // first, so the reliable arm always wins the race to exit. + // The teardown therefore lives in the reliable arm above. If this + // arm is ever made reachable, it needs the same `shutdown().await`. None => break, }, game_change = next_game_change(&mut game_rx) => { @@ -2726,9 +2752,9 @@ async fn run_core_loop( grace_timers, transport: transport.clone(), #[cfg(target_os = "linux")] - echo_cancel: echo_cancel_guard, - screenshare_host: None, - screenshare_viewers: Vec::<(String, tokio::process::Child)>::new(), + teardown: SessionTeardown::new(echo_cancel_guard), + #[cfg(not(target_os = "linux"))] + teardown: SessionTeardown::new(None), }; let self_id = endpoint.id().to_string(); @@ -3403,7 +3429,7 @@ async fn run_core_loop( .await; continue; }; - if session.screenshare_host.is_some() { + if session.teardown.is_sharing() { continue; // already sharing } let bin = match crate::screenshare::pixelpass_path(pixelpass_override.as_deref()) { @@ -3455,7 +3481,7 @@ async fn run_core_loop( { Ok((child, ticket)) => { crate::log_msg("Screen share host started"); - session.screenshare_host = Some(child); + session.teardown.set_host(child); current_sharing = Some(ticket.clone()); let self_state = presence.to_state( is_muted.load(Ordering::Relaxed), @@ -3476,8 +3502,7 @@ async fn run_core_loop( CoreCommand::StopScreenShare => { current_sharing = None; if let Some(session) = &mut active_session { - if let Some(mut child) = session.screenshare_host.take() { - let _ = child.kill().await; + if session.teardown.stop_host().await { crate::log_msg("Screen share host stopped"); } let self_state = presence.to_state( @@ -3505,16 +3530,12 @@ async fn run_core_loop( if let Some(session) = &mut active_session { // Drop viewers whose player window has already closed so the // list only tracks live players. - session - .screenshare_viewers - .retain_mut(|(_, child)| !matches!(child.try_wait(), Ok(Some(_)))); + session.teardown.sweep_exited_viewers(); // One player per share: a second Watch click on a share we're // already viewing is a retry (usually because the first window // froze), so replace the existing player rather than stacking a // second mpv — two players would double the shared audio. - if let Some(pos) = replace_viewer_index(&session.screenshare_viewers, &ticket) { - let (_, mut old) = session.screenshare_viewers.remove(pos); - let _ = old.kill().await; + if session.teardown.replace_viewer(&ticket).await { crate::log_msg("Screen share viewer replaced (re-watch)"); } } @@ -3522,7 +3543,7 @@ async fn run_core_loop( Ok(child) => { crate::log_msg("Screen share viewer started"); if let Some(session) = &mut active_session { - session.screenshare_viewers.push((ticket, child)); + session.teardown.push_viewer(ticket, child); } } Err(e) => { diff --git a/src/core/teardown.rs b/src/core/teardown.rs new file mode 100644 index 0000000..df1de40 --- /dev/null +++ b/src/core/teardown.rs @@ -0,0 +1,485 @@ +//! Destruction-order guarantees for the screen-share children and the +//! echo-cancel module (phase 0b of the screenshare audio-exclusion plan; +//! design v3.4 §7.1–§7.2, decision D4). +//! +//! # The invariant +//! +//! > **The echo-cancel module must not unload while a pixelpass host is alive +//! > and fanning out.** +//! +//! If it does, the AEC's virtual nodes vanish from under a live pixelpass that +//! still holds link proxies and a stale module index. Phase 6 makes this sharp +//! — it is the first phase whose objects live only as long as pixelpass does — +//! so the ordering guarantee has to exist *before* it. +//! +//! Two paths have to honour it, and only one of them is code we get to run: +//! +//! 1. **The explicit path** — [`ScreenshareTeardown::shutdown_children`], awaited +//! by `ActiveSession::shutdown` before the guard is dropped. +//! 2. **The drop/unwind path** — nobody calls anything. The core has numerous +//! `unwrap()` sites and no `panic=abort` profile, so unwind is reachable, and +//! on that path the only thing standing between us and a violated invariant +//! is *field declaration order* plus [`ReapOnDrop`]. +//! +//! Hence the two structural rules enforced here: +//! +//! - `echo_cancel` is the **last declared field** of [`ScreenshareTeardown`]. +//! Rust drops fields in declaration order, so last-declared is last-dropped. +//! This is not a style choice; reversing it reintroduces the bug. +//! - Killing is not enough — a child must be **reaped**. `kill_on_drop(true)` +//! only *signals*; it hands the child to the runtime's orphan queue and +//! returns, which on an unwinding runtime may never be drained. [`ReapOnDrop`] +//! therefore blocks, briefly and boundedly, until the child is actually gone. +//! +//! Everything here is generic over [`ChildProcess`] and over the guard type so +//! the ordering is unit-testable without spawning processes or loading PipeWire +//! modules — the same seam idiom as `replace_viewer_index` and +//! `rebuild_with_fallback` in the parent module. + +use std::future::Future; +use std::time::{Duration, Instant}; + +/// How long [`ReapOnDrop::drop`] will block waiting for a killed child to be +/// reaped before giving up and logging. This runs on the unwind path, so it is +/// a deliberate trade: a bounded stall is preferable to unloading the AEC out +/// from under a live pixelpass, and unbounded blocking in a `Drop` is not. +const REAP_BUDGET: Duration = Duration::from_millis(250); + +/// Poll interval while waiting out [`REAP_BUDGET`]. +const REAP_POLL: Duration = Duration::from_millis(5); + +/// The child-process operations the teardown ordering actually depends on. +/// +/// Deliberately narrow, and deliberately not `ExitStatus`-shaped: the ordering +/// rules care only about *whether* a child has been signalled and *whether* it +/// has been reaped, so the test double is a few lines instead of a fabricated +/// exit status. +pub(super) trait ChildProcess { + /// Signal the child to die. Does **not** wait. + fn start_kill(&mut self) -> std::io::Result<()>; + + /// Poll once. `true` once the child has exited **and been reaped**. + fn try_reap(&mut self) -> bool; + + /// Wait until the child has exited and been reaped. + fn wait_reaped(&mut self) -> impl Future + Send; +} + +impl ChildProcess for tokio::process::Child { + fn start_kill(&mut self) -> std::io::Result<()> { + tokio::process::Child::start_kill(self) + } + + fn try_reap(&mut self) -> bool { + matches!(self.try_wait(), Ok(Some(_))) + } + + async fn wait_reaped(&mut self) { + let _ = self.wait().await; + } +} + +/// A child that is killed **and reaped** when it is dropped. +/// +/// The explicit path calls [`shutdown`](Self::shutdown), which takes the child +/// out, so the `Drop` below is a no-op afterwards. `Drop` is the last-ditch +/// protection for the panic/unwind path only. +pub(super) struct ReapOnDrop { + /// `None` once the child has been reaped through the explicit path. + child: Option, + /// Names the child in the reap-timeout log line. + label: &'static str, +} + +impl ReapOnDrop { + pub(super) fn new(child: C, label: &'static str) -> Self { + Self { + child: Some(child), + label, + } + } + + /// Poll once, without killing. `true` if the child has exited on its own — + /// used to sweep player windows the user has already closed. + pub(super) fn has_exited(&mut self) -> bool { + match &mut self.child { + Some(child) => { + if child.try_reap() { + self.child = None; + true + } else { + false + } + } + // Already reaped through the explicit path. + None => true, + } + } + + /// Kill the child and wait for it to be reaped. Idempotent. + /// + /// The wait is the point: returning after `start_kill` would let the caller + /// proceed to unload the AEC while the child is still running. + pub(super) async fn shutdown(&mut self) { + let Some(mut child) = self.child.take() else { + return; + }; + let _ = child.start_kill(); + child.wait_reaped().await; + } +} + +impl Drop for ReapOnDrop { + fn drop(&mut self) { + let Some(child) = self.child.as_mut() else { + return; + }; + let _ = child.start_kill(); + // `Drop` cannot await, so poll on a bounded budget. See `REAP_BUDGET`. + let deadline = Instant::now() + REAP_BUDGET; + loop { + if child.try_reap() { + return; + } + if Instant::now() >= deadline { + crate::log_msg(&format!( + "teardown: {} did not exit within the reap budget; \ + continuing (the echo-cancel module may unload while it lives)", + self.label + )); + return; + } + std::thread::sleep(REAP_POLL); + } + } +} + +/// Everything in an `ActiveSession` whose **destruction order** is load-bearing. +/// +/// ⚠️ Field order below **is** the invariant. `echo_cancel` is declared last so +/// it is dropped last, after every screen-share child has been killed and +/// reaped. Do not reorder these fields. +pub(super) struct ScreenshareTeardown { + /// Our pixelpass screen-share host child while sharing. + host: Option>, + /// pixelpass viewer children we spawned to watch peers' shares, each paired + /// with the share ticket it is viewing so a re-watch of the same share can + /// replace (not stack) its player. + viewers: Vec<(String, ReapOnDrop)>, + /// Loaded PipeWire echo-cancel module (if enabled); unloads on drop. + /// + /// ⚠️ **LAST FIELD ON PURPOSE** — see the module docs and the struct note. + /// + /// Never read, and that is the design: the guard is held only so that its + /// `Drop` runs, and only so that it runs *here*, last. `dead_code` is right + /// that nothing reads it and wrong that it does nothing. + #[allow(dead_code)] + echo_cancel: Option, +} + +impl ScreenshareTeardown { + pub(super) fn new(echo_cancel: Option) -> Self { + Self { + host: None, + viewers: Vec::new(), + echo_cancel, + } + } + + pub(super) fn is_sharing(&self) -> bool { + self.host.is_some() + } + + pub(super) fn set_host(&mut self, child: C) { + self.host = Some(ReapOnDrop::new(child, "screen-share host")); + } + + /// Stop sharing: kill the host and wait for it to be reaped. + pub(super) async fn stop_host(&mut self) -> bool { + let Some(mut host) = self.host.take() else { + return false; + }; + host.shutdown().await; + true + } + + /// Drop viewers whose player window has already closed, so the list only + /// tracks live players. + pub(super) fn sweep_exited_viewers(&mut self) { + self.viewers.retain_mut(|(_, child)| !child.has_exited()); + } + + /// Kill and reap the viewer already showing `ticket`, if any, so a re-watch + /// replaces its player instead of stacking a second one. + pub(super) async fn replace_viewer(&mut self, ticket: &str) -> bool { + let Some(pos) = super::replace_viewer_index(&self.viewers, ticket) else { + return false; + }; + let (_, mut old) = self.viewers.remove(pos); + old.shutdown().await; + true + } + + pub(super) fn push_viewer(&mut self, ticket: String, child: C) { + self.viewers + .push((ticket, ReapOnDrop::new(child, "screen-share viewer"))); + } + + /// Kill and reap **every** screen-share child, host first so viewers see the + /// stream end promptly. + /// + /// The caller must await this before the echo-cancel guard is dropped. On + /// the drop/unwind path nothing calls it and field order carries the + /// invariant instead. + pub(super) async fn shutdown_children(&mut self) { + if let Some(host) = &mut self.host { + host.shutdown().await; + } + self.host = None; + for (_, viewer) in self.viewers.iter_mut() { + viewer.shutdown().await; + } + self.viewers.clear(); + } +} + +#[cfg(test)] +mod tests { + use super::{ChildProcess, ReapOnDrop, ScreenshareTeardown}; + use std::future::Future; + use std::sync::{Arc, Mutex}; + + type Log = Arc>>; + + fn log() -> Log { + Arc::new(Mutex::new(Vec::new())) + } + + fn entries(log: &Log) -> Vec { + log.lock().unwrap().clone() + } + + fn position(log: &Log, entry: &str) -> Option { + entries(log).iter().position(|e| e == entry) + } + + /// Records the two events the ordering rules turn on. `killed` gates + /// reaping so the double cannot report a reap that never followed a kill. + struct FakeChild { + log: Log, + label: &'static str, + killed: bool, + reaped: bool, + /// When true the child is already dead before anyone kills it — the + /// closed-player-window case that `sweep_exited_viewers` looks for. + exited_on_its_own: bool, + } + + impl FakeChild { + fn new(log: &Log, label: &'static str) -> Self { + Self { + log: log.clone(), + label, + killed: false, + reaped: false, + exited_on_its_own: false, + } + } + + fn already_exited(log: &Log, label: &'static str) -> Self { + Self { + exited_on_its_own: true, + ..Self::new(log, label) + } + } + + fn record(&self, event: &str) { + self.log + .lock() + .unwrap() + .push(format!("{}:{event}", self.label)); + } + + fn mark_reaped(&mut self) { + if !self.reaped { + self.reaped = true; + self.record("reap"); + } + } + } + + impl ChildProcess for FakeChild { + fn start_kill(&mut self) -> std::io::Result<()> { + if !self.killed { + self.killed = true; + self.record("kill"); + } + Ok(()) + } + + fn try_reap(&mut self) -> bool { + if self.killed || self.exited_on_its_own { + self.mark_reaped(); + return true; + } + false + } + + fn wait_reaped(&mut self) -> impl Future + Send { + self.mark_reaped(); + std::future::ready(()) + } + } + + /// Stands in for `EchoCancelGuard`, whose real `Drop` runs `pactl unload`. + struct FakeAec(Log); + + impl Drop for FakeAec { + fn drop(&mut self) { + self.0.lock().unwrap().push("aec:unload".to_string()); + } + } + + fn teardown(log: &Log) -> ScreenshareTeardown { + ScreenshareTeardown::new(Some(FakeAec(log.clone()))) + } + + // --- The drop/unwind path: field order + ReapOnDrop carry the invariant --- + + /// Mutation gate #5 (remove the reap loop from `ReapOnDrop::drop`). + /// + /// Asserts only that dropping a guard reaps, and reaps *after* killing — + /// deliberately says nothing about the AEC, so reversing the struct's field + /// order leaves this test green and only the ordering test below fails. + #[test] + fn dropping_a_guard_kills_and_then_reaps_the_child() { + let log = log(); + drop(ReapOnDrop::new(FakeChild::new(&log, "host"), "host")); + assert_eq!(entries(&log), vec!["host:kill", "host:reap"]); + } + + /// Mutation gate #4 (reverse the field order of `ScreenshareTeardown`). + /// + /// Asserts only kill-before-unload, so removing the reap loop leaves this + /// test green and only the reap test above fails. + #[test] + fn the_aec_unloads_after_the_children_on_the_drop_path() { + let log = log(); + let mut t = teardown(&log); + t.set_host(FakeChild::new(&log, "host")); + t.push_viewer("ticket-A".to_string(), FakeChild::new(&log, "viewer")); + drop(t); + + let unload = position(&log, "aec:unload").expect("the AEC guard must be dropped"); + let host_kill = position(&log, "host:kill").expect("the host must be killed"); + let viewer_kill = position(&log, "viewer:kill").expect("the viewer must be killed"); + assert!( + host_kill < unload, + "the AEC unloaded while the host was alive: {:?}", + entries(&log) + ); + assert!( + viewer_kill < unload, + "the AEC unloaded while a viewer was alive: {:?}", + entries(&log) + ); + } + + /// The whole invariant in one sequence, as documentation. + #[test] + fn the_drop_path_reaps_every_child_before_unloading_the_aec() { + let log = log(); + let mut t = teardown(&log); + t.set_host(FakeChild::new(&log, "host")); + drop(t); + assert_eq!(entries(&log), vec!["host:kill", "host:reap", "aec:unload"]); + } + + // --- The explicit path --- + + /// Mutation gate #3 (remove the wait after the host kill). + #[tokio::test] + async fn explicit_shutdown_reaps_the_host_before_the_aec_can_unload() { + let log = log(); + let mut t = teardown(&log); + t.set_host(FakeChild::new(&log, "host")); + t.push_viewer("ticket-A".to_string(), FakeChild::new(&log, "viewer")); + + t.shutdown_children().await; + + // Reaped by the explicit path — before the guard is anywhere near dropped. + assert_eq!( + entries(&log), + vec!["host:kill", "host:reap", "viewer:kill", "viewer:reap",], + "children must be killed and reaped by the explicit path" + ); + + drop(t); + let unload = position(&log, "aec:unload").expect("the AEC guard must be dropped"); + let host_reap = position(&log, "host:reap").expect("the host must be reaped"); + assert!(host_reap < unload); + } + + #[tokio::test] + async fn explicit_shutdown_is_idempotent_with_the_drop_path() { + let log = log(); + let mut t = teardown(&log); + t.set_host(FakeChild::new(&log, "host")); + t.shutdown_children().await; + drop(t); + // Exactly one kill and one reap: the drop path must not re-signal a + // child the explicit path already took. + assert_eq!(entries(&log), vec!["host:kill", "host:reap", "aec:unload"]); + } + + // --- Host/viewer bookkeeping --- + + #[tokio::test] + async fn stop_host_reports_whether_it_was_sharing() { + let log = log(); + let mut t = teardown(&log); + assert!(!t.is_sharing()); + assert!(!t.stop_host().await, "not sharing: nothing to stop"); + + t.set_host(FakeChild::new(&log, "host")); + assert!(t.is_sharing()); + assert!(t.stop_host().await); + assert!(!t.is_sharing()); + assert_eq!(entries(&log), vec!["host:kill", "host:reap"]); + } + + #[test] + fn sweeping_drops_only_the_players_that_already_closed() { + let log = log(); + let mut t = teardown(&log); + t.push_viewer( + "closed".to_string(), + FakeChild::already_exited(&log, "closed"), + ); + t.push_viewer("live".to_string(), FakeChild::new(&log, "live")); + + t.sweep_exited_viewers(); + + // The live player survives the sweep; only the closed one is dropped, + // and dropping it must not kill anything (it was already gone). + assert_eq!(t.viewers.len(), 1); + assert_eq!(t.viewers[0].0, "live"); + assert_eq!(entries(&log), vec!["closed:reap"]); + } + + #[tokio::test] + async fn re_watching_a_share_replaces_that_player_only() { + let log = log(); + let mut t = teardown(&log); + t.push_viewer("ticket-A".to_string(), FakeChild::new(&log, "a")); + t.push_viewer("ticket-B".to_string(), FakeChild::new(&log, "b")); + + assert!(t.replace_viewer("ticket-A").await); + assert_eq!(entries(&log), vec!["a:kill", "a:reap"]); + assert_eq!(t.viewers.len(), 1); + assert_eq!(t.viewers[0].0, "ticket-B"); + + // A share we are not watching has nothing to replace. + assert!(!t.replace_viewer("ticket-C").await); + } +}