diff --git a/src/host/audio_plan.rs b/src/host/audio_plan.rs index 21da94d..88ca789 100644 --- a/src/host/audio_plan.rs +++ b/src/host/audio_plan.rs @@ -21,6 +21,39 @@ use super::graph::BareCaptureSink; use super::health; use crate::cli::{CaptureMode, HostOpts}; +trait CapturePlanBackend { + async fn resolve_default_monitor(&mut self) -> Result; + async fn start_routing(&mut self, opts: &HostOpts, health: health::Reporter) + -> Result; + async fn start_bare_capture_sink( + &mut self, + health: health::Reporter, + ) -> Result; +} + +struct SystemCapturePlanBackend; + +impl CapturePlanBackend for SystemCapturePlanBackend { + async fn resolve_default_monitor(&mut self) -> Result { + DefaultMonitor::resolve().await + } + + async fn start_routing( + &mut self, + opts: &HostOpts, + health: health::Reporter, + ) -> Result { + Routing::start(opts, health).await + } + + async fn start_bare_capture_sink( + &mut self, + health: health::Reporter, + ) -> Result { + BareCaptureSink::start(health).await + } +} + #[derive(Clone, Copy, Debug, PartialEq, Eq)] pub(super) enum CapturePlanKind { LegacyDesktop, @@ -99,17 +132,35 @@ pub(super) enum CapturePlan { impl CapturePlan { pub(super) async fn start(opts: &HostOpts, health: health::Reporter) -> Result { + let mut backend = SystemCapturePlanBackend; + Self::start_with_backend(opts, health, &mut backend).await + } + + /// One production construction path with an injectable system boundary. + /// + /// Keeping failure policy here makes rows 8d/8e falsifiable in release + /// builds: a capture-sink or readiness failure must escape this function. + /// It must never be translated into a second attempt through + /// `LegacyDesktop`, whose default-monitor source would reintroduce the + /// audio this mode exists to exclude. + async fn start_with_backend( + opts: &HostOpts, + health: health::Reporter, + backend: &mut B, + ) -> Result { match CapturePlanKind::resolve(opts)? { CapturePlanKind::LegacyDesktop => Ok(Self::LegacyDesktop { - source: DefaultMonitor::resolve().await?, + source: backend.resolve_default_monitor().await?, }), CapturePlanKind::PerApp => Ok(Self::PerApp { - routing: Routing::start(opts, health) + routing: backend + .start_routing(opts, health) .await .context("audio routing setup failed")?, }), CapturePlanKind::DesktopExcluding => Ok(Self::DesktopExcluding { - capture_sink: BareCaptureSink::start(health) + capture_sink: backend + .start_bare_capture_sink(health) .await .context("desktop-excluding capture-sink setup failed")?, }), @@ -148,6 +199,8 @@ impl CapturePlan { mod tests { use super::*; use crate::cli::Quality; + use crate::host::observer::{Projection, Readiness}; + use crate::host::taint::fixture::Graph; use nix::sys::signal::Signal; use std::io::{BufRead, BufReader, Write}; use std::process::{Child, Command, Stdio}; @@ -224,6 +277,99 @@ mod tests { assert!(matches!(plan, CapturePlan::LegacyDesktop { .. })); } + #[derive(Clone, Copy, Debug, PartialEq, Eq)] + enum BackendCall { + DefaultMonitor, + Routing, + BareCaptureSink, + } + + struct FailingDesktopBackend { + failure: Option, + calls: Vec, + } + + impl CapturePlanBackend for FailingDesktopBackend { + async fn resolve_default_monitor(&mut self) -> Result { + self.calls.push(BackendCall::DefaultMonitor); + panic!("a DesktopExcluding failure must not resolve the legacy default monitor") + } + + async fn start_routing( + &mut self, + _opts: &HostOpts, + _health: health::Reporter, + ) -> Result { + self.calls.push(BackendCall::Routing); + panic!("a DesktopExcluding failure must not construct legacy Routing") + } + + async fn start_bare_capture_sink( + &mut self, + _health: health::Reporter, + ) -> Result { + self.calls.push(BackendCall::BareCaptureSink); + Err(self.failure.take().expect("one injected failure")) + } + } + + async fn assert_desktop_failure_is_closed(failure: anyhow::Error, expected_cause: &str) { + let opts = opts(false, false, CaptureMode::DesktopExcluding, false); + assert_eq!( + CapturePlanKind::resolve(&opts).expect("fixture mode resolves"), + CapturePlanKind::DesktopExcluding + ); + let mut backend = FailingDesktopBackend { + failure: Some(failure), + calls: Vec::new(), + }; + let (health, _) = health::channel(); + let error = match CapturePlan::start_with_backend(&opts, health, &mut backend).await { + Ok(_) => panic!("DesktopExcluding unexpectedly recovered through another plan"), + Err(error) => error, + }; + assert_eq!(backend.calls, vec![BackendCall::BareCaptureSink]); + let chain = format!("{error:#}"); + assert!(chain.contains("desktop-excluding capture-sink setup failed")); + assert!( + chain.contains(expected_cause), + "unexpected error chain: {chain}" + ); + } + + #[tokio::test] + async fn capture_sink_creation_failure_never_falls_back_to_legacy_desktop() { + assert_desktop_failure_is_closed( + anyhow::anyhow!("injected capture-sink creation failure"), + "injected capture-sink creation failure", + ) + .await; + } + + #[tokio::test] + async fn readiness_epoch_timeout_never_falls_back_to_legacy_desktop() { + let identity = super::super::graph::SinkIdentity { + name: "pixelpass_capture_timeout_fixture".to_string(), + monitor_name: "pixelpass_capture_timeout_fixture.monitor".to_string(), + global_id: 77, + serial: 88, + }; + let projection = Projection { + snapshot: Graph::new().build(), + pipewire_pulse_pid: None, + graph_ready: false, + readiness: Readiness::TimedOut, + }; + let readiness_error = super::super::graph::fanout_readiness(Some(&projection), &identity) + .expect_err("the observer's sticky timeout must fail readiness"); + + assert_desktop_failure_is_closed( + readiness_error, + "registry observer reached its sticky readiness timeout", + ) + .await; + } + fn pulse_source_exists(name: &str) -> bool { std::process::Command::new("pactl") .args(["get-source-volume", name]) diff --git a/src/host/fanout.rs b/src/host/fanout.rs index 30010ba..5b8a43c 100644 --- a/src/host/fanout.rs +++ b/src/host/fanout.rs @@ -51,6 +51,20 @@ impl DesiredLink { } } +/// Revalidate every desired endpoint against the projection at the mutation +/// edge. A plan made from an older graph is not authority to address a global +/// id after that id has been recycled. +pub(super) fn revalidated_links( + snapshot: &GraphSnapshot, + desired: &BTreeSet, +) -> BTreeSet { + desired + .iter() + .copied() + .filter(|link| link.is_current(snapshot)) + .collect() +} + fn endpoint_is_current(snapshot: &GraphSnapshot, endpoint: DesiredEndpoint) -> bool { snapshot .node(endpoint.node_serial) @@ -587,11 +601,7 @@ mod tests { snapshot: &GraphSnapshot, desired: &BTreeSet, ) -> BTreeMap { - let current: BTreeSet = desired - .iter() - .copied() - .filter(|link| link.is_current(snapshot)) - .collect(); + let current = revalidated_links(snapshot, desired); self.drops += self.held.difference(¤t).count(); self.creates += current.difference(&self.held).count(); self.held = current.clone(); @@ -801,6 +811,45 @@ mod tests { ); } + #[test] + fn identity_change_between_planning_and_create_makes_zero_create_calls() { + let (mut graph, app, sink, _) = stereo_graph(); + let (planned_snapshot, decisions) = decisions(&graph); + let StreamPlan::Capture { links: planned } = + &plan(&planned_snapshot, &decisions, sink)[&app] + else { + panic!("fixture must produce a clean link plan"); + }; + + // Recycle the output node's global id after evaluation but before the + // mutation edge sees the request. The old serial must authorize + // neither channel even though both numeric ids still exist. + let old_app_id = planned_snapshot.node(app).expect("old app").id; + let reborn = graph.node_with_id( + "reborn-between-plan-and-create", + 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 mutation_snapshot = graph.build_without(&[crate::host::taint::fixture::NodeRef { + serial: app, + id: old_app_id, + }]); + + let mut links = FakeLinks::default(); + let states = links.reconcile(&mutation_snapshot, planned); + assert_eq!(links.creates, 0, "no stale intent may reach create_link"); + assert!(links.held.is_empty()); + assert!( + states + .values() + .all(|state| *state == ManagedLinkState::Failed), + "every stale channel must fail closed" + ); + } + #[test] fn duplicate_projection_is_idempotent_and_all_links_must_activate() { let (graph, app, sink, _) = stereo_graph(); diff --git a/src/host/graph.rs b/src/host/graph.rs index 2c64bd9..e1d2639 100644 --- a/src/host/graph.rs +++ b/src/host/graph.rs @@ -11,8 +11,8 @@ 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::{FanoutControl, RegistryObserverHandle}; +use super::observer::{Projection, Readiness}; use super::owned_thread::OwnedThread; use super::taint::snapshot::{GlobalId, Serial}; use crate::common::output::AudioExclusionEvent; @@ -293,17 +293,9 @@ async fn wait_for_fanout_ready( ) -> 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(()); - } + let projection = observer.latest(); + if fanout_readiness(projection.as_ref(), identity)? { + return Ok(()); } if Instant::now() >= deadline { bail!( @@ -315,6 +307,23 @@ async fn wait_for_fanout_ready( } } +pub(super) fn fanout_readiness( + projection: Option<&Projection>, + identity: &SinkIdentity, +) -> Result { + let Some(projection) = projection else { + return Ok(false); + }; + 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)); + Ok(projection.graph_ready && sink_is_current) +} + 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 6f29575..5f1af6f 100644 --- a/src/host/observer/adapter.rs +++ b/src/host/observer/adapter.rs @@ -11,7 +11,9 @@ 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::fanout::{ + DesiredLink, LinkMutation, ManagedLinkState, MutationProjectionSink, revalidated_links, +}; use crate::host::taint::snapshot::{ ClientSnapshot, GlobalId, MediaRole, NodeProps, PortDirection, PortSnapshot, Serial, }; @@ -303,12 +305,14 @@ impl LinkMutation for PipeWireLinkMutation { 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)); + // Revoke before creating. Revalidation applies to retained proxies as + // well as new requests: a stale serial-guarded intent is no longer + // authority merely because it already reached `owned`. + let current = revalidated_links(snapshot, desired); + self.owned.retain(|link, _| current.contains(link)); self.states .borrow_mut() - .retain(|link, _| desired.contains(link)); + .retain(|link, _| current.contains(link)); // A Link that reached Error stays failed until the graph changes // enough to remove this exact serial-guarded intent. Retrying the same @@ -323,18 +327,10 @@ impl LinkMutation for PipeWireLinkMutation { self.owned.remove(&link); } - for &link in desired { + for &link in ¤t { 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);