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

feat(host): fan out desktop audio through owned links

Browse Source
This commit is contained in:
mollusk
2026-08-21 16:12:52 -04:00
parent 781defcd84
commit 98cde2c19b
4 changed files with 1171 additions and 84 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/host/audio_plan.rs
+346 -32
View File
@@ -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<Vec<serde_json::Value>> {
fn value_u64(value: &serde_json::Value) -> Option<u64> {
value
.as_u64()
.or_else(|| value.as_str().and_then(|value| value.parse::<u64>().ok()))
}
fn native_links(output_name: &str, sink_name: &str) -> Result<Vec<serde_json::Value>> {
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::<u64>().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<u64> {
object
.pointer("/info/props/object.serial")
.and_then(value_u64)
}
fn object_serial_is_live(serial: u64) -> Result<bool> {
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<u64> {
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<u64> = 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<Child>,
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<u64> = 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"
);
}
}
src/host/fanout.rs
+516 -34
View File
@@ -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<DesiredLink> },
}
/// 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<DesiredLink>,
) -> BTreeMap<DesiredLink, ManagedLinkState>;
}
/// 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<Serial, StreamCaptureState>,
}
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<Serial, StreamCaptureState> {
&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<DesiredLink> = 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<Serial, StreamCaptureState> = 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<ManagedLinkState> = 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<Item = &'a PortSnapshot>,
capture_ports: impl Iterator<Item = &'a PortSnapshot>,
output_node: GlobalId,
output_node_serial: Serial,
input_node: GlobalId,
input_node_serial: Serial,
) -> Result<BTreeSet<DesiredLink>, 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<DesiredLink>,
creates: usize,
drops: usize,
next_state: BTreeMap<DesiredLink, ManagedLinkState>,
}
impl LinkMutation for FakeLinks {
fn reconcile(
&mut self,
snapshot: &GraphSnapshot,
desired: &BTreeSet<DesiredLink>,
) -> BTreeMap<DesiredLink, ManagedLinkState> {
let current: BTreeSet<DesiredLink> = desired
.iter()
.copied()
.filter(|link| link.is_current(snapshot))
.collect();
self.drops += self.held.difference(&current).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<_>>(),
BTreeSet::from([mono])
);
assert_eq!(
links
.iter()
.map(|link| link.input_port)
.map(|link| link.input.port_id)
.collect::<BTreeSet<_>>(),
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<_>>(),
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
}
);
}
}
src/host/graph.rs
+63 -4
View File
@@ -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<RegistryObserverHandle>,
graph_owner: Option<AudioGraphOwner>,
monitor_name: String,
}
@@ -140,11 +152,25 @@ pub(super) struct BareCaptureSink {
impl BareCaptureSink {
pub(super) async fn start(health: health::Reporter) -> Result<Self> {
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<String>,
src/host/observer/adapter.rs
+246 -14
View File
@@ -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<Box<dyn ProjectionSink>>) -> Result<Self> {
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<dyn MutationProjectionSink>,
health: crate::host::health::Reporter,
) -> Result<Self> {
Self::spawn_with_consumer(ObserverConsumer::Mutating { sink, health })
}
fn spawn_with_consumer(consumer: ObserverConsumer) -> Result<Self> {
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<Box<dyn ProjectionSink>>),
Mutating {
sink: Box<dyn MutationProjectionSink>,
health: crate::host::health::Reporter,
},
}
impl ObserverConsumer {
fn mutation_health(&self) -> Option<crate::host::health::Reporter> {
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<BoundProxy>,
}
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<DesiredLink, OwnedFanoutLink>,
states: Rc<RefCell<BTreeMap<DesiredLink, ManagedLinkState>>>,
}
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<OwnedFanoutLink> {
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::<pw::link::Link>("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<DesiredLink>,
) -> BTreeMap<DesiredLink, ManagedLinkState> {
// 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<DesiredLink> = 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<Mutex<Option<Projection>>>,
@@ -134,7 +334,8 @@ struct ObserverState {
/// it are read from `/proc`.
last_candidates: BTreeSet<u32>,
live_globals: BTreeMap<GlobalId, VecDeque<LiveGlobal>>,
sink: Option<Box<dyn ProjectionSink>>,
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<Mutex<Option<Projection>>>,
sink: Option<Box<dyn ProjectionSink>>,
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<Mutex<Option<Projection>>>,
shutdown_rx: pw::channel::Receiver<()>,
sink: Option<Box<dyn ProjectionSink>>,
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(())
}