From 98cde2c19b34a4995453fd74317515db271aebe3 Mon Sep 17 00:00:00 2001 From: Mollusk Date: Fri, 21 Aug 2026 16:12:52 -0400 Subject: [PATCH] feat(host): fan out desktop audio through owned links --- src/host/audio_plan.rs | 378 ++++++++++++++++++++++-- src/host/fanout.rs | 550 ++++++++++++++++++++++++++++++++--- src/host/graph.rs | 67 ++++- src/host/observer/adapter.rs | 260 ++++++++++++++++- 4 files changed, 1171 insertions(+), 84 deletions(-) diff --git a/src/host/audio_plan.rs b/src/host/audio_plan.rs index 21b3989..0977cbe 100644 --- a/src/host/audio_plan.rs +++ b/src/host/audio_plan.rs @@ -9,9 +9,9 @@ //! - [`CapturePlan::DesktopExcluding`] contains only [`BareCaptureSink`], which //! has no module ledger or loopback constructor. //! -//! Phase 6 will add retained native PipeWire links to the last variant. Until -//! then it intentionally captures silence through a real, connection-owned -//! sink reached by the hidden phase-0d CLI trigger. +//! Phase 6 feeds the last variant through retained, non-lingering native +//! PipeWire links. The selector remains the hidden phase-0d trigger until the +//! public capability and cross-repository protocol land in phases 7–8. use anyhow::{Context, Result, bail}; use tokio::process::Command; @@ -148,8 +148,15 @@ impl CapturePlan { mod tests { use super::*; use crate::cli::Quality; + use nix::sys::signal::Signal; + use std::io::{BufRead, BufReader, Write}; + use std::process::{Child, Command, Stdio}; + use std::sync::mpsc as std_mpsc; use std::time::{Duration, Instant}; + const PHASE6_SIGKILL_HELPER: &str = "PIXELPASS_PHASE6_SIGKILL_HELPER"; + const PHASE6_READY: &str = "PIXELPASS_PHASE6_READY="; + fn opts( app: bool, strict_audio: bool, @@ -224,7 +231,13 @@ mod tests { .is_ok_and(|output| output.status.success()) } - fn incoming_links(sink_name: &str) -> Result> { + fn value_u64(value: &serde_json::Value) -> Option { + value + .as_u64() + .or_else(|| value.as_str().and_then(|value| value.parse::().ok())) + } + + fn native_links(output_name: &str, sink_name: &str) -> Result> { let output = std::process::Command::new("pw-dump") .output() .context("run pw-dump for the phase-0d graph assertion")?; @@ -239,19 +252,22 @@ mod tests { let objects = objects .as_array() .context("pw-dump root was not an array")?; - let sink_id = objects - .iter() - .find(|object| { - object.get("type").and_then(serde_json::Value::as_str) - == Some("PipeWire:Interface:Node") - && object - .pointer("/info/props/node.name") - .and_then(serde_json::Value::as_str) - == Some(sink_name) - }) - .and_then(|object| object.get("id")) - .and_then(serde_json::Value::as_u64) - .context("bare capture sink was absent from pw-dump")?; + let node_id = |name: &str| { + objects + .iter() + .find(|object| { + object.get("type").and_then(serde_json::Value::as_str) + == Some("PipeWire:Interface:Node") + && object + .pointer("/info/props/node.name") + .and_then(serde_json::Value::as_str) + == Some(name) + }) + .and_then(|object| object.get("id")) + .and_then(value_u64) + }; + let output_id = node_id(output_name).context("fixture output was absent from pw-dump")?; + let sink_id = node_id(sink_name).context("bare capture sink was absent from pw-dump")?; Ok(objects .iter() @@ -261,24 +277,259 @@ mod tests { { return false; } - let Some(input_node) = object.pointer("/info/props/link.input.node") else { + let Some(input_node) = object + .pointer("/info/props/link.input.node") + .and_then(value_u64) + else { return false; }; - input_node.as_u64().or_else(|| { - input_node - .as_str() - .and_then(|value| value.parse::().ok()) - }) == Some(sink_id) + let Some(output_node) = object + .pointer("/info/props/link.output.node") + .and_then(value_u64) + else { + return false; + }; + input_node == sink_id && output_node == output_id }) .cloned() .collect()) } - /// Phase-0d graph assertion: the hidden mode reaches the real native sink, - /// but neither a Pulse default-monitor module nor any native link feeds it. + fn object_serial(object: &serde_json::Value) -> Option { + object + .pointer("/info/props/object.serial") + .and_then(value_u64) + } + + fn object_serial_is_live(serial: u64) -> Result { + let output = Command::new("pw-dump") + .output() + .context("run pw-dump for the Phase-6 SIGKILL 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")?; + Ok(objects + .as_array() + .context("pw-dump root was not an array")? + .iter() + .any(|object| object_serial(object) == Some(serial))) + } + + fn sink_serial(name: &str) -> Result { + let output = Command::new("pw-dump") + .output() + .context("run pw-dump for the Phase-6 sink identity")?; + let objects: serde_json::Value = + serde_json::from_slice(&output.stdout).context("parse pw-dump JSON")?; + objects + .as_array() + .context("pw-dump root was not an array")? + .iter() + .find(|object| { + object + .pointer("/info/props/node.name") + .and_then(serde_json::Value::as_str) + == Some(name) + }) + .and_then(object_serial) + .context("capture sink had no object.serial") + } + + /// Subprocess half of the SIGKILL gate. The outer test kills this process, + /// deliberately skipping every Rust destructor. Its GStreamer child uses + /// the contained spawn path so parent death removes the fixture stream too. + #[tokio::test] + async fn phase6_sigkill_helper() { + if std::env::var_os(PHASE6_SIGKILL_HELPER).is_none() { + return; + } + + let opts = opts(false, false, CaptureMode::DesktopExcluding, false); + let sink_name = crate::repair::plan::sink_name_for(std::process::id()); + let (health, _) = health::channel(); + let _plan = CapturePlan::start(&opts, health) + .await + .expect("start Phase-6 SIGKILL helper plan"); + let fixture_name = format!("pixelpass_phase6_sigkill_{}", std::process::id()); + let mut command = Command::new("gst-launch-1.0"); + command + .args([ + "-q", + "audiotestsrc", + "is-live=true", + "volume=0", + "!", + "audioconvert", + "!", + "audio/x-raw,channels=2", + "!", + "pulsesink", + ]) + .env("PULSE_PROP", format!("node.name={fixture_name}")) + .stdin(Stdio::null()) + .stdout(Stdio::null()) + .stderr(Stdio::inherit()); + // This helper is deliberately SIGKILLed by the outer test, so neither it nor this + // retained child can reach a normal wait. `contained::spawn` gives the fixture a + // parent-death signal; the outer test proves that path removes the fixture and links. + #[allow( + clippy::zombie_processes, + reason = "the SIGKILL cleanup gate intentionally bypasses normal child waiting" + )] + let _fixture = crate::common::contained::spawn(&mut command) + .expect("start contained Phase-6 SIGKILL fixture"); + + let deadline = Instant::now() + Duration::from_secs(5); + let links = loop { + match native_links(&fixture_name, &sink_name) { + Ok(links) + if links.len() == 2 + && links.iter().all(|link| { + link.pointer("/info/state") + .and_then(serde_json::Value::as_str) + == Some("active") + }) => + { + break links; + } + _ if Instant::now() < deadline => { + tokio::time::sleep(Duration::from_millis(20)).await; + } + _ => panic!("SIGKILL fixture never reached two ACTIVE links"), + } + }; + let sink_serial = sink_serial(&sink_name).expect("read capture sink serial"); + let link_serials: Vec = links + .iter() + .map(|link| object_serial(link).expect("owned link object.serial")) + .collect(); + println!( + "{PHASE6_READY}{sink_name} {sink_serial} {},{}", + link_serials[0], link_serials[1] + ); + std::io::stdout().flush().expect("flush Phase-6 readiness"); + std::future::pending::<()>().await; + } + + struct Phase6Host { + child: Option, + pid: u32, + } + + impl Drop for Phase6Host { + 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(); + } + } + } + + #[test] + #[ignore = "live: SIGKILLs a graph-mutating helper; run alone with --test-threads=1"] + fn live_sigkill_removes_capture_sink_and_fanout_links() { + let executable = std::env::current_exe().expect("locate test executable"); + let mut command = Command::new(executable); + command + .args([ + "--exact", + "host::audio_plan::tests::phase6_sigkill_helper", + "--nocapture", + "--test-threads=1", + ]) + .env(PHASE6_SIGKILL_HELPER, "1") + .stdin(Stdio::null()) + .stdout(Stdio::piped()) + .stderr(Stdio::inherit()); + let child = + crate::common::contained::spawn(&mut command).expect("spawn contained Phase-6 helper"); + let pid = child.id(); + let mut host = Phase6Host { + child: Some(child), + pid, + }; + let stdout = host + .child + .as_mut() + .expect("helper stayed owned") + .stdout + .take() + .expect("helper stdout was piped"); + let (line_tx, line_rx) = std_mpsc::sync_channel(1); + std::thread::spawn(move || { + for line in BufReader::new(stdout).lines().map_while(Result::ok) { + if line.contains(PHASE6_READY) { + let _ = line_tx.send(line); + return; + } + } + }); + let line = line_rx + .recv_timeout(Duration::from_secs(10)) + .expect("Phase-6 helper did not report readiness"); + let marker = line + .split_once(PHASE6_READY) + .map(|(_, marker)| marker) + .expect("malformed Phase-6 readiness marker"); + let mut fields = marker.split_whitespace(); + let sink_name = fields + .next() + .expect("readiness omitted sink name") + .to_string(); + let sink_serial: u64 = fields + .next() + .expect("readiness omitted sink serial") + .parse() + .expect("sink serial was not numeric"); + let link_serials: Vec = fields + .next() + .expect("readiness omitted link serials") + .split(',') + .map(|serial| serial.parse().expect("link serial was not numeric")) + .collect(); + assert_eq!(link_serials.len(), 2); + + crate::common::contained::signal_group(pid, Signal::SIGKILL) + .expect("SIGKILL the Phase-6 helper"); + host.child + .as_mut() + .expect("helper stayed owned") + .wait() + .expect("reap the Phase-6 helper"); + host.child.take(); + + let deadline = Instant::now() + Duration::from_secs(3); + while Instant::now() < deadline + && (pulse_source_exists(&format!("{sink_name}.monitor")) + || object_serial_is_live(sink_serial).unwrap_or(true) + || link_serials + .iter() + .any(|serial| object_serial_is_live(*serial).unwrap_or(true))) + { + std::thread::sleep(Duration::from_millis(20)); + } + assert!(!pulse_source_exists(&format!("{sink_name}.monitor"))); + assert!(!object_serial_is_live(sink_serial).expect("inspect sink residue")); + for serial in link_serials { + assert!( + !object_serial_is_live(serial).expect("inspect link residue"), + "fan-out link serial {serial} survived its owning process" + ); + } + } + + /// First Phase-6 live mutation gate: the hidden production path links a + /// late eligible stream into the native sink without constructing the + /// forbidden Pulse default-monitor loopback. Removing the stream revokes + /// the link; shutting down removes the sink and every retained proxy. #[tokio::test] #[ignore = "live: mutates the shared PipeWire graph; run alone with --test-threads=1"] - async fn live_desktop_excluding_sink_has_no_legacy_feed() { + async fn live_desktop_excluding_fans_out_and_cleans_up_native_links() { let opts = opts(false, false, CaptureMode::DesktopExcluding, false); let sink_name = crate::repair::plan::sink_name_for(std::process::id()); assert!(!pulse_source_exists(&format!("{sink_name}.monitor"))); @@ -308,12 +559,67 @@ mod tests { feeding_modules.is_empty(), "DesktopExcluding must not construct a Pulse loopback into its sink: {feeding_modules:?}" ); - assert!( - incoming_links(&sink_name) - .expect("inspect incoming native links") - .is_empty(), - "phase 0d must leave the bare sink unfed until phase 6 owns native links" - ); + let fixture_name = format!("pixelpass_phase6_fixture_{}", std::process::id()); + let mut fixture = tokio::process::Command::new("gst-launch-1.0"); + fixture + .args([ + "-q", + "audiotestsrc", + "is-live=true", + "volume=0", + "!", + "audioconvert", + "!", + "audio/x-raw,channels=2", + "!", + "pulsesink", + ]) + .env("PULSE_PROP", format!("node.name={fixture_name}")) + .stdin(std::process::Stdio::null()) + .stdout(std::process::Stdio::null()) + .stderr(std::process::Stdio::piped()) + .kill_on_drop(true); + let mut fixture = fixture + .spawn() + .expect("start the late eligible audio fixture"); + + let deadline = Instant::now() + Duration::from_secs(5); + let owned_links = loop { + match native_links(&fixture_name, &sink_name) { + Ok(links) + if links.len() == 2 + && links.iter().all(|link| { + link.pointer("/info/state") + .and_then(serde_json::Value::as_str) + == Some("active") + }) => + { + break Some(links); + } + Ok(_) | Err(_) if Instant::now() < deadline => { + tokio::time::sleep(Duration::from_millis(20)).await; + } + _ => break None, + } + }; + + fixture + .kill() + .await + .expect("stop the late eligible fixture"); + let deadline = Instant::now() + Duration::from_secs(2); + let link_was_revoked = loop { + if native_links(&fixture_name, &sink_name) + .map(|links| links.is_empty()) + .unwrap_or(true) + { + break true; + } + if Instant::now() >= deadline { + break false; + } + tokio::time::sleep(Duration::from_millis(20)).await; + }; plan.shutdown().await; let deadline = Instant::now() + Duration::from_secs(2); @@ -322,5 +628,13 @@ mod tests { } assert!(!pulse_source_exists(&format!("{sink_name}.monitor"))); assert!(health.fault().is_none()); + assert!( + owned_links.is_some(), + "the late eligible stereo stream never reached two ACTIVE native fan-out links" + ); + assert!( + link_was_revoked, + "the owned fan-out link survived its output stream" + ); } } diff --git a/src/host/fanout.rs b/src/host/fanout.rs index 00caf51..d91211e 100644 --- a/src/host/fanout.rs +++ b/src/host/fanout.rs @@ -1,29 +1,61 @@ //! Pure Phase-6 fan-out planning. //! //! This module converts one coherent taint decision into exact per-port link -//! specifications. It deliberately cannot create PipeWire objects: the -//! mutation edge will consume this plan on the observer thread, re-evaluate in -//! that same callback, then retain the resulting non-lingering link proxies. +//! specifications and owns the pure state around the mutation edge. PipeWire +//! object creation remains behind [`LinkMutation`], implemented by the registry +//! adapter on its main-loop thread. That split keeps policy and the link-manager +//! matrix deterministic while making it impossible for Tokio to address a +//! recyclable global id directly. //! //! Port identity is `audio.channel`, never registry enumeration order or a //! recyclable global id. An unknown or incompatible layout makes that stream //! unsupported; it never produces a guessed partial capture. -#![allow(dead_code)] - use std::collections::{BTreeMap, BTreeSet}; +use super::aec::{AecConfig, AecState, AecValidator}; +use super::observer::{EventKind, Millis, Projection}; use super::taint::snapshot::{ GlobalId, GraphSnapshot, MediaRole, PortDirection, PortSnapshot, Serial, }; -use super::taint::{Decisions, Reason}; +use super::taint::{Decisions, ExclusionCtx, Reason, StickyState, evaluate}; + +const AEC_VALIDATION_TIMEOUT_MILLIS: Millis = 5_000; + +/// One endpoint identity carried from a coherent graph snapshot to the +/// immediate mutation edge. IDs select the server objects; serials prove the +/// IDs still name those same objects. Neither may be omitted. +#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord)] +pub(super) struct DesiredEndpoint { + pub(super) node_id: GlobalId, + pub(super) node_serial: Serial, + pub(super) port_id: GlobalId, + pub(super) port_serial: Serial, +} #[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord)] pub(super) struct DesiredLink { - pub(super) output_node: GlobalId, - pub(super) output_port: GlobalId, - pub(super) input_node: GlobalId, - pub(super) input_port: GlobalId, + pub(super) output: DesiredEndpoint, + pub(super) input: DesiredEndpoint, +} + +impl DesiredLink { + /// Revalidate every recyclable id against its never-recycled serial. + /// Called by the production mutator immediately before `create_object`. + pub(super) fn is_current(self, snapshot: &GraphSnapshot) -> bool { + endpoint_is_current(snapshot, self.output) && endpoint_is_current(snapshot, self.input) + } +} + +fn endpoint_is_current(snapshot: &GraphSnapshot, endpoint: DesiredEndpoint) -> bool { + snapshot + .node(endpoint.node_serial) + .is_some_and(|node| node.id == endpoint.node_id) + && snapshot.ports().any(|port| { + port.serial == endpoint.port_serial + && port.id == endpoint.port_id + && port.node == endpoint.node_id + }) } #[derive(Clone, Copy, Debug, PartialEq, Eq)] @@ -37,6 +69,7 @@ pub(super) enum PlanIssue { DuplicateOutputChannel, IncompatibleChannelLayout, ExclusiveCapturePort, + LinkCreationFailed, } impl PlanIssue { @@ -51,6 +84,7 @@ impl PlanIssue { Self::DuplicateOutputChannel => "duplicate-output-channel", Self::IncompatibleChannelLayout => "incompatible-channel-layout", Self::ExclusiveCapturePort => "exclusive-capture-port", + Self::LinkCreationFailed => "link-creation-failed", } } } @@ -62,6 +96,165 @@ pub(super) enum StreamPlan { Capture { links: BTreeSet }, } +/// State of one retained link proxy as observed from its bound PipeWire Link. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub(super) enum ManagedLinkState { + Pending, + Active, + Failed, +} + +/// The narrow mutation capability available to the Phase-6 controller. The +/// adapter owns the PipeWire implementation and all proxies. Tests inject a +/// ledger with no PipeWire dependency. +pub(super) trait LinkMutation { + /// Replace the owned set with `desired`, dropping obsolete proxies first. + /// Implementations must revalidate [`DesiredLink::is_current`] against + /// this exact snapshot immediately before each creation. + fn reconcile( + &mut self, + snapshot: &GraphSnapshot, + desired: &BTreeSet, + ) -> BTreeMap; +} + +/// A projection consumer that is explicitly allowed to mutate the graph. +/// Kept separate from the read-only observer sink trait so the Phase-5 audit +/// still has no mutation capability in its type signature. +pub(super) trait MutationProjectionSink: Send { + fn on_projection( + &mut self, + projection: &Projection, + kind: EventKind, + now_us: u64, + links: &mut dyn LinkMutation, + ); +} + +#[derive(Clone, Debug, PartialEq, Eq)] +pub(super) enum StreamCaptureState { + Excluded { reason: Reason }, + Unsupported { issue: PlanIssue }, + Gated { aec: AecState }, + Linking, + Captured, +} + +/// Phases 2, 4, and the pure port planner folded into one per-event controller. +/// It never retains a global id between callbacks: only sticky serial identity +/// and the capture sink's serial survive a recompute. +pub(super) struct FanoutController { + capture_sink: Serial, + validator: AecValidator, + sticky: StickyState, + states: BTreeMap, +} + +impl FanoutController { + pub(super) fn new(capture_sink: Serial, aec: AecConfig) -> Self { + Self { + capture_sink, + validator: AecValidator::new(aec, AEC_VALIDATION_TIMEOUT_MILLIS), + sticky: StickyState::default(), + states: BTreeMap::new(), + } + } + + #[cfg(test)] + fn states(&self) -> &BTreeMap { + &self.states + } +} + +impl MutationProjectionSink for FanoutController { + fn on_projection( + &mut self, + projection: &Projection, + _kind: EventKind, + now_us: u64, + links: &mut dyn LinkMutation, + ) { + self.validator + .observe(&projection.snapshot, projection.graph_ready, now_us / 1_000); + let aec_state = self.validator.state(); + let ctx = ExclusionCtx { + aec_module_id: self.validator.validated_module_id(), + pipewire_pulse_pid: projection.pipewire_pulse_pid, + pixelpass_owned: BTreeSet::from([self.capture_sink]), + graph_ready: projection.graph_ready, + }; + let (decisions, sticky) = evaluate(&projection.snapshot, &ctx, &self.sticky); + self.sticky = sticky; + let plans = plan(&projection.snapshot, &decisions, self.capture_sink); + + let desired: BTreeSet = if self.validator.fan_out_permitted() { + plans + .values() + .filter_map(|plan| match plan { + StreamPlan::Capture { links } => Some(links.iter().copied()), + _ => None, + }) + .flatten() + .collect() + } else { + BTreeSet::new() + }; + let managed = links.reconcile(&projection.snapshot, &desired); + + let next_states: BTreeMap = plans + .into_iter() + .map(|(serial, plan)| { + let state = match plan { + StreamPlan::Excluded { reason } => StreamCaptureState::Excluded { reason }, + StreamPlan::Unsupported { issue } => StreamCaptureState::Unsupported { issue }, + StreamPlan::Capture { .. } if !self.validator.fan_out_permitted() => { + StreamCaptureState::Gated { aec: aec_state } + } + StreamPlan::Capture { links: planned } => { + let link_states: Vec = planned + .iter() + .filter_map(|link| managed.get(link).copied()) + .collect(); + if link_states.len() != planned.len() + || link_states.contains(&ManagedLinkState::Failed) + { + StreamCaptureState::Unsupported { + issue: PlanIssue::LinkCreationFailed, + } + } else if link_states + .iter() + .all(|state| *state == ManagedLinkState::Active) + { + StreamCaptureState::Captured + } else { + StreamCaptureState::Linking + } + } + }; + (serial, state) + }) + .collect(); + for (&serial, state) in &next_states { + if self.states.get(&serial) == Some(state) { + continue; + } + match state { + StreamCaptureState::Unsupported { issue } => tracing::warn!( + serial = serial.0, + reason = issue.code(), + "audio fan-out: stream unsupported" + ), + StreamCaptureState::Captured => tracing::info!( + serial = serial.0, + "audio fan-out: every required link is active" + ), + _ => {} + } + } + self.states = next_states; + } +} + /// Plan every candidate in the decision universe against one capture sink. /// /// The caller must pass `decisions` produced from this exact `snapshot`. The @@ -96,7 +289,9 @@ pub(super) fn plan( snapshot.ports_of(node.id), snapshot.ports_of(sink.expect("validated above").id), node.id, + node.serial, sink.expect("validated above").id, + sink.expect("validated above").serial, ) { Ok(links) => StreamPlan::Capture { links }, Err(issue) => StreamPlan::Unsupported { issue }, @@ -111,7 +306,9 @@ fn pair_ports<'a>( candidate_ports: impl Iterator, capture_ports: impl Iterator, output_node: GlobalId, + output_node_serial: Serial, input_node: GlobalId, + input_node_serial: Serial, ) -> Result, PlanIssue> { let outputs: Vec<&PortSnapshot> = candidate_ports .filter(|port| port.direction == PortDirection::Out && !port.monitor) @@ -129,13 +326,13 @@ fn pair_ports<'a>( return Err(PlanIssue::ExclusiveCapturePort); } - let mut inputs_by_channel: BTreeMap<&str, GlobalId> = BTreeMap::new(); + let mut inputs_by_channel: BTreeMap<&str, &PortSnapshot> = BTreeMap::new(); for port in &inputs { let channel = port .channel .as_deref() .ok_or(PlanIssue::UnidentifiedChannel)?; - if inputs_by_channel.insert(channel, port.id).is_some() { + if inputs_by_channel.insert(channel, port).is_some() { return Err(PlanIssue::AmbiguousCaptureChannel); } } @@ -151,24 +348,40 @@ fn pair_ports<'a>( return Err(PlanIssue::DuplicateOutputChannel); } if channel == "MONO" { - for &input_port in inputs_by_channel.values() { + for &input in inputs_by_channel.values() { links.insert(DesiredLink { - output_node, - output_port: output.id, - input_node, - input_port, + output: DesiredEndpoint { + node_id: output_node, + node_serial: output_node_serial, + port_id: output.id, + port_serial: output.serial, + }, + input: DesiredEndpoint { + node_id: input_node, + node_serial: input_node_serial, + port_id: input.id, + port_serial: input.serial, + }, }); } continue; } - let Some(&input_port) = inputs_by_channel.get(channel) else { + let Some(&input) = inputs_by_channel.get(channel) else { return Err(PlanIssue::IncompatibleChannelLayout); }; links.insert(DesiredLink { - output_node, - output_port: output.id, - input_node, - input_port, + output: DesiredEndpoint { + node_id: output_node, + node_serial: output_node_serial, + port_id: output.id, + port_serial: output.serial, + }, + input: DesiredEndpoint { + node_id: input_node, + node_serial: input_node_serial, + port_id: input.id, + port_serial: input.serial, + }, }); } Ok(links) @@ -177,8 +390,9 @@ fn pair_ports<'a>( #[cfg(test)] mod tests { use super::*; + use crate::host::observer::Readiness; use crate::host::taint::fixture::{Graph, PULSE_PID}; - use crate::host::taint::snapshot::{MediaRole, PortDirection}; + use crate::host::taint::snapshot::{MediaRole, NodeProps, PortDirection}; use crate::host::taint::{ExclusionCtx, StickyState, evaluate}; fn decisions(graph: &Graph) -> (GraphSnapshot, Decisions) { @@ -211,6 +425,77 @@ mod tests { ) } + fn endpoint( + snapshot: &GraphSnapshot, + node_serial: Serial, + port_id: GlobalId, + ) -> DesiredEndpoint { + let node = snapshot.node(node_serial).expect("fixture node"); + let port = snapshot + .ports() + .find(|port| port.id == port_id && port.node == node.id) + .expect("fixture port"); + DesiredEndpoint { + node_id: node.id, + node_serial, + port_id, + port_serial: port.serial, + } + } + + fn projection(graph: &Graph) -> Projection { + projection_from(graph.build()) + } + + fn projection_from(snapshot: GraphSnapshot) -> Projection { + Projection { + snapshot, + pipewire_pulse_pid: Some(PULSE_PID), + graph_ready: true, + readiness: Readiness::Complete, + } + } + + #[derive(Default)] + struct FakeLinks { + held: BTreeSet, + creates: usize, + drops: usize, + next_state: BTreeMap, + } + + impl LinkMutation for FakeLinks { + fn reconcile( + &mut self, + snapshot: &GraphSnapshot, + desired: &BTreeSet, + ) -> BTreeMap { + let current: BTreeSet = desired + .iter() + .copied() + .filter(|link| link.is_current(snapshot)) + .collect(); + self.drops += self.held.difference(¤t).count(); + self.creates += current.difference(&self.held).count(); + self.held = current.clone(); + current + .into_iter() + .map(|link| { + let state = self + .next_state + .get(&link) + .copied() + .unwrap_or(ManagedLinkState::Active); + (link, state) + }) + .collect() + } + } + + fn drive(controller: &mut FanoutController, graph: &Graph, links: &mut FakeLinks) { + controller.on_projection(&projection(graph), EventKind::Graph, 1_000, links); + } + #[test] fn stereo_ports_pair_by_channel_not_enumeration_order() { let (graph, app, sink, [app_fl, app_fr, sink_fl, sink_fr]) = stereo_graph(); @@ -220,16 +505,12 @@ mod tests { Some(&StreamPlan::Capture { links: BTreeSet::from([ DesiredLink { - output_node: snapshot.node(app).unwrap().id, - output_port: app_fl, - input_node: snapshot.node(sink).unwrap().id, - input_port: sink_fl, + output: endpoint(&snapshot, app, app_fl), + input: endpoint(&snapshot, sink, sink_fl), }, DesiredLink { - output_node: snapshot.node(app).unwrap().id, - output_port: app_fr, - input_node: snapshot.node(sink).unwrap().id, - input_port: sink_fr, + output: endpoint(&snapshot, app, app_fr), + input: endpoint(&snapshot, sink, sink_fr), }, ]), }) @@ -254,14 +535,14 @@ mod tests { assert_eq!( links .iter() - .map(|link| link.output_port) + .map(|link| link.output.port_id) .collect::>(), BTreeSet::from([mono]) ); assert_eq!( links .iter() - .map(|link| link.input_port) + .map(|link| link.input.port_id) .collect::>(), BTreeSet::from([left, right]) ); @@ -325,9 +606,210 @@ mod tests { assert_ne!(old, new); assert_eq!( new.iter() - .map(|link| link.input_port) + .map(|link| link.input.port_id) .collect::>(), BTreeSet::from([new_fl, new_fr]) ); } + + #[test] + fn serial_revalidation_rejects_a_recycled_node_id_before_mutation() { + let (mut graph, app, sink, _) = stereo_graph(); + let (snapshot, decisions) = decisions(&graph); + let StreamPlan::Capture { links } = &plan(&snapshot, &decisions, sink)[&app] else { + panic!("fixture must plan links"); + }; + assert!(links.iter().all(|link| link.is_current(&snapshot))); + + let old_app_id = snapshot.node(app).expect("old app").id; + let reborn = graph.node_with_id( + "reborn", + MediaRole::StreamOutput, + old_app_id, + NodeProps::default(), + ); + graph.port_on_channel(reborn, PortDirection::Out, false, Some("FL")); + graph.port_on_channel(reborn, PortDirection::Out, false, Some("FR")); + let recycled = graph.build_without(&[crate::host::taint::fixture::NodeRef { + serial: app, + id: old_app_id, + }]); + assert!( + links.iter().all(|link| !link.is_current(&recycled)), + "the same global id with a new serial must not be addressable" + ); + } + + #[test] + fn duplicate_projection_is_idempotent_and_all_links_must_activate() { + let (graph, app, sink, _) = stereo_graph(); + let first_projection = projection(&graph); + let decisions = evaluate( + &first_projection.snapshot, + &ExclusionCtx { + graph_ready: true, + pipewire_pulse_pid: Some(PULSE_PID), + ..ExclusionCtx::default() + }, + &StickyState::default(), + ) + .0; + let StreamPlan::Capture { links: planned } = + &plan(&first_projection.snapshot, &decisions, sink)[&app] + else { + panic!("fixture must plan links"); + }; + let pending = *planned.iter().next().expect("stereo link"); + + let mut links = FakeLinks::default(); + links.next_state.insert(pending, ManagedLinkState::Pending); + let mut controller = FanoutController::new(sink, AecConfig::Off); + drive(&mut controller, &graph, &mut links); + assert_eq!(links.creates, 2); + assert_eq!( + controller.states()[&app], + StreamCaptureState::Linking, + "one pending channel prevents a captured verdict" + ); + + drive(&mut controller, &graph, &mut links); + assert_eq!( + links.creates, 2, + "the same node must not get a second link set" + ); + links.next_state.clear(); + drive(&mut controller, &graph, &mut links); + assert_eq!(controller.states()[&app], StreamCaptureState::Captured); + } + + #[test] + fn newly_unsafe_ancestry_drops_owned_links_before_any_replacement() { + let (mut graph, app, sink, _) = stereo_graph(); + let mut links = FakeLinks::default(); + let mut controller = FanoutController::new(sink, AecConfig::Off); + drive(&mut controller, &graph, &mut links); + assert_eq!(links.held.len(), 2); + + let call = graph.peerspeak_node("call", 7); + let app_id = graph.build().node(app).expect("app remains live").id; + graph.link( + call, + crate::host::taint::fixture::NodeRef { + serial: app, + id: app_id, + }, + ); + drive(&mut controller, &graph, &mut links); + assert!(links.held.is_empty()); + assert_eq!(links.drops, 2); + assert!(matches!( + controller.states()[&app], + StreamCaptureState::Excluded { .. } + )); + } + + #[test] + fn eligible_late_arrival_is_positively_captured() { + let mut graph = Graph::new(); + let sink = graph.native_virtual_node("pixelpass_capture_test", MediaRole::Sink, 43); + graph.port_on_channel(sink, PortDirection::In, false, Some("FL")); + graph.port_on_channel(sink, PortDirection::In, false, Some("FR")); + let mut links = FakeLinks::default(); + let mut controller = FanoutController::new(sink.serial, AecConfig::Off); + drive(&mut controller, &graph, &mut links); + assert!(links.held.is_empty()); + + let late = graph.app_node("late", MediaRole::StreamOutput, 44); + graph.port_on_channel(late, PortDirection::Out, false, Some("FL")); + graph.port_on_channel(late, PortDirection::Out, false, Some("FR")); + drive(&mut controller, &graph, &mut links); + assert_eq!(links.creates, 2); + assert_eq!( + controller.states()[&late.serial], + StreamCaptureState::Captured + ); + } + + #[test] + fn configured_but_unvalidated_aec_gates_every_mutation() { + let (graph, app, sink, _) = stereo_graph(); + let mut links = FakeLinks::default(); + let mut controller = FanoutController::new(sink, AecConfig::PulseModule(999)); + drive(&mut controller, &graph, &mut links); + assert_eq!(links.creates, 0); + assert!(matches!( + controller.states()[&app], + StreamCaptureState::Gated { + aec: AecState::Validating + } + )); + } + + #[test] + fn link_creation_failure_is_sticky_for_the_same_intent() { + let (graph, app, sink, _) = stereo_graph(); + let first_projection = projection(&graph); + let decisions = evaluate( + &first_projection.snapshot, + &ExclusionCtx { + graph_ready: true, + pipewire_pulse_pid: Some(PULSE_PID), + ..ExclusionCtx::default() + }, + &StickyState::default(), + ) + .0; + let StreamPlan::Capture { links: planned } = + &plan(&first_projection.snapshot, &decisions, sink)[&app] + else { + panic!("fixture must plan links"); + }; + + let mut links = FakeLinks::default(); + links.next_state.insert( + *planned.iter().next().expect("stereo link"), + ManagedLinkState::Failed, + ); + let mut controller = FanoutController::new(sink, AecConfig::Off); + drive(&mut controller, &graph, &mut links); + assert_eq!( + controller.states()[&app], + StreamCaptureState::Unsupported { + issue: PlanIssue::LinkCreationFailed + } + ); + assert_eq!(links.creates, 2); + drive(&mut controller, &graph, &mut links); + assert_eq!( + links.creates, 2, + "an identical failed intent must not hot-loop" + ); + } + + #[test] + fn aec_revocation_drops_previously_active_fanout() { + let (mut graph, app, sink, _) = stereo_graph(); + let aec = graph.module_node("aec-playback", MediaRole::StreamOutput, 999); + graph.port_on_channel(aec, PortDirection::Out, false, Some("FL")); + let mut links = FakeLinks::default(); + let mut controller = FanoutController::new(sink, AecConfig::PulseModule(999)); + drive(&mut controller, &graph, &mut links); + assert_eq!(controller.states()[&app], StreamCaptureState::Captured); + assert_eq!(links.held.len(), 2); + + controller.on_projection( + &projection_from(graph.build_without(&[aec])), + EventKind::Graph, + 2_000, + &mut links, + ); + assert!(links.held.is_empty()); + assert_eq!(links.drops, 2); + assert_eq!( + controller.states()[&app], + StreamCaptureState::Gated { + aec: AecState::Revoked + } + ); + } } diff --git a/src/host/graph.rs b/src/host/graph.rs index 5db2784..765ae25 100644 --- a/src/host/graph.rs +++ b/src/host/graph.rs @@ -7,9 +7,14 @@ //! 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::repair::plan as repair_plan; use anyhow::{Context, Result, bail}; use pipewire::proxy::ProxyT; @@ -38,6 +43,10 @@ const GRAPH_QUIESCE_BUDGET: Duration = Duration::from_secs(2); 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; @@ -129,10 +138,13 @@ pub(super) struct AudioGraphOwner { /// 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 actor connection. The link manager added in phase 6 will feed it through -/// retained native PipeWire links; legacy `Routing` remains a separate type in -/// `host::audio`. +/// 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) before the sink-owning connection. + fanout: Option, graph_owner: Option, monitor_name: String, } @@ -140,11 +152,25 @@ pub(super) struct BareCaptureSink { 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) + 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 fanout = RegistryObserverHandle::spawn_with_mutation_sink( + Box::new(FanoutController::new( + Serial(identity.serial), + AecConfig::Off, + )), + health, + ) + .context("failed to start the desktop-excluding fan-out observer")?; + if let Err(error) = wait_for_fanout_ready(&fanout, &identity).await { + drop(fanout); + graph_owner.shutdown().await; + return Err(error).context("desktop-excluding fan-out did not become ready"); + } Ok(Self { + fanout: Some(fanout), graph_owner: Some(graph_owner), monitor_name: identity.monitor_name, }) @@ -162,6 +188,11 @@ impl BareCaptureSink { } 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(graph_owner) = self.graph_owner.take() && !graph_owner.shutdown().await { @@ -170,6 +201,34 @@ impl BareCaptureSink { } } +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, diff --git a/src/host/observer/adapter.rs b/src/host/observer/adapter.rs index b375291..bda8d97 100644 --- a/src/host/observer/adapter.rs +++ b/src/host/observer/adapter.rs @@ -1,19 +1,23 @@ -//! PipeWire I/O adapter for the pure registry observer. +//! PipeWire I/O adapter for the registry observer and Phase-6 link owner. //! -//! This module owns a read-only PipeWire main-loop thread, translates registry -//! callbacks into [`RegEvent`]s, and publishes the latest [`Projection`] for -//! consumers running outside the PipeWire thread. +//! The default entry points are read-only: they translate registry callbacks +//! into [`RegEvent`]s and publish the latest [`Projection`]. The explicit +//! mutating entry point additionally owns retained non-lingering fan-out Link +//! proxies on that same ordered main-loop thread. The Phase-5 audit can only +//! receive the read-only trait and therefore cannot reach the mutator. use super::classify::{DeviceClaim, DeviceProps}; use super::{ EventKind, LinkEndpoints, NodeObservation, Outcome, Projection, RegEvent, RegistryModel, }; use crate::host::audio::parse_object_serial; +use crate::host::fanout::{DesiredLink, LinkMutation, ManagedLinkState, MutationProjectionSink}; use crate::host::taint::snapshot::{ ClientSnapshot, GlobalId, MediaRole, NodeProps, PortDirection, PortSnapshot, Serial, }; use crate::host::taint::{PEERSPEAK_OWNED_PROP, PEERSPEAK_OWNED_VALUE}; use anyhow::{Context, Result}; +use pipewire::proxy::ProxyT; use pipewire::{self as pw, types::ObjectType}; use std::cell::{Cell, RefCell}; use std::collections::{BTreeMap, BTreeSet, VecDeque}; @@ -65,16 +69,34 @@ impl RegistryObserverHandle { /// exits, which is what lets a sink emit a final summary on shutdown without /// the caller arranging one. pub fn spawn_with_sink(sink: Option>) -> Result { + Self::spawn_with_consumer(ObserverConsumer::ReadOnly(sink)) + } + + /// Spawn the observer with the explicit Phase-6 mutation capability. + /// This is intentionally a different entry point from `spawn_with_sink`: + /// the Phase-5 audit's trait has no path to a PipeWire mutator. + pub(in crate::host) fn spawn_with_mutation_sink( + sink: Box, + health: crate::host::health::Reporter, + ) -> Result { + Self::spawn_with_consumer(ObserverConsumer::Mutating { sink, health }) + } + + fn spawn_with_consumer(consumer: ObserverConsumer) -> Result { + let mutation_health = consumer.mutation_health(); let latest = Arc::new(Mutex::new(None)); let latest_for_thread = Arc::clone(&latest); let (shutdown_tx, shutdown_rx) = pw::channel::channel::<()>(); let thread = std::thread::Builder::new() .name("pixelpass-pw-observer".to_string()) .spawn(move || { - if let Err(e) = run_observer(latest_for_thread, shutdown_rx, sink) { + if let Err(e) = run_observer(latest_for_thread, shutdown_rx, consumer) { tracing::warn!( "registry observer: libpipewire thread exited with error: {e:#}" ); + if let Some(health) = mutation_health { + health.poison(format!("audio fan-out observer failed: {e:#}")); + } } }) .context("failed to spawn libpipewire registry observer thread")?; @@ -96,6 +118,23 @@ impl RegistryObserverHandle { } } +enum ObserverConsumer { + ReadOnly(Option>), + Mutating { + sink: Box, + health: crate::host::health::Reporter, + }, +} + +impl ObserverConsumer { + fn mutation_health(&self) -> Option { + match self { + Self::ReadOnly(_) => None, + Self::Mutating { health, .. } => Some(health.clone()), + } + } +} + impl Drop for RegistryObserverHandle { fn drop(&mut self) { let _ = self.shutdown_tx.send(()); @@ -127,6 +166,167 @@ struct LiveGlobal { bound_proxy: Option, } +struct OwnedFanoutLink { + // Both listeners must unhook callbacks before the proxy is dropped. + _info_listener: pw::link::LinkListener, + _proxy_listener: pw::proxy::ProxyListener, + _proxy: pw::link::Link, +} + +struct PipeWireLinkMutation { + core: pw::core::CoreRc, + owned: BTreeMap, + states: Rc>>, +} + +impl PipeWireLinkMutation { + fn new(core: pw::core::CoreRc) -> Self { + Self { + core, + owned: BTreeMap::new(), + states: Rc::new(RefCell::new(BTreeMap::new())), + } + } + + fn create_link(&mut self, desired: DesiredLink) -> Result { + let output_node = desired.output.node_id.0.to_string(); + let output_port = desired.output.port_id.0.to_string(); + let input_node = desired.input.node_id.0.to_string(); + let input_port = desired.input.port_id.0.to_string(); + let mut props = pw::properties::properties! { + // Load-bearing: dropping the retained proxy or losing this + // connection removes the server-side link. + "object.linger" => "false" + }; + props.insert("link.output.node", output_node.as_str()); + props.insert("link.output.port", output_port.as_str()); + props.insert("link.input.node", input_node.as_str()); + props.insert("link.input.port", input_port.as_str()); + + let link = self + .core + .create_object::("link-factory", &props) + .context("PipeWire link-factory rejected a fan-out link")?; + self.states + .borrow_mut() + .insert(desired, ManagedLinkState::Pending); + + let weak_states = Rc::downgrade(&self.states); + let info_listener = link + .add_listener_local() + .info(move |info| { + let Some(states) = weak_states.upgrade() else { + return; + }; + let state = match info.state() { + pw::link::LinkState::Active => ManagedLinkState::Active, + pw::link::LinkState::Error(error) => { + tracing::warn!(%error, "audio fan-out: owned link entered error state"); + ManagedLinkState::Failed + } + _ => ManagedLinkState::Pending, + }; + states.borrow_mut().insert(desired, state); + }) + .register(); + + let weak_states = Rc::downgrade(&self.states); + let weak_states_for_error = Rc::downgrade(&self.states); + let proxy_listener = link + .upcast_ref() + .add_listener_local() + .removed(move || { + if let Some(states) = weak_states.upgrade() { + states + .borrow_mut() + .insert(desired, ManagedLinkState::Failed); + } + }) + .error(move |seq, res, message| { + tracing::warn!(seq, result = res, %message, "audio fan-out: link proxy error"); + if let Some(states) = weak_states_for_error.upgrade() { + states + .borrow_mut() + .insert(desired, ManagedLinkState::Failed); + } + }) + .register(); + + Ok(OwnedFanoutLink { + _info_listener: info_listener, + _proxy_listener: proxy_listener, + _proxy: link, + }) + } +} + +impl LinkMutation for PipeWireLinkMutation { + fn reconcile( + &mut self, + snapshot: &crate::host::taint::snapshot::GraphSnapshot, + desired: &BTreeSet, + ) -> BTreeMap { + // Revoke before creating. An unsafe link is not allowed to coexist + // briefly with its replacement set. + self.owned.retain(|link, _| desired.contains(link)); + self.states + .borrow_mut() + .retain(|link, _| desired.contains(link)); + + // A Link that reached Error stays failed until the graph changes + // enough to remove this exact serial-guarded intent. Retrying the same + // broken request on every 250 ms observer tick would hot-loop. + let failed: Vec = self + .owned + .keys() + .copied() + .filter(|link| self.states.borrow().get(link) == Some(&ManagedLinkState::Failed)) + .collect(); + for link in failed { + self.owned.remove(&link); + } + + for &link in desired { + if self.owned.contains_key(&link) || self.states.borrow().contains_key(&link) { + continue; + } + // Same-callback revalidation. Global ids are never sufficient: + // every node and port must still carry the serial the planner saw. + if !link.is_current(snapshot) { + self.states + .borrow_mut() + .insert(link, ManagedLinkState::Failed); + continue; + } + match self.create_link(link) { + Ok(owned) => { + self.owned.insert(link, owned); + } + Err(error) => { + tracing::warn!(error = %error, "audio fan-out: could not create link"); + self.states + .borrow_mut() + .insert(link, ManagedLinkState::Failed); + } + } + } + + let states = self.states.borrow(); + desired + .iter() + .map(|link| { + ( + *link, + states + .get(link) + .copied() + .unwrap_or(ManagedLinkState::Failed), + ) + }) + .collect() + } +} + struct ObserverState { model: RegistryModel, latest: Arc>>, @@ -134,7 +334,8 @@ struct ObserverState { /// it are read from `/proc`. last_candidates: BTreeSet, live_globals: BTreeMap>, - sink: Option>, + consumer: ObserverConsumer, + link_mutation: PipeWireLinkMutation, started_at: Instant, } @@ -144,7 +345,8 @@ impl ObserverState { /// `now` are the same clock, not two that drift. fn new( latest: Arc>>, - sink: Option>, + consumer: ObserverConsumer, + link_mutation: PipeWireLinkMutation, started_at: Instant, ) -> Self { Self { @@ -152,7 +354,8 @@ impl ObserverState { latest, last_candidates: BTreeSet::new(), live_globals: BTreeMap::new(), - sink, + consumer, + link_mutation, started_at, } } @@ -202,9 +405,15 @@ impl ObserverState { fn publish(&mut self, kind: EventKind) { let projection = self.model.project(); - if let Some(sink) = self.sink.as_mut() { - let now_us = u64::try_from(self.started_at.elapsed().as_micros()).unwrap_or(u64::MAX); - sink.on_projection(&projection, kind, now_us); + let now_us = u64::try_from(self.started_at.elapsed().as_micros()).unwrap_or(u64::MAX); + match &mut self.consumer { + ObserverConsumer::ReadOnly(Some(sink)) => { + sink.on_projection(&projection, kind, now_us); + } + ObserverConsumer::ReadOnly(None) => {} + ObserverConsumer::Mutating { sink, .. } => { + sink.on_projection(&projection, kind, now_us, &mut self.link_mutation); + } } // Published after the sink has seen it, so the projection is moved // rather than cloned — the snapshot is the largest thing the observer @@ -281,7 +490,7 @@ impl ObserverState { fn run_observer( latest: Arc>>, shutdown_rx: pw::channel::Receiver<()>, - sink: Option>, + consumer: ObserverConsumer, ) -> Result<()> { let started_at = Instant::now(); let main_loop = @@ -292,16 +501,28 @@ fn run_observer( .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(ObserverState::new(latest, sink, started_at))); + let mutation_health = consumer.mutation_health(); + let link_mutation = PipeWireLinkMutation::new(core.clone()); + let state = Rc::new(RefCell::new(ObserverState::new( + latest, + consumer, + link_mutation, + started_at, + ))); + let shutdown_observed = Rc::new(Cell::new(false)); + let shutdown_for_receiver = Rc::clone(&shutdown_observed); let main_loop_for_shutdown = main_loop.clone(); let _shutdown_receiver = shutdown_rx.attach(main_loop.loop_(), move |()| { + shutdown_for_receiver.set(true); main_loop_for_shutdown.quit(); }); let pending_sync = Rc::new(Cell::new(None)); let pending_sync_for_done = Rc::clone(&pending_sync); let state_for_done = Rc::clone(&state); + let main_loop_for_error = main_loop.clone(); + let mutation_health_for_error = mutation_health.clone(); let _core_listener = core .add_listener_local() .done(move |id, seq| { @@ -310,7 +531,7 @@ fn run_observer( state_for_done.borrow_mut().apply(RegEvent::ServerSynced); } }) - .error(|id, seq, res, message| { + .error(move |id, seq, res, message| { tracing::warn!( id, seq, @@ -318,6 +539,12 @@ fn run_observer( %message, "registry observer: PipeWire core error" ); + if let Some(health) = &mutation_health_for_error { + health.poison(format!( + "audio fan-out PipeWire core error {res}: {message}" + )); + main_loop_for_error.quit(); + } }) .register(); @@ -643,6 +870,11 @@ fn run_observer( tracing::info!("registry observer: pw thread running"); main_loop.run(); tracing::info!("registry observer: pw thread exiting"); + if let Some(health) = mutation_health + && !shutdown_observed.get() + { + health.poison("audio fan-out observer exited without a shutdown request"); + } Ok(()) }