mollusk/pixelpass
1
1
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

host/observer: phase 3r adapter — bind every Node and Device

Browse Source 
The I/O half of round 8 (Codex, gpt-5.6-sol xhigh; reviewed, formatted and
extended here). The adapter now reads `object.serial` and nothing else off a
Node or Device global, binds the object, and takes every property the engine
reasons about from its `info` props.

- `BoundProxy` generalises `BoundLink` to Node/Device/Link, each holding its
  listener *before* its proxy so the listener is dropped first — the original
  Link variant had that order inverted.
- Bind attachment now finds its slot by never-recycled serial rather than
  taking the queue's back, so nested callback activity during a bind cannot
  attach one generation's proxy to another's slot on a recycled id. A proxy
  that finds no slot is returned to the caller and dropped after the borrow
  ends. Removal still pops oldest-first, matching the model's `live_ids`.
- An `info` is parsed and emitted on the first callback carrying props and
  thereafter only when `change_mask` contains PROPS. I considered emitting
  unconditionally and leaning on the model's suppression rule, and rejected
  it: if a state-only `info` ever delivered a partial props dict, that would
  overwrite a complete observation with an incomplete one — a worse failure
  than the one it guards against, and the same class as F1.
- Ports stay unbound (v3.5 §6.7 / impl plan §4 item 6).

Gates: exit-gate row 1 (live prop recovery) passes on this host — the tagged
null sink projects `peerspeak.owned`, `pulse.module.id`, `node.passthrough`,
the loopback legs share a `node.link-group`, and a real ALSA node classifies
`session_device`.

Added a second live test for the Device half. Row 1's `session_device`
assertion is satisfied by a *union*: WirePlumber 0.5.15 copies `device.api`
and `alsa.driver_name` onto ALSA nodes here, so it passes through the node
fallback and would keep passing if the Device bind delivered nothing —
leaving §6.7 decision 4 ungated on the development machine. The new test
binds every Device and requires an ALSA card to announce both keys.
Mutation-verified: breaking the Device-side driver read fails the new test
while row 1 still passes, which is the gap as claimed.

195 unit + 3 live green, clippy -D warnings and fmt clean.
This commit is contained in:
mollusk
2026-07-25 18:32:27 -04:00
parent b3d71724ae
commit 306b601490
1 changed files with 409 additions and 39 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/host/observer/adapter.rs
+409 -39
View File
@@ -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<BoundLink>,
serial: Serial,
bound_proxy: Option<BoundProxy>,
}
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<BoundProxy> {
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<BoundLink> {
let (bound_link, empty) = {
fn remove_global(&mut self, id: GlobalId) -> Option<BoundProxy> {
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::<u64>().ok()),
link_group: props.get("node.link-group").map(str::to_string),
client_id: props
.get("client.id")
.and_then(|value| value.parse::<u32>().ok())
.map(GlobalId),
process_id: props
.get("application.process.id")
.and_then(|value| value.parse::<u32>().ok()),
passthrough: truthy(props.get("node.passthrough")),
session_device: false,
},
device_claim: DeviceClaim {
device_id: props
.get("device.id")
.and_then(|value| value.parse::<u32>().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<LinkEndpoints> {
let output_node = props.get("link.output.node")?.parse::<u32>().ok()?;
let input_node = props.get("link.input.node")?.parse::<u32>().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<RefCell<Vec<(pw::device::Device, pw::device::DeviceListener)>>> =
Rc::new(RefCell::new(Vec::new()));
let observed: Rc<RefCell<Vec<DeviceProps>>> = 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::<pw::device::Device, _>(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() {