diff --git a/src/host/fanout.rs b/src/host/fanout.rs index 5b8a43c..0ec4da4 100644 --- a/src/host/fanout.rs +++ b/src/host/fanout.rs @@ -332,6 +332,17 @@ impl MutationProjectionSink for FanoutController { continue; } match state { + StreamCaptureState::Excluded { reason } if reason.reports_stream_unsupported() => { + tracing::warn!( + serial = serial.0, + reason = reason.code(), + "audio fan-out: stream refused" + ); + stream_statuses.push(AudioExclusionEvent::StreamUnsupported { + stream_serial: serial.0, + reason: reason.code(), + }); + } StreamCaptureState::Unsupported { issue } => { tracing::warn!( serial = serial.0, @@ -523,7 +534,7 @@ mod tests { use super::*; use crate::host::observer::Readiness; use crate::host::taint::fixture::{Graph, PULSE_PID}; - use crate::host::taint::snapshot::{MediaRole, NodeProps, PortDirection}; + use crate::host::taint::snapshot::{MediaRole, NodeProps, PortDirection, StreamFormat}; use crate::host::taint::{ExclusionCtx, StickyState, evaluate}; fn decisions(graph: &Graph) -> (GraphSnapshot, Decisions) { @@ -637,6 +648,115 @@ mod tests { ) } + fn refusal_graph( + stream_format: StreamFormat, + passthrough: bool, + exclusive: bool, + ) -> (Graph, Serial, Serial) { + let mut graph = Graph::new(); + let client = graph.client_of_app(70); + let stream = graph.node( + "refused", + MediaRole::StreamOutput, + NodeProps { + client_id: Some(client), + process_id: Some(70), + passthrough, + stream_format, + ..NodeProps::default() + }, + ); + let sink = graph.native_virtual_node("pixelpass_capture_test", MediaRole::Sink, 43); + graph.port_on_channel(stream, PortDirection::Out, exclusive, Some("FL")); + graph.port_on_channel(sink, PortDirection::In, false, Some("FL")); + (graph, stream.serial, sink.serial) + } + + fn assert_local_refusal(graph: &Graph, stream: Serial, sink: Serial, expected: Reason) { + let mut links = FakeLinks::default(); + let (mut controller, mut status_rx) = controller_with_status(sink, AecConfig::Off); + drive(&mut controller, graph, &mut links); + + assert_eq!( + links.creates, 0, + "a refused stream must never reach create_link" + ); + assert!(links.held.is_empty()); + assert_eq!( + controller.states()[&stream], + StreamCaptureState::Excluded { reason: expected } + ); + assert_eq!( + status_rx.try_recv(), + Ok(AudioExclusionEvent::StreamUnsupported { + stream_serial: stream.0, + reason: expected.code(), + }) + ); + + drive(&mut controller, graph, &mut links); + assert_eq!(links.creates, 0); + assert_eq!( + status_rx.try_recv(), + Err(mpsc::error::TryRecvError::Empty), + "an unchanged refusal must not repeat the status" + ); + } + + #[test] + fn row_6a_exclusive_output_port_is_refused_with_its_reason() { + let (graph, stream, sink) = refusal_graph(StreamFormat::Raw, false, true); + assert_local_refusal(&graph, stream, sink, Reason::PortExclusive); + } + + #[test] + fn row_6b_encoded_stream_is_refused_with_its_reason() { + let (graph, stream, sink) = refusal_graph(StreamFormat::Encoded, false, false); + assert_local_refusal(&graph, stream, sink, Reason::Encoded); + } + + #[test] + fn row_6c_iec958_stream_is_refused_with_its_reason() { + let (graph, stream, sink) = refusal_graph(StreamFormat::Iec958, false, false); + assert_local_refusal(&graph, stream, sink, Reason::Iec958Passthrough); + } + + #[test] + fn explicit_passthrough_property_is_an_independent_iec958_refusal() { + let (graph, stream, sink) = refusal_graph(StreamFormat::Raw, true, false); + assert_local_refusal(&graph, stream, sink, Reason::Iec958Passthrough); + } + + #[test] + fn pending_format_is_fail_closed_without_a_false_unsupported_status() { + // `refusal_graph` is deterministic, so these snapshots name the same + // serials/ids before and after the Format reply arrives. + let (pending, stream, sink) = refusal_graph(StreamFormat::Unknown, false, false); + let (raw, raw_stream, raw_sink) = refusal_graph(StreamFormat::Raw, false, false); + assert_eq!((stream, sink), (raw_stream, raw_sink)); + + let mut links = FakeLinks::default(); + let (mut controller, mut status_rx) = controller_with_status(sink, AecConfig::Off); + drive(&mut controller, &pending, &mut links); + assert_eq!(links.creates, 0); + assert_eq!( + controller.states()[&stream], + StreamCaptureState::Excluded { + reason: Reason::FormatUnknown, + } + ); + assert_eq!( + status_rx.try_recv(), + Err(mpsc::error::TryRecvError::Empty), + "the normal Node-info-before-Format ordering must not warn the user" + ); + + drive(&mut controller, &raw, &mut links); + assert_eq!(links.creates, 1); + assert_eq!(controller.states()[&stream], StreamCaptureState::Captured); + assert_eq!(status_rx.try_recv(), Err(mpsc::error::TryRecvError::Empty)); + } + #[test] fn stereo_ports_pair_by_channel_not_enumeration_order() { let (graph, app, sink, [app_fl, app_fr, sink_fl, sink_fr]) = stereo_graph(); diff --git a/src/host/observer/adapter.rs b/src/host/observer/adapter.rs index 5f1af6f..60db729 100644 --- a/src/host/observer/adapter.rs +++ b/src/host/observer/adapter.rs @@ -16,6 +16,7 @@ use crate::host::fanout::{ }; use crate::host::taint::snapshot::{ ClientSnapshot, GlobalId, MediaRole, NodeProps, PortDirection, PortSnapshot, Serial, + StreamFormat, }; use crate::host::taint::{PEERSPEAK_OWNED_PROP, PEERSPEAK_OWNED_VALUE}; use anyhow::{Context, Result}; @@ -188,7 +189,7 @@ impl Drop for RegistryObserverHandle { enum BoundProxy { Node { _listener: pw::node::NodeListener, - _proxy: pw::node::Node, + _proxy: Rc, }, Device { _listener: pw::device::DeviceListener, @@ -648,11 +649,15 @@ fn run_observer( return; } }; + let node = Rc::new(node); // 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 format_subscribed = Cell::new(false); + let node_for_info = Rc::downgrade(&node); let state_for_info = Rc::downgrade(&state_for_global); + let state_for_format = Rc::downgrade(&state_for_global); let listener = node .add_listener_local() .info(move |info| { @@ -660,15 +665,65 @@ fn run_observer( return; }; let first = first_info.replace(false); - if !first - && !info.change_mask().contains(pw::node::NodeChangeMask::PROPS) - { + let props_changed = first + || info.change_mask().contains(pw::node::NodeChangeMask::PROPS); + let params_changed = first + || info + .change_mask() + .contains(pw::node::NodeChangeMask::PARAMS); + if !props_changed && !params_changed { return; } - if let Some(state) = state_for_info.upgrade() { + let observation = node_observation_from_props(props); + if props_changed && let Some(state) = state_for_info.upgrade() { state.borrow_mut().apply(RegEvent::NodeInfo { serial, - observation: node_observation_from_props(props), + observation: observation.clone(), + }); + } + + if !observation.role.is_candidate() { + return; + } + let format_readable = info.params().iter().any(|param| { + param.id() == pw::spa::param::ParamType::Format + && param.flags().contains(pw::spa::param::ParamInfoFlags::READ) + }); + if !format_readable { + if params_changed && let Some(state) = state_for_info.upgrade() { + state.borrow_mut().apply(RegEvent::NodeFormat { + serial, + format: StreamFormat::Unknown, + }); + } + return; + } + if !format_subscribed.replace(true) + && let Some(node) = node_for_info.upgrade() + { + // Subscription covers later renegotiation; + // enumeration supplies the current configured + // format. Only candidates advertising a + // readable Format are queried, so ordinary + // driver Nodes cannot turn their expected + // ENOENT/EIO replies into host health faults. + node.subscribe_params(&[pw::spa::param::ParamType::Format]); + node.enum_params( + 0, + Some(pw::spa::param::ParamType::Format), + 0, + u32::MAX, + ); + } + }) + .param(move |_seq, id, _index, _next, param| { + if id != pw::spa::param::ParamType::Format { + return; + } + if let Some(state) = state_for_format.upgrade() { + state.borrow_mut().apply(RegEvent::NodeFormat { + serial, + format: stream_format_from_pod(param), }); } }) @@ -958,6 +1013,63 @@ fn truthy(value: Option<&str>) -> bool { value.is_some_and(|value| value != "false" && value != "0") } +/// Classify the configured SPA Format without depending on producer-specific +/// property aliases. `Unknown` is the safe result for an absent or malformed +/// param; the taint engine refuses that stream until a usable format arrives. +fn stream_format_from_pod(param: Option<&pw::spa::pod::Pod>) -> StreamFormat { + let Some(param) = param else { + return StreamFormat::Unknown; + }; + let Ok((media_type, media_subtype)) = pw::spa::param::format_utils::parse_format(param) else { + tracing::warn!("registry observer: could not parse Node Format media type"); + return StreamFormat::Unknown; + }; + let audio_format = if media_type == pw::spa::param::format::MediaType::Audio + && media_subtype == pw::spa::param::format::MediaSubtype::Raw + { + let mut raw = pw::spa::param::audio::AudioInfoRaw::new(); + match raw.parse(param) { + Ok(_) => Some(raw.format()), + Err(error) => { + tracing::warn!( + ?error, + "registry observer: could not parse raw audio Format" + ); + None + } + } + } else { + None + }; + classify_stream_format(media_type, media_subtype, audio_format) +} + +fn classify_stream_format( + media_type: pw::spa::param::format::MediaType, + media_subtype: pw::spa::param::format::MediaSubtype, + audio_format: Option, +) -> StreamFormat { + use pw::spa::param::audio::AudioFormat; + use pw::spa::param::format::{MediaSubtype, MediaType}; + + if media_type != MediaType::Audio { + return StreamFormat::Unknown; + } + + match media_subtype { + MediaSubtype::Unknown => StreamFormat::Unknown, + MediaSubtype::Iec958 => StreamFormat::Iec958, + MediaSubtype::Raw => match audio_format { + Some(AudioFormat::Encoded) => StreamFormat::Encoded, + Some(AudioFormat::Unknown) | None => StreamFormat::Unknown, + Some(_) => StreamFormat::Raw, + }, + // Any configured non-raw audio subtype is encoded. Keeping this + // conservative also covers codecs newer than this libspa binding. + _ => StreamFormat::Encoded, + } +} + /// The ownership carrier is matched **exactly**, not leniently (round 10, /// R10-4). /// @@ -1004,6 +1116,7 @@ fn node_observation_from_props(props: &pw::spa::utils::dict::DictRef) -> NodeObs .get("application.process.id") .and_then(|value| value.parse::().ok()), passthrough: truthy(props.get("node.passthrough")), + stream_format: StreamFormat::Unknown, device_id, session_device: false, }, @@ -1056,6 +1169,53 @@ mod tests { assert!(!recoverable_core_error(0)); } + #[test] + fn configured_format_classifier_separates_raw_encoded_and_iec958() { + use pw::spa::param::audio::AudioFormat; + use pw::spa::param::format::{MediaSubtype, MediaType}; + + assert_eq!( + classify_stream_format( + MediaType::Audio, + MediaSubtype::Raw, + Some(AudioFormat::F32LE), + ), + StreamFormat::Raw + ); + assert_eq!( + classify_stream_format(MediaType::Audio, MediaSubtype::Mp3, None), + StreamFormat::Encoded + ); + assert_eq!( + classify_stream_format( + MediaType::Audio, + MediaSubtype::Raw, + Some(AudioFormat::Encoded), + ), + StreamFormat::Encoded + ); + assert_eq!( + classify_stream_format(MediaType::Audio, MediaSubtype::Iec958, None), + StreamFormat::Iec958 + ); + assert_eq!( + classify_stream_format(MediaType::Video, MediaSubtype::Raw, None), + StreamFormat::Unknown + ); + assert_eq!( + classify_stream_format( + MediaType::Audio, + MediaSubtype::Raw, + Some(AudioFormat::Unknown), + ), + StreamFormat::Unknown + ); + assert_eq!( + classify_stream_format(MediaType::Audio, MediaSubtype::Unknown, None), + StreamFormat::Unknown + ); + } + /// **R10-4.** The ownership carrier is matched exactly; the lenient /// [`truthy`] spelling is wrong for it. /// diff --git a/src/host/observer/mod.rs b/src/host/observer/mod.rs index 82a70dd..3c0e7af 100644 --- a/src/host/observer/mod.rs +++ b/src/host/observer/mod.rs @@ -88,7 +88,7 @@ mod tests; use crate::host::taint::snapshot::{ ClientSnapshot, GlobalId, GraphSnapshot, LinkSnapshot, MediaRole, NodeProps, NodeSnapshot, - PortSnapshot, Serial, + PortSnapshot, Serial, StreamFormat, }; use classify::{Classification, DeviceClaim, DeviceProps}; use std::collections::{BTreeMap, BTreeSet, VecDeque}; @@ -145,6 +145,13 @@ pub enum RegEvent { serial: Serial, observation: NodeObservation, }, + /// The Node's configured `SPA_PARAM_Format`. This is independent of + /// [`RegEvent::NodeInfo`]: property and parameter callbacks have separate + /// lifetimes, and either may change without the other. + NodeFormat { + serial: Serial, + format: StreamFormat, + }, /// A Port global appeared. PortAdded(PortSnapshot), /// A Client global appeared. Feeds pulse-PID derivation via `sec_pid`. @@ -305,6 +312,10 @@ struct NodeEntry { /// `None` while the bind is outstanding — withheld from the snapshot and /// an outstanding readiness obligation (v3.5 §6.7 decision 3). obs: Option, + /// `Unknown` until the bound Node returns its configured Format param. + /// Unknown streams remain projected but are refused locally, so one idle + /// app cannot hold the entire graph's readiness epoch open. + format: StreamFormat, } /// A live Device: its global id plus its bound properties once they arrive. @@ -427,7 +438,14 @@ impl RegistryModel { match event { RegEvent::NodeAdded { serial, id } => { self.push_id(id, Slot::Node(serial)); - self.nodes.insert(serial, NodeEntry { id, obs: None }); + self.nodes.insert( + serial, + NodeEntry { + id, + obs: None, + format: StreamFormat::Unknown, + }, + ); // A node awaiting its bind is a fresh obligation, so this can // only ever *hold* readiness, never complete it — but the // re-check is cheap and keeps the invariant local. @@ -438,6 +456,7 @@ impl RegistryModel { serial, observation, } => self.on_node_info(serial, observation), + RegEvent::NodeFormat { serial, format } => self.on_node_format(serial, format), RegEvent::PortAdded(port) => { self.push_id(port.id, Slot::Port(port.serial)); self.ports.insert(port.serial, port); @@ -517,6 +536,21 @@ impl RegistryModel { Outcome::Applied } + fn on_node_format(&mut self, serial: Serial, format: StreamFormat) -> Outcome { + let Some(entry) = self.nodes.get_mut(&serial) else { + tracing::debug!( + serial = serial.0, + "observer: node format for an unknown node" + ); + return Outcome::Suppressed; + }; + if entry.format == format { + return Outcome::Suppressed; + } + entry.format = format; + Outcome::Applied + } + fn on_device_info(&mut self, serial: Serial, props: DeviceProps) -> Outcome { let Some(entry) = self.devices.get_mut(&serial) else { tracing::debug!( @@ -737,6 +771,7 @@ impl RegistryModel { }; let mut props = obs.props.clone(); props.session_device = session_device; + props.stream_format = entry.format; Some(NodeSnapshot { serial, id: entry.id, diff --git a/src/host/observer/tests.rs b/src/host/observer/tests.rs index e5c8afe..8798ad9 100644 --- a/src/host/observer/tests.rs +++ b/src/host/observer/tests.rs @@ -15,6 +15,7 @@ use super::pulse_pid; use super::*; use crate::host::taint::snapshot::{ ClientSnapshot, GlobalId, IdLookup, MediaRole, NodeProps, PortDirection, PortSnapshot, Serial, + StreamFormat, }; // ---- builders ------------------------------------------------------------- @@ -657,6 +658,46 @@ fn model_adds_all_four_object_types() { assert_eq!(snap.links().count(), 1); } +#[test] +fn model_projects_configured_node_format_independently_of_props() { + let mut m = model(); + add_stream_out(&mut m, 100, 50); + assert_eq!( + m.project() + .snapshot + .node(ser(100)) + .unwrap() + .props + .stream_format, + StreamFormat::Unknown + ); + + assert_eq!( + m.apply(RegEvent::NodeFormat { + serial: ser(100), + format: StreamFormat::Encoded, + }), + Outcome::Applied + ); + assert_eq!( + m.project() + .snapshot + .node(ser(100)) + .unwrap() + .props + .stream_format, + StreamFormat::Encoded + ); + assert_eq!( + m.apply(RegEvent::NodeFormat { + serial: ser(100), + format: StreamFormat::Encoded, + }), + Outcome::Suppressed, + "an unchanged param must not inflate the graph event stream" + ); +} + #[test] fn model_removes_all_four_object_types() { let mut m = model(); diff --git a/src/host/taint/mod.rs b/src/host/taint/mod.rs index 45dee3c..d88052a 100644 --- a/src/host/taint/mod.rs +++ b/src/host/taint/mod.rs @@ -201,9 +201,15 @@ pub enum Reason { /// A `port.exclusive` port — fan-out will be refused (v3.4 §6.2). Local /// to the node; does not propagate. PortExclusive, - /// An encoded/passthrough stream — a second link would corrupt it. + /// No usable configured Format param has arrived. Fan-out cannot prove a + /// second link is safe. Local to the node; does not propagate. + FormatUnknown, + /// An encoded stream — a second raw-audio link would refuse or corrupt. /// Local to the node; does not propagate. - Passthrough, + Encoded, + /// An IEC958/S/PDIF passthrough stream. Local to the node; does not + /// propagate. + Iec958Passthrough, } impl Reason { @@ -219,10 +225,25 @@ impl Reason { Self::UnresolvedOwner => "unresolved-owner", Self::GraphNotReady => "graph-not-ready", Self::PortExclusive => "port-exclusive", - Self::Passthrough => "passthrough", + Self::FormatUnknown => "format-unknown", + Self::Encoded => "encoded", + Self::Iec958Passthrough => "iec958-passthrough", } } + /// A clean, otherwise-eligible stream whose known format/port shape this + /// fan-out mode cannot link safely. These reasons are user-visible + /// `stream_unsupported` statuses; taint roots and observation gating are + /// intentional exclusions, not failures. `FormatUnknown` is deliberately + /// omitted because the initial Node-info callback can precede the Format + /// reply; reporting that transient would produce a false warning. + pub(super) fn reports_stream_unsupported(self) -> bool { + matches!( + self, + Self::PortExclusive | Self::Encoded | Self::Iec958Passthrough + ) + } + /// Lower wins. A node can acquire taint several ways in one recompute /// and the reported reason must not depend on traversal order, or the /// audit output is unstable and the fixture tests are flaky. Explicit @@ -241,7 +262,9 @@ impl Reason { // because it is only consulted for untainted candidates. Self::GraphNotReady => 8, Self::PortExclusive => 9, - Self::Passthrough => 10, + Self::FormatUnknown => 10, + Self::Encoded => 11, + Self::Iec958Passthrough => 12, } } @@ -1057,13 +1080,18 @@ fn build_decisions( /// clean. These do not propagate — an exclusive-port stream is unlinkable, /// not hazardous. fn local_exclusion(snapshot: &GraphSnapshot, node: &NodeSnapshot) -> Option { - if node.props.passthrough { - return Some(Reason::Passthrough); - } if snapshot.ports_of(node.id).any(|port| port.exclusive) { return Some(Reason::PortExclusive); } - None + if node.props.passthrough { + return Some(Reason::Iec958Passthrough); + } + match node.props.stream_format { + snapshot::StreamFormat::Raw => None, + snapshot::StreamFormat::Unknown => Some(Reason::FormatUnknown), + snapshot::StreamFormat::Encoded => Some(Reason::Encoded), + snapshot::StreamFormat::Iec958 => Some(Reason::Iec958Passthrough), + } } /// Sticky bookkeeping for the next recompute: every tainted owner, with diff --git a/src/host/taint/snapshot.rs b/src/host/taint/snapshot.rs index 8709ba8..a2d6800 100644 --- a/src/host/taint/snapshot.rs +++ b/src/host/taint/snapshot.rs @@ -52,6 +52,25 @@ pub enum MediaRole { Other, } +/// The configured format of an application playback stream. +/// +/// The production observer reads this from the Node's `SPA_PARAM_Format`. +/// Fixtures default to [`Self::Raw`] because almost every graph fixture models +/// ordinary PCM playback; the observer explicitly uses [`Self::Unknown`] +/// until PipeWire supplies a format. Unknown is fail-closed at eligibility. +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] +pub enum StreamFormat { + /// Ordinary PCM audio, safe to fan out subject to the other predicates. + #[default] + Raw, + /// No usable configured format has been observed yet. + Unknown, + /// Encoded audio other than IEC958 passthrough. + Encoded, + /// IEC958/S/PDIF passthrough. + Iec958, +} + impl MediaRole { pub fn parse(media_class: Option<&str>) -> Self { match media_class { @@ -136,9 +155,13 @@ pub struct NodeProps { /// module-created streams this is pipewire-pulse's own PID, which is /// why [`super::ExclusionCtx::pipewire_pulse_pid`] exists. pub process_id: Option, - /// The stream negotiated an encoded/passthrough format; a second link - /// would refuse or corrupt it (v3.4 §6.2). + /// `node.passthrough`; an explicit producer/session-manager refusal flag. + /// Kept separate from [`Self::stream_format`] so the two Phase-6 refusal + /// predicates cannot accidentally mask one another. pub passthrough: bool, + /// The Node's configured `SPA_PARAM_Format`, classified at the observer + /// boundary. Encoded and IEC958 streams are distinct refusal predicates. + pub stream_format: StreamFormat, /// `device.id` — the snapshot-local PipeWire Device this node belongs /// to. This is retained separately from [`Self::session_device`]: the /// latter says the node is a positively-classified passive hardware diff --git a/src/host/taint/tests.rs b/src/host/taint/tests.rs index d7999da..080e605 100644 --- a/src/host/taint/tests.rs +++ b/src/host/taint/tests.rs @@ -599,7 +599,7 @@ fn port_exclusive_and_passthrough_are_local_exclusions() { &[("ok", ok)], &[ ("exclusive", exclusive, "port-exclusive"), - ("passthrough", passthrough, "passthrough"), + ("passthrough", passthrough, "iec958-passthrough"), ], ); // Neither is hazardous — an unlinkable stream must not taint anything.