diff --git a/src/host/observer/adapter.rs b/src/host/observer/adapter.rs index fe02da4..f0e31ee 100644 --- a/src/host/observer/adapter.rs +++ b/src/host/observer/adapter.rs @@ -4,9 +4,14 @@ //! callbacks into [`RegEvent`]s, and publishes the latest [`Projection`] for //! consumers running outside the PipeWire thread. -use super::{EventKind, LinkEndpoints, Outcome, Projection, RegEvent, RegistryModel}; +use super::classify::{DeviceClaim, DeviceProps}; +use super::{ + EventKind, LinkEndpoints, NodeObservation, Outcome, Projection, RegEvent, RegistryModel, +}; use crate::host::audio::parse_object_serial; -use crate::host::taint::snapshot::{ClientSnapshot, GlobalId, PortDirection, PortSnapshot, Serial}; +use crate::host::taint::snapshot::{ + ClientSnapshot, GlobalId, MediaRole, NodeProps, PortDirection, PortSnapshot, Serial, +}; use anyhow::{Context, Result}; use pipewire::{self as pw, types::ObjectType}; use std::cell::{Cell, RefCell}; @@ -101,14 +106,24 @@ impl Drop for RegistryObserverHandle { } } -struct BoundLink { - _proxy: pw::link::Link, - _listener: pw::link::LinkListener, +enum BoundProxy { + Node { + _listener: pw::node::NodeListener, + _proxy: pw::node::Node, + }, + Device { + _listener: pw::device::DeviceListener, + _proxy: pw::device::Device, + }, + Link { + _listener: pw::link::LinkListener, + _proxy: pw::link::Link, + }, } -#[derive(Default)] struct LiveGlobal { - bound_link: Option, + serial: Serial, + bound_proxy: Option, } struct ObserverState { @@ -139,12 +154,13 @@ impl ObserverState { } } - fn apply(&mut self, event: RegEvent) { + fn apply(&mut self, event: RegEvent) -> Outcome { // Taken before the model consumes the event: the sink is told what kind // of observation produced the projection, and deriving that from the // event itself is what stops the two from ever disagreeing. let kind = event.kind(); - let mut outcome = self.model.apply(event); + let event_outcome = self.model.apply(event); + let mut outcome = event_outcome; let candidate = self.model.pulse_pid_candidate(); if candidate != self.last_candidate { @@ -169,6 +185,7 @@ impl ObserverState { if outcome == Outcome::Applied { self.publish(kind); } + event_outcome } fn publish(&mut self, kind: EventKind) { @@ -187,40 +204,65 @@ impl ObserverState { } /// Record the global's id and apply its add event as one step, so the - /// bound-link FIFO stays provably lockstep with the model's own `live_ids` + /// bound-proxy FIFO stays provably lockstep with the model's own `live_ids` /// index. Recording only on *applied* adds (never on unknown object types /// or globals dropped for a missing serial) is what keeps the two id - /// queues the same length per id — otherwise a phantom slot ahead of a - /// bound Link would be popped on removal, leaking that Link's proxy. - fn add(&mut self, id: GlobalId, event: RegEvent) { - self.live_globals - .entry(id) - .or_default() - .push_back(LiveGlobal::default()); - self.apply(event); + /// queues the same length per id — otherwise a phantom slot could pop + /// another generation's proxy after an id is recycled. + fn add(&mut self, serial: Serial, id: GlobalId, event: RegEvent) { + if self.apply(event) == Outcome::Applied { + self.live_globals + .entry(id) + .or_default() + .push_back(LiveGlobal { + serial, + bound_proxy: None, + }); + } } - fn attach_bound_link(&mut self, id: GlobalId, bound_link: BoundLink) { - let Some(global) = self.live_globals.get_mut(&id).and_then(VecDeque::back_mut) else { + /// Return a proxy that could not be attached so its listener is dropped + /// after the caller releases the `RefCell` borrow. + fn attach_bound_proxy( + &mut self, + id: GlobalId, + serial: Serial, + bound_proxy: BoundProxy, + ) -> Option { + let Some(global) = self + .live_globals + .get_mut(&id) + .and_then(|globals| globals.iter_mut().find(|global| global.serial == serial)) + else { tracing::warn!( global_id = id.0, - "registry observer: link bind completed without a live global slot" + serial = serial.0, + "registry observer: bind completed without a live global slot" ); - return; + return Some(bound_proxy); }; - global.bound_link = Some(bound_link); + if global.bound_proxy.is_some() { + tracing::warn!( + global_id = id.0, + serial = serial.0, + "registry observer: live global slot already has a bound proxy" + ); + return Some(bound_proxy); + } + global.bound_proxy = Some(bound_proxy); + None } - fn remove_global(&mut self, id: GlobalId) -> Option { - let (bound_link, empty) = { + fn remove_global(&mut self, id: GlobalId) -> Option { + let (bound_proxy, empty) = { let globals = self.live_globals.get_mut(&id)?; - let bound_link = globals.pop_front().and_then(|global| global.bound_link); - (bound_link, globals.is_empty()) + let bound_proxy = globals.pop_front().and_then(|global| global.bound_proxy); + (bound_proxy, globals.is_empty()) }; if empty { self.live_globals.remove(&id); } - bound_link + bound_proxy } } @@ -291,9 +333,59 @@ fn run_observer( else { return; }; - state_for_global - .borrow_mut() - .add(id, RegEvent::NodeAdded { serial, id }); + state_for_global.borrow_mut().add( + serial, + id, + RegEvent::NodeAdded { serial, id }, + ); + + let Some(registry) = registry_weak.upgrade() else { + return; + }; + let node: pw::node::Node = match registry.bind(obj) { + Ok(node) => node, + Err(e) => { + tracing::warn!( + node_id = obj.id, + "registry observer: failed to bind Node for properties: {e}" + ); + return; + } + }; + // This bit only recognizes the initial callback for the + // change-mask fast path. Admission vs update remains + // entirely the model's decision. + let first_info = Cell::new(true); + let state_for_info = Rc::downgrade(&state_for_global); + let listener = node + .add_listener_local() + .info(move |info| { + let Some(props) = info.props() else { + return; + }; + let first = first_info.replace(false); + if !first + && !info.change_mask().contains(pw::node::NodeChangeMask::PROPS) + { + return; + } + if let Some(state) = state_for_info.upgrade() { + state.borrow_mut().apply(RegEvent::NodeInfo { + serial, + observation: node_observation_from_props(props), + }); + } + }) + .register(); + let unattached = state_for_global.borrow_mut().attach_bound_proxy( + id, + serial, + BoundProxy::Node { + _listener: listener, + _proxy: node, + }, + ); + drop(unattached); } ObjectType::Port => { let Some(props) = obj.props.as_ref() else { @@ -332,6 +424,7 @@ fn run_observer( } }; state_for_global.borrow_mut().add( + serial, id, RegEvent::PortAdded(PortSnapshot { serial, @@ -356,6 +449,7 @@ fn run_observer( return; }; state_for_global.borrow_mut().add( + serial, id, RegEvent::ClientAdded(ClientSnapshot { serial, @@ -381,9 +475,58 @@ fn run_observer( else { return; }; - state_for_global - .borrow_mut() - .add(id, RegEvent::DeviceAdded { serial, id }); + state_for_global.borrow_mut().add( + serial, + id, + RegEvent::DeviceAdded { serial, id }, + ); + + let Some(registry) = registry_weak.upgrade() else { + return; + }; + let device: pw::device::Device = match registry.bind(obj) { + Ok(device) => device, + Err(e) => { + tracing::warn!( + device_id = obj.id, + "registry observer: failed to bind Device for properties: {e}" + ); + return; + } + }; + let first_info = Cell::new(true); + let state_for_info = Rc::downgrade(&state_for_global); + let listener = device + .add_listener_local() + .info(move |info| { + let Some(props) = info.props() else { + return; + }; + let first = first_info.replace(false); + if !first + && !info + .change_mask() + .contains(pw::device::DeviceChangeMask::PROPS) + { + return; + } + if let Some(state) = state_for_info.upgrade() { + state.borrow_mut().apply(RegEvent::DeviceInfo { + serial, + props: device_props_from_props(props), + }); + } + }) + .register(); + let unattached = state_for_global.borrow_mut().attach_bound_proxy( + id, + serial, + BoundProxy::Device { + _listener: listener, + _proxy: device, + }, + ); + drop(unattached); } ObjectType::Link => { let Some(props) = obj.props.as_ref() else { @@ -399,6 +542,7 @@ fn run_observer( }; let endpoints = link_endpoints_from_props(props); state_for_global.borrow_mut().add( + serial, id, RegEvent::LinkAdded { serial, @@ -445,24 +589,26 @@ fn run_observer( } }) .register(); - state_for_global.borrow_mut().attach_bound_link( + let unattached = state_for_global.borrow_mut().attach_bound_proxy( id, - BoundLink { - _proxy: link, + serial, + BoundProxy::Link { _listener: listener, + _proxy: link, }, ); + drop(unattached); } _ => {} } }) .global_remove(move |id| { let id = GlobalId(id); - let bound_link = state_for_remove.borrow_mut().remove_global(id); + let bound_proxy = state_for_remove.borrow_mut().remove_global(id); state_for_remove .borrow_mut() .apply(RegEvent::Removed { id }); - drop(bound_link); + drop(bound_proxy); }) .register(); @@ -506,6 +652,45 @@ fn truthy(value: Option<&str>) -> bool { value.is_some_and(|value| value != "false" && value != "0") } +fn node_observation_from_props(props: &pw::spa::utils::dict::DictRef) -> NodeObservation { + NodeObservation { + name: props.get("node.name").map(str::to_string), + role: MediaRole::parse(props.get("media.class")), + props: NodeProps { + peerspeak_owned: truthy(props.get("peerspeak.owned")), + pulse_module_id: props + .get("pulse.module.id") + .and_then(|value| value.parse::().ok()), + link_group: props.get("node.link-group").map(str::to_string), + client_id: props + .get("client.id") + .and_then(|value| value.parse::().ok()) + .map(GlobalId), + process_id: props + .get("application.process.id") + .and_then(|value| value.parse::().ok()), + passthrough: truthy(props.get("node.passthrough")), + session_device: false, + }, + device_claim: DeviceClaim { + device_id: props + .get("device.id") + .and_then(|value| value.parse::().ok()) + .map(GlobalId), + device_api: props.get("device.api").map(str::to_string), + factory_name: props.get("factory.name").map(str::to_string), + alsa_driver_name: props.get("alsa.driver_name").map(str::to_string), + }, + } +} + +fn device_props_from_props(props: &pw::spa::utils::dict::DictRef) -> DeviceProps { + DeviceProps { + device_api: props.get("device.api").map(str::to_string), + alsa_driver_name: props.get("alsa.driver_name").map(str::to_string), + } +} + fn link_endpoints_from_props(props: &pw::spa::utils::dict::DictRef) -> Option { let output_node = props.get("link.output.node")?.parse::().ok()?; let input_node = props.get("link.input.node")?.parse::().ok()?; @@ -606,6 +791,191 @@ mod tests { .any(|node| node.name.as_deref() == Some(name)) } + /// Phase 3r exit-gate row 1, the Device half — and the reason it needs its + /// own test. + /// + /// `live_bound_properties_recover_node_and_device_inputs` asserts + /// `session_device`, which the classifier grants on a **union**: + /// `device.api` and `alsa.driver_name` may come from the bound Device *or* + /// from the node's own copies. On this host (WirePlumber 0.5.15 ≥ 0.5.13) + /// the session manager *does* copy both onto ALSA nodes, so that assertion + /// passes through the node fallback and would keep passing if the Device + /// bind delivered nothing at all — leaving v3.5 §6.7 decision 4, the whole + /// authoritative path, ungated on the machine we develop on. + /// + /// So assert the Device side directly: bind every Device global and require + /// that at least one ALSA card announces **both** keys on its `info` props. + /// A failure here means the fix for the phase-3 review's owed finding (a + /// real card over-excluded on installs that do not copy `alsa.*` onto the + /// node) rests on nothing. + #[test] + #[ignore = "needs live pipewire"] + fn live_device_bind_carries_api_and_driver_name() { + pw::init(); + let main_loop = pw::main_loop::MainLoopRc::new(None).expect("pw main loop"); + let context = pw::context::ContextRc::new(&main_loop, None).expect("pw context"); + let core = context.connect_rc(None).expect("pw core connect"); + let registry = core.get_registry_rc().expect("pw registry"); + + // Devices bound off the registry, each holding its proxy + listener so + // the callback lives long enough to fire, exactly as the adapter does. + let bound: Rc>> = + Rc::new(RefCell::new(Vec::new())); + let observed: Rc>> = Rc::new(RefCell::new(Vec::new())); + + let bound_for_global = Rc::clone(&bound); + let observed_for_global = Rc::clone(&observed); + let registry_weak = registry.downgrade(); + let _listener = registry + .add_listener_local() + .global(move |obj| { + if obj.type_ != ObjectType::Device { + return; + } + let Some(registry) = registry_weak.upgrade() else { + return; + }; + let Ok(device) = registry.bind::(obj) else { + return; + }; + let observed_for_info = Rc::clone(&observed_for_global); + let listener = device + .add_listener_local() + .info(move |info| { + if let Some(props) = info.props() { + observed_for_info + .borrow_mut() + .push(device_props_from_props(props)); + } + }) + .register(); + bound_for_global.borrow_mut().push((device, listener)); + }) + .register(); + + // Two seconds is the same budget the observer gives its own binds. + let main_loop_for_timer = main_loop.clone(); + let timer = main_loop + .loop_() + .add_timer(move |_| main_loop_for_timer.quit()); + timer + .update_timer(Some(Duration::from_secs(2)), None) + .into_result() + .expect("arm the test deadline"); + main_loop.run(); + + let observed = observed.borrow(); + assert!( + !observed.is_empty(), + "no Device delivered info props at all — the Device bind path is dead" + ); + assert!( + observed.iter().any(|props| { + props.device_api.as_deref() == Some("alsa") && props.alsa_driver_name.is_some() + }), + "no bound Device carried both device.api=alsa and alsa.driver_name; \ + observed: {observed:?}" + ); + } + + // Phase 3r exit-gate row 1: failure means the observation boundary regressed. + #[test] + #[ignore = "needs live pipewire"] + fn live_bound_properties_recover_node_and_device_inputs() { + pw::init(); + let observer = RegistryObserverHandle::spawn().expect("observer thread must spawn"); + wait_for(&observer, |projection| projection.graph_ready); + + let unique = format!("pixelpass_observer_props_test_{}", std::process::id()); + let capture_name = format!("{unique}_capture"); + let playback_name = format!("{unique}_playback"); + let null_sink = PactlModule::load( + "module-null-sink", + &[ + format!("sink_name={unique}"), + "sink_properties=peerspeak.owned=true node.passthrough=true".to_string(), + ], + ); + let null_sink_id = null_sink.id.expect("null-sink module must have an id"); + let loopback = PactlModule::load( + "module-loopback", + &[ + format!("source={unique}.monitor"), + format!("sink={unique}"), + format!("source_output_properties=node.name={capture_name}"), + format!("sink_input_properties=node.name={playback_name}"), + ], + ); + + let projection = wait_for(&observer, |projection| { + projection.graph_ready + && has_node(projection, &unique) + && has_node(projection, &capture_name) + && has_node(projection, &playback_name) + }); + let tagged_sink = projection + .snapshot + .nodes() + .find(|node| node.name.as_deref() == Some(&unique)) + .expect("tagged null sink must be projected"); + assert!(tagged_sink.props.peerspeak_owned); + assert!(tagged_sink.props.passthrough); + assert_eq!( + tagged_sink.props.pulse_module_id, + Some(u64::from(null_sink_id)) + ); + + let capture = projection + .snapshot + .nodes() + .find(|node| node.name.as_deref() == Some(&capture_name)) + .expect("loopback capture leg must be projected"); + let playback = projection + .snapshot + .nodes() + .find(|node| node.name.as_deref() == Some(&playback_name)) + .expect("loopback playback leg must be projected"); + let capture_group = capture + .props + .link_group + .as_ref() + .expect("loopback capture leg must carry node.link-group"); + let playback_group = playback + .props + .link_group + .as_ref() + .expect("loopback playback leg must carry node.link-group"); + assert_eq!(capture_group, playback_group); + assert!( + projection + .snapshot + .nodes() + .any(|node| node.props.process_id.is_some()), + "at least one projected node must carry application.process.id" + ); + let session_device = projection.snapshot.nodes().find(|node| { + node.props.session_device + && (node + .name + .as_deref() + .is_some_and(|name| name.contains("alsa")) + || matches!(node.role, MediaRole::Sink | MediaRole::Source)) + }); + assert!( + session_device.is_some(), + "a named ALSA or Audio/Sink/Audio/Source node must classify as a session device" + ); + assert!(projection.graph_ready); + + loopback.unload(); + null_sink.unload(); + wait_for(&observer, |projection| { + !has_node(projection, &unique) + && !has_node(projection, &capture_name) + && !has_node(projection, &playback_name) + }); + } + #[test] #[ignore = "needs live pipewire"] fn live_topology_diff_tracks_null_sink_and_loopback() {