//! Connection-owned PipeWire graph actor. //! //! The actor owns one non-lingering native capture sink plus the existing //! per-application `target.object` router. The sink is therefore scoped to the //! actor's PipeWire connection: closing or losing that connection removes it, //! including after SIGKILL. Tokio never receives a bare PipeWire global id to //! mutate; matching, serial validation, and metadata writes stay ordered on the //! PipeWire main-loop thread. use super::aec::AecConfig; use super::audio::parse_object_serial; use super::fanout::FanoutController; use super::health; use super::observer::Readiness; use super::observer::adapter::RegistryObserverHandle; use super::owned_thread::OwnedThread; use super::taint::snapshot::{GlobalId, Serial}; use crate::common::output::AudioExclusionEvent; use crate::repair::plan as repair_plan; use anyhow::{Context, Result, bail}; use pipewire::proxy::ProxyT; use pipewire::{self as pw, types::ObjectType}; use std::cell::RefCell; use std::collections::BTreeMap; use std::io; use std::process::Stdio; use std::rc::Rc; use std::sync::Arc; use std::sync::atomic::{AtomicBool, AtomicU8, Ordering}; use std::time::{Duration, Instant}; use tokio::sync::{mpsc, oneshot}; /// A running main loop normally stops in a few milliseconds. Two seconds is /// deliberately generous while still bounding host teardown. const GRAPH_RUNNING_STOP_BUDGET: Duration = Duration::from_secs(2); /// A thread stuck inside PipeWire initialisation cannot process `Stop`. Keep a /// separate, longer policy deadline for that different failure mode. const GRAPH_STARTING_STOP_BUDGET: Duration = Duration::from_secs(5); const GRAPH_NATIVE_READY_BUDGET: Duration = Duration::from_secs(5); const GRAPH_QUIESCE_BUDGET: Duration = Duration::from_secs(2); /// Warm measurements on this host are single-digit milliseconds. This wider /// bound covers a lagging pipewire-pulse bridge without borrowing the actor's /// thread or turning a missing Pulse namespace into an unbounded wait. const PULSE_MONITOR_READY_BUDGET: Duration = Duration::from_secs(3); const PULSE_PROBE_BUDGET: Duration = Duration::from_millis(500); const PULSE_PROBE_INTERVAL: Duration = Duration::from_millis(20); /// The observer has its own sticky 2 s readiness deadline. This outer budget /// includes thread startup and lets us turn either a timeout or a missing /// capture-sink observation into a failed `DesktopExcluding` construction. const FANOUT_READY_BUDGET: Duration = Duration::from_secs(3); const GRAPH_STARTING: u8 = 0; const GRAPH_RUNNING: u8 = 1; const GRAPH_EXITED: u8 = 2; /// Exact native capture-sink shape used by S4. #[derive(Clone, Debug, PartialEq, Eq)] pub(super) struct CaptureSinkSpec { name: String, monitor_name: String, } impl CaptureSinkSpec { pub(super) fn for_pid(pid: u32) -> Self { let name = repair_plan::sink_name_for(pid); let monitor_name = format!("{name}.monitor"); Self { name, monitor_name } } pub(super) fn name(&self) -> &str { &self.name } pub(super) fn monitor_name(&self) -> &str { &self.monitor_name } fn properties(&self) -> pw::properties::PropertiesBox { let mut props = pw::properties::properties! { "factory.name" => "support.null-audio-sink", "media.class" => "Audio/Sink", "audio.channels" => "2", "audio.position" => "[FL,FR]", "node.virtual" => "true", "monitor.channel-volumes" => "true", // Load-bearing: the remote object must die with this connection. "object.linger" => "false" }; props.insert("node.name", self.name.as_str()); props } } /// Identity established from the created proxy's bound id and that exact /// registry global's serial. `node.name` is intentionally not an identity /// input: PipeWire accepts duplicate names and Pulse selects the older one. #[derive(Clone, Debug, PartialEq, Eq)] pub(super) struct SinkIdentity { pub(super) name: String, pub(super) monitor_name: String, pub(super) global_id: u32, pub(super) serial: u64, } #[derive(Debug)] pub(super) enum GraphEvent { FirstRoutedStream, LastRoutedStreamGone, } #[derive(Clone, Copy, Debug, PartialEq, Eq)] pub(super) enum QuiesceOutcome { Confirmed, Unconfirmed, } enum Cmd { /// Stop producing route events, restore routes still owned by PixelPass, /// and acknowledge the writes with a core round-trip. The sink remains /// alive so Tokio can unload every dependent Pulse loopback first. Quiesce { ack: oneshot::Sender }, /// Restore any still-owned routes, wait for a core round-trip, then quit /// the main loop. Dropping the connection removes the native non-lingering /// sink only after those restoration writes have reached the server. Stop, } /// Tokio-side owner of the PipeWire actor thread. pub(super) struct AudioGraphOwner { cmd_tx: pw::channel::Sender, thread: OwnedThread, phase: Arc, identity: SinkIdentity, } /// A connection-owned capture sink with no Pulse-module ledger and no API for /// constructing the legacy default-monitor loopback. /// /// This is the load-bearing type boundary for the phase-0d /// `DesktopExcluding` plan: owning this value proves the sink exists, but the /// only operations available are reading its monitor name and shutting down /// its graph connections. Phase 6 feeds it through a separate observer /// connection that retains non-lingering native PipeWire links; legacy /// `Routing` remains a separate type in `host::audio`. pub(super) struct BareCaptureSink { // Drop order is load-bearing: close the fan-out observer connection (and // therefore every non-lingering link), then detach the status task, before // the sink-owning connection. fanout: Option, status_forwarder: Option>, graph_owner: Option, monitor_name: String, } impl BareCaptureSink { pub(super) async fn start(health: health::Reporter) -> Result { let spec = CaptureSinkSpec::for_pid(std::process::id()); let (graph_owner, _event_rx) = AudioGraphOwner::start(None, spec, health.clone()) .await .context("failed to start the bare connection-owned capture sink")?; let identity = graph_owner.identity().clone(); let (status_tx, status_rx) = mpsc::unbounded_channel(); let status_forwarder = tokio::spawn(forward_audio_exclusion_status(status_rx)); let fanout = match RegistryObserverHandle::spawn_with_mutation_sink( Box::new(FanoutController::with_status_sender( Serial(identity.serial), AecConfig::Off, status_tx, )), health, ) .context("failed to start the desktop-excluding fan-out observer") { Ok(fanout) => fanout, Err(error) => { let _ = status_forwarder.await; graph_owner.shutdown().await; return Err(error); } }; if let Err(error) = wait_for_fanout_ready(&fanout, &identity).await { drop(fanout); let _ = status_forwarder.await; graph_owner.shutdown().await; return Err(error).context("desktop-excluding fan-out did not become ready"); } Ok(Self { fanout: Some(fanout), status_forwarder: Some(status_forwarder), graph_owner: Some(graph_owner), monitor_name: identity.monitor_name, }) } #[cfg(test)] pub(super) fn sink_name(&self) -> &str { self.monitor_name .strip_suffix(".monitor") .expect("capture-sink monitor names always end in .monitor") } pub(super) fn monitor_name(&self) -> &str { &self.monitor_name } pub(super) async fn shutdown(mut self) { // Joining the observer closes its PipeWire connection and drops every // retained non-lingering link before the capture sink can disappear. if let Some(fanout) = self.fanout.take() { drop(fanout); } if let Some(status_forwarder) = self.status_forwarder.take() && let Err(error) = status_forwarder.await { tracing::warn!(%error, "audio fan-out: status forwarder task failed"); } if let Some(graph_owner) = self.graph_owner.take() && !graph_owner.shutdown().await { tracing::warn!("audio capture: bare AudioGraphOwner shutdown was not confirmed"); } } } async fn forward_audio_exclusion_status( mut status_rx: mpsc::UnboundedReceiver, ) { while let Some(event) = status_rx.recv().await { event.emit(); } } async fn wait_for_fanout_ready( observer: &RegistryObserverHandle, identity: &SinkIdentity, ) -> Result<()> { let deadline = Instant::now() + FANOUT_READY_BUDGET; loop { if let Some(projection) = observer.latest() { if projection.readiness == Readiness::TimedOut { bail!("registry observer reached its sticky readiness timeout"); } let sink_is_current = projection .snapshot .node(Serial(identity.serial)) .is_some_and(|node| node.id == GlobalId(identity.global_id)); if projection.graph_ready && sink_is_current { return Ok(()); } } if Instant::now() >= deadline { bail!( "capture sink serial {} was not present in a graph-ready projection within {FANOUT_READY_BUDGET:?}", identity.serial ); } tokio::time::sleep(PULSE_PROBE_INTERVAL).await; } } impl AudioGraphOwner { pub(super) async fn start( filter_name: Option, spec: CaptureSinkSpec, health: health::Reporter, ) -> Result<(Self, mpsc::UnboundedReceiver)> { let (cmd_tx, cmd_rx) = pw::channel::channel::(); let (event_tx, event_rx) = mpsc::unbounded_channel::(); let (ready_tx, ready_rx) = oneshot::channel::(); let phase = Arc::new(AtomicU8::new(GRAPH_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 spec_for_thread = spec.clone(); let thread = std::thread::Builder::new() .name("pixelpass-audio-graph".to_string()) .spawn(move || { let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| { run_graph( filter_name, spec_for_thread, cmd_rx, event_tx, ready_tx, Arc::clone(&phase_for_thread), Arc::clone(&shutdown_for_thread), health_for_thread.clone(), ) })); phase_for_thread.store(GRAPH_EXITED, Ordering::Release); match result { Ok(result) => report_graph_exit( &health_for_thread, shutdown_for_thread.load(Ordering::Acquire), result, ), Err(_) => { health_for_thread.poison("AudioGraphOwner thread panicked"); } } }) .context("failed to spawn AudioGraphOwner thread")?; // Construct the owner before the first await. Cancellation therefore // keeps the OS handle owned and sends Stop from Drop. let mut owner = Self { cmd_tx, thread: OwnedThread::new("AudioGraphOwner thread", thread, health.clone()), phase, // Replaced after the native handshake; this value is never exposed. identity: SinkIdentity { name: spec.name.clone(), monitor_name: spec.monitor_name.clone(), global_id: u32::MAX, serial: 0, }, }; let identity = match tokio::time::timeout(GRAPH_NATIVE_READY_BUDGET, ready_rx).await { Ok(Ok(identity)) => identity, Ok(Err(_)) => { owner.shutdown().await; bail!("AudioGraphOwner exited before the native sink became ready"); } Err(_) => { owner.shutdown().await; bail!( "AudioGraphOwner did not establish the native sink within {GRAPH_NATIVE_READY_BUDGET:?}" ); } }; owner.identity = identity.clone(); // A PipeWire core round-trip proves only this client's namespace. The // GStreamer `pulsesrc` path uses pipewire-pulse, so readiness requires a // separate Pulse lookup from Tokio, never a blocking subprocess on the // actor thread. let pulse_ready = wait_for_pulse_monitor(&identity.monitor_name).await; if let Err(error) = pulse_ready { owner.shutdown().await; return Err(error).context("native capture sink was not exported to Pulse"); } tracing::info!( global_id = identity.global_id, serial = identity.serial, sink = %identity.name, monitor = %identity.monitor_name, "audio graph: connection-owned capture sink ready" ); Ok((owner, event_rx)) } pub(super) fn identity(&self) -> &SinkIdentity { &self.identity } pub(super) async fn quiesce(&mut self) -> QuiesceOutcome { let (ack_tx, ack_rx) = oneshot::channel(); if self.cmd_tx.send(Cmd::Quiesce { ack: ack_tx }).is_err() { tracing::warn!("audio graph: actor exited before routes could be quiesced"); return QuiesceOutcome::Unconfirmed; } match tokio::time::timeout(GRAPH_QUIESCE_BUDGET, ack_rx).await { Ok(Ok(true)) => QuiesceOutcome::Confirmed, Ok(Ok(false) | Err(_)) => { tracing::warn!("audio graph: route restoration was not acknowledged"); QuiesceOutcome::Unconfirmed } Err(_) => { tracing::warn!( "audio graph: route restoration was not acknowledged within {GRAPH_QUIESCE_BUDGET:?}" ); QuiesceOutcome::Unconfirmed } } } pub(super) async fn shutdown(mut self) -> bool { let _ = self.cmd_tx.send(Cmd::Stop); let budget = graph_shutdown_budget(self.phase.load(Ordering::Acquire)); self.thread.join_within(budget).await } } impl Drop for AudioGraphOwner { fn drop(&mut self) { // If async construction or shutdown is cancelled, wake the MainLoop // before OwnedThread poisons/quarantines the still-owned handle. let _ = self.cmd_tx.send(Cmd::Stop); } } fn graph_shutdown_budget(phase: u8) -> Duration { if phase == GRAPH_STARTING { GRAPH_STARTING_STOP_BUDGET } else { GRAPH_RUNNING_STOP_BUDGET } } fn report_graph_exit(health: &health::Reporter, shutdown_observed: bool, result: Result<()>) { match result { Ok(()) if shutdown_observed => {} Ok(()) => { health.poison("AudioGraphOwner exited without a Stop command"); } Err(error) => { tracing::warn!("audio graph: actor thread exited with error: {error:#}"); health.poison(format!("AudioGraphOwner failed: {error:#}")); } } } async fn wait_for_pulse_monitor(monitor_name: &str) -> Result { let started = Instant::now(); let deadline = started + PULSE_MONITOR_READY_BUDGET; loop { let mut command = tokio::process::Command::new("pactl"); command .args(["get-source-volume", monitor_name]) .stdin(Stdio::null()) .stdout(Stdio::null()) .stderr(Stdio::piped()) .kill_on_drop(true); let last_error = match tokio::time::timeout(PULSE_PROBE_BUDGET, command.output()).await { Ok(Ok(output)) if output.status.success() => return Ok(started.elapsed()), Ok(Ok(output)) => String::from_utf8_lossy(&output.stderr).trim().to_string(), Ok(Err(error)) if error.kind() == io::ErrorKind::NotFound => { return Err(error).context("failed to run `pactl get-source-volume`"); } Ok(Err(error)) => error.to_string(), Err(_) => format!("pactl probe exceeded {PULSE_PROBE_BUDGET:?}"), }; if Instant::now() >= deadline { bail!( "Pulse monitor {monitor_name:?} did not become ready within {PULSE_MONITOR_READY_BUDGET:?}: {last_error}" ); } tokio::time::sleep(PULSE_PROBE_INTERVAL).await; } } #[derive(Clone, Debug, PartialEq, Eq)] struct ObservedNode { global_id: u32, serial: u64, /// Diagnostic only. Serial equality is the identity gate; epochs do not /// invalidate an observation merely because unrelated graph traffic moved. epoch: u64, } #[derive(Clone, Debug, Default, PartialEq, Eq)] struct MetadataValue { type_: Option, value: Option, } #[derive(Clone, Debug)] struct RouteRecord { observed: ObservedNode, prior_target: MetadataValue, } struct BoundMetadata { // Listener first: its Drop unhooks callbacks that reference the proxy. _listener: pw::metadata::MetadataListener, metadata: pw::metadata::Metadata, } struct OwnedSink { // Listener first for the same teardown-order invariant as BoundMetadata. _listener: pw::proxy::ProxyListener, _proxy: pw::node::Node, } #[derive(Default)] struct NativeReadiness { bound_global_id: Option, globals: BTreeMap, core_synced: bool, } impl NativeReadiness { fn identity(&self, spec: &CaptureSinkSpec) -> Option { let global_id = self.bound_global_id?; let serial = *self.globals.get(&global_id)?; self.core_synced.then(|| SinkIdentity { name: spec.name().to_string(), monitor_name: spec.monitor_name().to_string(), global_id, serial, }) } } struct ActorState { spec: CaptureSinkSpec, filter_lower: Option, readiness: NativeReadiness, ready_tx: Option>, identity: Option, epoch: u64, pending: BTreeMap, routed: BTreeMap, current_targets: BTreeMap, default_metadata: Option>, event_tx: Option>, initial_sync: Option, quiesce_sync: Option<(pw::spa::utils::result::AsyncSeq, oneshot::Sender)>, stop_sync: Option, closing: bool, } impl ActorState { fn publish_ready_if_complete(&mut self) { if self.identity.is_some() { return; } let Some(identity) = self.readiness.identity(&self.spec) else { return; }; self.identity = Some(identity.clone()); if let Some(tx) = self.ready_tx.take() { let _ = tx.send(identity); } } fn observe_global(&mut self, id: u32, serial: u64) { self.epoch = self.epoch.wrapping_add(1); self.readiness.globals.insert(id, serial); self.publish_ready_if_complete(); } fn observe_bound(&mut self, id: u32) { self.readiness.bound_global_id = Some(id); self.publish_ready_if_complete(); } fn observe_initial_sync(&mut self) { self.readiness.core_synced = true; self.publish_ready_if_complete(); } fn note_matching_stream(&mut self, id: u32, serial: u64) { if self.closing { return; } self.pending.insert( id, ObservedNode { global_id: id, serial, epoch: self.epoch, }, ); } fn remove_global(&mut self, id: u32) -> bool { self.epoch = self.epoch.wrapping_add(1); self.readiness.globals.remove(&id); self.pending.remove(&id); self.current_targets.remove(&id); let was_routed = !self.routed.is_empty(); self.routed.remove(&id); if was_routed && self.routed.is_empty() && let Some(tx) = &self.event_tx { let _ = tx.send(GraphEvent::LastRoutedStreamGone); } self.readiness.bound_global_id == Some(id) } fn update_metadata( &mut self, subject: u32, key: Option<&str>, type_: Option<&str>, value: Option<&str>, ) { match key { Some("target.object") => { self.current_targets.insert( subject, MetadataValue { type_: type_.map(str::to_string), value: value.map(str::to_string), }, ); } None => self.current_targets.clear(), _ => {} } } } fn observation_is_current(globals: &BTreeMap, observed: &ObservedNode) -> bool { globals.get(&observed.global_id) == Some(&observed.serial) } fn try_flush(state: &Rc>) { let (metadata, actions, notify_first) = { let mut state = state.borrow_mut(); if state.closing || !state.readiness.core_synced { return; } let Some(identity) = state.identity.clone() else { return; }; let Some(metadata) = state.default_metadata.clone() else { return; }; let owned_target = identity.serial.to_string(); let was_empty = state.routed.is_empty(); let pending = std::mem::take(&mut state.pending); let mut actions = Vec::new(); for observed in pending.into_values() { if !observation_is_current(&state.readiness.globals, &observed) || state.routed.contains_key(&observed.global_id) { continue; } let prior_target = state .current_targets .get(&observed.global_id) .cloned() .unwrap_or_default(); state.current_targets.insert( observed.global_id, MetadataValue { type_: Some("Spa:Id".to_string()), value: Some(owned_target.clone()), }, ); state.routed.insert( observed.global_id, RouteRecord { observed: observed.clone(), prior_target, }, ); actions.push(observed.global_id); } let notify_first = was_empty && !actions.is_empty(); (metadata, actions, notify_first) }; let owned_target = state .borrow() .identity .as_ref() .expect("identity existed while staging routes") .serial .to_string(); for id in &actions { metadata .metadata .set_property(*id, "target.object", Some("Spa:Id"), Some(&owned_target)); tracing::info!(node_id = *id, sink_serial = %owned_target, "audio graph: stream routed"); } if notify_first && let Some(tx) = state.borrow().event_tx.as_ref() { let _ = tx.send(GraphEvent::FirstRoutedStream); } } fn restoration_for( record: &RouteRecord, current: Option<&MetadataValue>, owned_target: &str, ) -> Option { let current = current?; (current.value.as_deref() == Some(owned_target)).then(|| record.prior_target.clone()) } fn prepare_quiesce( state: &Rc>, ) -> (Option>, Vec<(u32, MetadataValue)>) { let mut state = state.borrow_mut(); state.closing = true; state.event_tx.take(); state.pending.clear(); let metadata = state.default_metadata.clone(); let owned_target = state .identity .as_ref() .map(|identity| identity.serial.to_string()) .unwrap_or_default(); let actions = state .routed .values() .filter_map(|record| { restoration_for( record, state.current_targets.get(&record.observed.global_id), &owned_target, ) .map(|value| (record.observed.global_id, value)) }) .collect(); state.routed.clear(); (metadata, actions) } #[allow(clippy::too_many_arguments)] fn run_graph( filter_name: Option, spec: CaptureSinkSpec, cmd_rx: pw::channel::Receiver, event_tx: mpsc::UnboundedSender, ready_tx: oneshot::Sender, phase: Arc, shutdown_observed: Arc, health: health::Reporter, ) -> Result<()> { let main_loop = pw::main_loop::MainLoopRc::new(None).context("pw main loop construction failed")?; let context = pw::context::ContextRc::new(&main_loop, None).context("pw context construction failed")?; let core = context .connect_rc(None) .context("pw core connect failed (is the daemon running?)")?; let registry = core.get_registry_rc().context("pw get_registry failed")?; let state = Rc::new(RefCell::new(ActorState { spec: spec.clone(), filter_lower: filter_name.map(|name| name.to_ascii_lowercase()), readiness: NativeReadiness::default(), ready_tx: Some(ready_tx), identity: None, epoch: 0, pending: BTreeMap::new(), routed: BTreeMap::new(), current_targets: BTreeMap::new(), default_metadata: None, event_tx: Some(event_tx), initial_sync: None, quiesce_sync: None, stop_sync: None, closing: false, })); let main_loop_for_cmd = main_loop.clone(); let core_for_cmd = core.clone(); let state_for_cmd = Rc::clone(&state); let shutdown_for_cmd = Arc::clone(&shutdown_observed); let _cmd_receiver = cmd_rx.attach(main_loop.loop_(), move |cmd| match cmd { Cmd::Quiesce { ack } => { let (metadata, restorations) = prepare_quiesce(&state_for_cmd); if let Some(metadata) = metadata { for (id, prior) in restorations { metadata.metadata.set_property( id, "target.object", prior.type_.as_deref(), prior.value.as_deref(), ); } } match core_for_cmd.sync(0) { Ok(seq) => state_for_cmd.borrow_mut().quiesce_sync = Some((seq, ack)), Err(error) => { tracing::warn!("audio graph: quiesce core.sync failed: {error}"); let _ = ack.send(false); } } } Cmd::Stop => { shutdown_for_cmd.store(true, Ordering::Release); let (metadata, restorations) = prepare_quiesce(&state_for_cmd); if let Some(metadata) = metadata { for (id, prior) in restorations { metadata.metadata.set_property( id, "target.object", prior.type_.as_deref(), prior.value.as_deref(), ); } } if let Some((_, ack)) = state_for_cmd.borrow_mut().quiesce_sync.take() { let _ = ack.send(false); } match core_for_cmd.sync(0) { Ok(seq) => state_for_cmd.borrow_mut().stop_sync = Some(seq), Err(error) => { tracing::warn!( "audio graph: final route-restoration core.sync failed: {error}" ); main_loop_for_cmd.quit(); } } } }); let state_for_done = Rc::clone(&state); let main_loop_for_done = main_loop.clone(); let state_for_core_error = Rc::clone(&state); let main_loop_for_core_error = main_loop.clone(); let health_for_core_error = health.clone(); let _core_listener = core .add_listener_local() .done(move |id, seq| { if id != pw::core::PW_ID_CORE { return; } let (initial_completed, stop_completed) = { let mut state = state_for_done.borrow_mut(); let initial_completed = state.initial_sync == Some(seq); if initial_completed { state.initial_sync = None; state.observe_initial_sync(); } if state .quiesce_sync .as_ref() .is_some_and(|(pending, _)| *pending == seq) && let Some((_, ack)) = state.quiesce_sync.take() { let _ = ack.send(true); } let stop_completed = state.stop_sync == Some(seq); if stop_completed { state.stop_sync = None; } (initial_completed, stop_completed) }; if initial_completed { // Existing matching streams may have been observed before the // initial round-trip. Readiness becoming true is itself the // event that makes those observations routable. try_flush(&state_for_done); } if stop_completed { main_loop_for_done.quit(); } }) .error(move |id, seq, res, message| { tracing::warn!(id, seq, result = res, %message, "audio graph: PipeWire core error"); if let Some((_, ack)) = state_for_core_error.borrow_mut().quiesce_sync.take() { let _ = ack.send(false); } health_for_core_error.poison(format!("PipeWire core error {res}: {message}")); main_loop_for_core_error.quit(); }) .register(); let registry_weak = registry.downgrade(); let state_for_global = Rc::clone(&state); let state_for_remove = Rc::clone(&state); let main_loop_for_remove = main_loop.clone(); let health_for_remove = health.clone(); let shutdown_for_remove = Arc::clone(&shutdown_observed); let _registry_listener = registry .add_listener_local() .global(move |obj| { match obj.type_ { ObjectType::Node => { let Some(props) = obj.props.as_ref() else { return; }; let Some(serial) = props.get("object.serial").and_then(parse_object_serial) else { return; }; { let mut state = state_for_global.borrow_mut(); state.observe_global(obj.id, serial); let matches_filter = state.filter_lower.as_ref().is_some_and(|filter| { props.get("media.class") == Some("Stream/Output/Audio") && props .get("application.name") .is_some_and(|app| app.eq_ignore_ascii_case(filter)) }); if matches_filter && state.readiness.bound_global_id != Some(obj.id) { state.note_matching_stream(obj.id, serial); } } try_flush(&state_for_global); } ObjectType::Metadata => { let Some(props) = obj.props.as_ref() else { return; }; if props.get("metadata.name") != Some("default") { return; } let Some(registry) = registry_weak.upgrade() else { return; }; let metadata: pw::metadata::Metadata = match registry.bind(obj) { Ok(metadata) => metadata, Err(error) => { tracing::warn!("audio graph: bind default metadata failed: {error}"); return; } }; let weak_state = Rc::downgrade(&state_for_global); let listener = metadata .add_listener_local() .property(move |subject, key, type_, value| { if let Some(state) = weak_state.upgrade() { state .borrow_mut() .update_metadata(subject, key, type_, value); // A property callback may be the last missing // observation for a stream queued earlier. try_flush(&state); } 0 }) .register(); state_for_global.borrow_mut().default_metadata = Some(Rc::new(BoundMetadata { _listener: listener, metadata, })); try_flush(&state_for_global); } _ => {} } }) .global_remove(move |id| { let own_sink_removed = state_for_remove.borrow_mut().remove_global(id); if own_sink_removed && !shutdown_for_remove.load(Ordering::Acquire) { health_for_remove.poison("connection-owned capture sink disappeared unexpectedly"); main_loop_for_remove.quit(); } }) .register(); let sink = core .create_object::("adapter", &spec.properties()) .context("could not create the connection-owned capture sink")?; let state_for_bound = Rc::clone(&state); let main_loop_for_proxy_remove = main_loop.clone(); let health_for_proxy_remove = health.clone(); let shutdown_for_proxy_remove = Arc::clone(&shutdown_observed); let main_loop_for_proxy_error = main_loop.clone(); let health_for_proxy_error = health.clone(); let shutdown_for_proxy_error = Arc::clone(&shutdown_observed); let listener = sink .upcast_ref() .add_listener_local() .bound(move |global_id| { state_for_bound.borrow_mut().observe_bound(global_id); }) .removed(move || { if !shutdown_for_proxy_remove.load(Ordering::Acquire) { health_for_proxy_remove.poison("connection-owned capture-sink proxy was removed"); main_loop_for_proxy_remove.quit(); } }) .error(move |seq, res, message| { tracing::warn!(seq, result = res, %message, "audio graph: capture-sink proxy error"); if !shutdown_for_proxy_error.load(Ordering::Acquire) { health_for_proxy_error.poison(format!( "connection-owned capture-sink proxy error {res}: {message}" )); main_loop_for_proxy_error.quit(); } }) .register(); let _owned_sink = OwnedSink { _listener: listener, _proxy: sink, }; let initial_sync = core .sync(0) .context("AudioGraphOwner initial core.sync failed")?; state.borrow_mut().initial_sync = Some(initial_sync); phase.store(GRAPH_RUNNING, Ordering::Release); tracing::info!(sink = %spec.name, "audio graph: actor main loop running"); main_loop.run(); tracing::info!(sink = %spec.name, "audio graph: actor main loop exiting"); Ok(()) } #[cfg(test)] mod tests { use super::*; use crate::cli::{HostOpts, Quality}; use crate::repair::plan::{ModuleObservation, Shape, classify}; use nix::sys::signal::Signal; use std::io::{BufRead, BufReader, Write}; use std::process::{Child, Command, Stdio}; use std::sync::mpsc as std_mpsc; const S5_HOST_HELPER: &str = "PIXELPASS_S5_HOST_HELPER"; const S5_READY_PREFIX: &str = "PIXELPASS_S5_READY="; fn spec(name: &str) -> CaptureSinkSpec { CaptureSinkSpec { name: name.to_string(), monitor_name: format!("{name}.monitor"), } } #[test] fn capture_sink_spec_is_non_lingering_and_pulse_compatible() { let spec = CaptureSinkSpec::for_pid(4242); assert_eq!(spec.name(), "pixelpass_capture_4242"); assert_eq!(spec.monitor_name(), "pixelpass_capture_4242.monitor"); let props = spec.properties(); let props = props.dict(); assert_eq!(props.get("factory.name"), Some("support.null-audio-sink")); assert_eq!(props.get("node.name"), Some("pixelpass_capture_4242")); assert_eq!(props.get("media.class"), Some("Audio/Sink")); assert_eq!(props.get("audio.channels"), Some("2")); assert_eq!(props.get("audio.position"), Some("[FL,FR]")); assert_eq!(props.get("node.virtual"), Some("true")); assert_eq!(props.get("monitor.channel-volumes"), Some("true")); assert_eq!(props.get("object.linger"), Some("false")); } #[test] fn readiness_correlates_the_proxy_bound_id_not_a_duplicate_name() { let spec = spec("duplicate_name_is_not_identity"); let mut readiness = NativeReadiness::default(); // A different global could carry the same node.name. Readiness has no // name input at all and therefore cannot accidentally select it. readiness.globals.insert(11, 101); readiness.bound_global_id = Some(12); readiness.core_synced = true; assert!(readiness.identity(&spec).is_none()); readiness.globals.insert(12, 202); assert_eq!( readiness.identity(&spec), Some(SinkIdentity { name: spec.name.clone(), monitor_name: spec.monitor_name.clone(), global_id: 12, serial: 202, }) ); } #[test] fn readiness_requires_the_core_round_trip_as_well_as_identity() { let spec = spec("roundtrip_gate"); let mut readiness = NativeReadiness::default(); readiness.globals.insert(7, 70); readiness.bound_global_id = Some(7); assert!(readiness.identity(&spec).is_none()); readiness.core_synced = true; assert_eq!(readiness.identity(&spec).unwrap().serial, 70); } #[test] fn route_gate_revalidates_serial_but_not_unrelated_epoch_churn() { let observed = ObservedNode { global_id: 9, serial: 90, epoch: 1, }; let mut globals = BTreeMap::from([(9, 90)]); assert!(observation_is_current(&globals, &observed)); // Epoch is diagnostic only: unrelated graph traffic must not invalidate // a still-live identity. let later_epoch = ObservedNode { epoch: 500, ..observed.clone() }; assert!(observation_is_current(&globals, &later_epoch)); // A recycled global id is a different object and must never be routed // from the stale observation. globals.insert(9, 91); assert!(!observation_is_current(&globals, &observed)); } #[test] fn route_restoration_never_overwrites_a_later_owner() { let record = RouteRecord { observed: ObservedNode { global_id: 9, serial: 90, epoch: 1, }, prior_target: MetadataValue { type_: Some("Spa:Id".to_string()), value: Some("44".to_string()), }, }; let ours = MetadataValue { type_: Some("Spa:Id".to_string()), value: Some("55".to_string()), }; assert_eq!( restoration_for(&record, Some(&ours), "55"), Some(record.prior_target.clone()) ); let user_override = MetadataValue { type_: Some("Spa:Id".to_string()), value: Some("66".to_string()), }; assert_eq!(restoration_for(&record, Some(&user_override), "55"), None); assert_eq!(restoration_for(&record, None, "55"), None); } #[test] fn shutdown_has_distinct_starting_and_running_budgets() { assert_eq!( graph_shutdown_budget(GRAPH_STARTING), GRAPH_STARTING_STOP_BUDGET ); assert_eq!( graph_shutdown_budget(GRAPH_RUNNING), GRAPH_RUNNING_STOP_BUDGET ); assert!(GRAPH_STARTING_STOP_BUDGET > GRAPH_RUNNING_STOP_BUDGET); } #[test] fn graph_exit_is_healthy_only_after_stop_was_observed() { let (clean, _) = health::channel(); report_graph_exit(&clean, true, Ok(())); assert!(clean.fault().is_none()); let (unexpected, _) = health::channel(); report_graph_exit(&unexpected, false, Ok(())); assert_eq!( unexpected.fault().as_deref(), Some("AudioGraphOwner exited without a Stop command") ); let (failed, _) = health::channel(); report_graph_exit(&failed, false, Err(anyhow::anyhow!("fixture failure"))); assert!( failed .fault() .as_deref() .is_some_and(|reason| reason.contains("fixture failure")) ); } #[tokio::test] async fn cancelling_shutdown_keeps_the_graph_thread_owned_and_poisons() { let (cmd_tx, _cmd_rx) = pw::channel::channel::(); let (release_tx, release_rx) = std_mpsc::channel(); let (health, _) = health::channel(); let thread = std::thread::spawn(move || { let _ = release_rx.recv(); }); let owner = AudioGraphOwner { cmd_tx, thread: OwnedThread::new("graph cancellation fixture", thread, health.clone()), phase: Arc::new(AtomicU8::new(GRAPH_STARTING)), identity: SinkIdentity { name: "fixture".to_string(), monitor_name: "fixture.monitor".to_string(), global_id: 1, serial: 1, }, }; assert!( tokio::time::timeout(Duration::from_millis(20), owner.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"); } async fn pulse_source_exists(name: &str) -> bool { tokio::process::Command::new("pactl") .args(["get-source-volume", name]) .stdout(Stdio::null()) .stderr(Stdio::null()) .status() .await .is_ok_and(|status| status.success()) } fn whole_desktop_opts() -> HostOpts { HostOpts { window: false, app: None, strict_audio: false, display_server: None, quality: Quality::Auto, bitrate: None, framerate: None, max_height: None, no_hwencode: false, max_viewers: None, interactive: false, capture_mode: crate::cli::CaptureMode::Legacy, legacy_null_sink: false, relay: None, } } /// Subprocess half of the S5 two-host gate. The outer test invokes this /// exact test twice, so each helper owns a distinct PipeWire connection, /// process id, native sink, and ownership-tagged Pulse loopback. SIGINT /// requests an ordinary teardown; SIGKILL deliberately skips it. #[tokio::test] async fn s5_connection_owned_host_helper() { if std::env::var_os(S5_HOST_HELPER).is_none() { return; } let (health, _) = health::channel(); let routing = super::super::audio::Routing::start(&whole_desktop_opts(), health.clone()) .await .expect("S5 helper routing starts"); assert!(health.fault().is_none()); println!( "{S5_READY_PREFIX}{} {}", std::process::id(), routing.sink_name() ); std::io::stdout().flush().expect("flush S5 readiness"); tokio::signal::ctrl_c() .await .expect("S5 parent requests a clean SIGINT stop"); routing.shutdown().await; assert!(health.fault().is_none()); } struct S5Host { child: Option, pid: u32, sink_name: String, } impl S5Host { fn spawn() -> Result { let executable = std::env::current_exe().context("locate the test executable")?; let mut command = Command::new(executable); command .args([ "--exact", "host::graph::tests::s5_connection_owned_host_helper", "--nocapture", "--test-threads=1", ]) .env(S5_HOST_HELPER, "1") .stdin(Stdio::null()) .stdout(Stdio::piped()) .stderr(Stdio::inherit()); let child = crate::common::contained::spawn(&mut command) .context("spawn a contained S5 host helper")?; let child_pid = child.id(); // From this point every early return owns a kill-and-reap fallback. // A readiness timeout must not strand a graph-mutating helper. let mut host = Self { child: Some(child), pid: child_pid, sink_name: String::new(), }; let stdout = host .child .as_mut() .expect("S5 host was just constructed") .stdout .take() .context("S5 helper stdout was not piped")?; let (line_tx, line_rx) = std_mpsc::sync_channel(1); std::thread::spawn(move || { let mut readiness_sent = false; for line in BufReader::new(stdout) .lines() .map_while(std::result::Result::ok) { if !readiness_sent && line.contains(S5_READY_PREFIX) { readiness_sent = true; let _ = line_tx.send(Some(line)); } // Keep draining after readiness. Dropping this pipe while the // helper still owns it makes libtest fail its final result // write with BrokenPipe, masking a clean graph shutdown. } if !readiness_sent { let _ = line_tx.send(None); } }); let ready_line = line_rx .recv_timeout(Duration::from_secs(10)) .context("S5 helper did not report readiness within 10 seconds")? .context("S5 helper exited before reporting readiness")?; let marker = ready_line .split_once(S5_READY_PREFIX) .map(|(_, value)| value) .context("malformed S5 readiness marker")?; let mut fields = marker.split_whitespace(); let reported_pid: u32 = fields .next() .context("S5 readiness omitted the pid")? .parse() .context("S5 readiness pid was not numeric")?; let sink_name = fields .next() .context("S5 readiness omitted the sink name")? .to_string(); if reported_pid != child_pid { bail!("S5 helper reported pid {reported_pid}, expected {child_pid}"); } host.sink_name = sink_name; Ok(host) } fn kill_and_reap(&mut self) -> Result<()> { if self.child.is_none() { return Ok(()); } crate::common::contained::signal_group(self.pid, Signal::SIGKILL) .context("SIGKILL the first S5 host")?; let mut child = self.child.take().expect("S5 child stayed owned"); child.wait().context("reap the first S5 host")?; Ok(()) } fn interrupt_and_reap_within(&mut self, budget: Duration) -> Result { if self.child.is_none() { return Ok(Duration::ZERO); } let started = Instant::now(); crate::common::contained::signal_group(self.pid, Signal::SIGINT) .context("SIGINT the surviving S5 host")?; let deadline = started + budget; loop { let status = self .child .as_mut() .expect("S5 child stayed owned across the wait") .try_wait() .context("poll the S5 helper")?; if let Some(status) = status { if status.success() { self.child.take(); return Ok(started.elapsed()); } bail!("S5 helper SIGINT stop exited with {status}"); } if Instant::now() >= deadline { let _ = crate::common::contained::signal_group(self.pid, Signal::SIGKILL); if let Some(mut child) = self.child.take() { let _ = child.wait(); } bail!("S5 helper did not stop cleanly within {budget:?}"); } std::thread::sleep(Duration::from_millis(20)); } } } impl Drop for S5Host { fn drop(&mut self) { if let Some(mut child) = self.child.take() { let _ = crate::common::contained::signal_group(self.pid, Signal::SIGKILL); let _ = child.wait(); } } } fn capture_sink_serial(name: &str) -> Result> { let output = Command::new("pw-dump") .output() .context("run pw-dump for the S5 ownership gate")?; if !output.status.success() { bail!( "pw-dump failed: {}", String::from_utf8_lossy(&output.stderr).trim() ); } let objects: serde_json::Value = serde_json::from_slice(&output.stdout).context("parse pw-dump JSON")?; let Some(objects) = objects.as_array() else { bail!("pw-dump root was not an array"); }; for object in objects { if object.get("type").and_then(serde_json::Value::as_str) != Some("PipeWire:Interface:Node") { continue; } let Some(props) = object.pointer("/info/props") else { continue; }; if props.get("node.name").and_then(serde_json::Value::as_str) != Some(name) { continue; } let Some(serial) = props.get("object.serial") else { bail!("capture sink {name:?} had no object.serial"); }; let serial = serial .as_u64() .or_else(|| serial.as_str().and_then(|value| value.parse::().ok())); return serial .map(Some) .context("capture sink object.serial was not an integer"); } Ok(None) } fn owned_loopbacks(pid: u32) -> Result> { let mut pulse = crate::repair::introspect::PulseSession::connect() .context("connect to the local Pulse server")?; let modules = pulse.list_modules().context("list Pulse modules")?; Ok(modules .into_iter() .filter_map(|module| { classify(&ModuleObservation::new( module.id, &module.name, &module.args, )) }) .filter(|fingerprint| { fingerprint.pid == pid && fingerprint.shape == Shape::LoopbackIntoCapture }) .map(|fingerprint| fingerprint.id) .collect()) } fn wait_for_sink_state(name: &str, expected_serial: Option) -> Result<()> { let deadline = Instant::now() + Duration::from_secs(3); loop { if capture_sink_serial(name)? == expected_serial { return Ok(()); } if Instant::now() >= deadline { bail!("capture sink {name:?} did not reach serial state {expected_serial:?}"); } std::thread::sleep(Duration::from_millis(20)); } } /// S5 ownership exit gate: two real processes own two independent native /// sinks. SIGKILL must remove exactly the killed process's sink, and /// ownership-aware repair must remove its orphaned Pulse loopback without /// touching the still-live host. The survivor then exercises ordinary /// teardown so the gate itself leaves no graph residue. #[tokio::test] #[ignore = "live: starts two graph-owning processes, SIGKILLs one, and runs --repair"] async fn live_two_host_sigkill_and_repair_preserve_the_survivor() { let mut first = S5Host::spawn().expect("start the first S5 host"); let mut second = S5Host::spawn().expect("start the second S5 host"); let first_serial = capture_sink_serial(&first.sink_name) .expect("observe the first S5 sink") .expect("the first S5 sink is present"); let second_serial = capture_sink_serial(&second.sink_name) .expect("observe the second S5 sink") .expect("the second S5 sink is present"); assert_ne!(first.sink_name, second.sink_name); assert_ne!(first_serial, second_serial); assert_eq!(owned_loopbacks(first.pid).unwrap().len(), 1); let second_modules = owned_loopbacks(second.pid).unwrap(); assert_eq!(second_modules.len(), 1); first.kill_and_reap().expect("SIGKILL the first S5 host"); wait_for_sink_state(&first.sink_name, None) .expect("the killed host's connection-owned sink disappears"); wait_for_sink_state(&second.sink_name, Some(second_serial)) .expect("the live host's connection-owned sink remains"); crate::repair::run(false) .await .expect("ownership-aware repair succeeds"); assert!(owned_loopbacks(first.pid).unwrap().is_empty()); assert_eq!( owned_loopbacks(second.pid).unwrap(), second_modules, "repair must not unload the live host's module" ); wait_for_sink_state(&second.sink_name, Some(second_serial)) .expect("repair must not disturb the live host's native sink"); let stop_elapsed = second .interrupt_and_reap_within(Duration::from_secs(2)) .expect("the surviving S5 host honours SIGINT inside PeerSpeak's grace"); println!("S5_ACTIVE_SIGINT_ELAPSED_MS={}", stop_elapsed.as_millis()); assert!( stop_elapsed < Duration::from_secs(2), "active graph teardown reached PeerSpeak's SIGKILL fallback boundary: {stop_elapsed:?}" ); wait_for_sink_state(&second.sink_name, None) .expect("the surviving host's sink disappears after clean shutdown"); assert!(owned_loopbacks(second.pid).unwrap().is_empty()); } #[tokio::test] #[ignore = "creates a real connection-owned PipeWire sink; serialize live audio tests"] async fn live_sink_survives_quiesce_and_dies_with_its_owner() { let spec = CaptureSinkSpec::for_pid(std::process::id()); assert!( !pulse_source_exists(spec.monitor_name()).await, "live fixture name is already present" ); let (health, _) = health::channel(); let (mut owner, _events) = AudioGraphOwner::start(None, spec.clone(), health.clone()) .await .expect("connection-owned sink starts"); assert_ne!(owner.identity().global_id, u32::MAX); assert!(owner.identity().serial > 0); assert!(pulse_source_exists(spec.monitor_name()).await); assert_eq!(owner.quiesce().await, QuiesceOutcome::Confirmed); assert!( pulse_source_exists(spec.monitor_name()).await, "quiesce must retain the sink until dependent modules are removed" ); assert!(owner.shutdown().await); let deadline = Instant::now() + Duration::from_secs(2); while pulse_source_exists(spec.monitor_name()).await && Instant::now() < deadline { tokio::time::sleep(Duration::from_millis(20)).await; } assert!( !pulse_source_exists(spec.monitor_name()).await, "sink must disappear with the actor connection" ); assert!(health.fault().is_none()); } }