From ce909afc4b64ecd669c6b50d5971421cc3a85102 Mon Sep 17 00:00:00 2001 From: Mollusk Date: Sat, 22 Aug 2026 02:41:26 -0400 Subject: [PATCH] fix: clear resolved audio exclusion statuses --- src/common/output.rs | 22 ++++ src/host/fanout.rs | 261 ++++++++++++++++++++++++++++++++++++++++++- 2 files changed, 282 insertions(+), 1 deletion(-) diff --git a/src/common/output.rs b/src/common/output.rs index 2dee599..5f1a6b2 100644 --- a/src/common/output.rs +++ b/src/common/output.rs @@ -77,6 +77,9 @@ pub enum Event<'a> { stream_serial: u64, reason: &'a str, }, + /// A previously reported per-stream exclusion no longer applies because + /// the stream became capturable or disappeared from the live graph. + StreamStatusCleared { version: u8, stream_serial: u64 }, /// The configured AEC identity never appeared before its validation /// deadline. Fan-out remains fail-closed. AecFailed { version: u8, module_index: u64 }, @@ -111,6 +114,9 @@ pub(crate) enum AudioExclusionEvent { stream_serial: u64, reason: &'static str, }, + StreamStatusCleared { + stream_serial: u64, + }, AecFailed { module_index: u64, }, @@ -133,6 +139,10 @@ impl AudioExclusionEvent { stream_serial: *stream_serial, reason, }, + Self::StreamStatusCleared { stream_serial } => Event::StreamStatusCleared { + version: AUDIO_EXCLUSION_EVENT_VERSION, + stream_serial: *stream_serial, + }, Self::AecFailed { module_index } => Event::AecFailed { version: AUDIO_EXCLUSION_EVENT_VERSION, module_index: *module_index, @@ -209,6 +219,18 @@ mod tests { r#"{"event":"stream_unsupported","version":1,"stream_serial":4294967297,"reason":"link-creation-failed"}"# ); + let cleared = serde_json::to_string( + &AudioExclusionEvent::StreamStatusCleared { + stream_serial: 4_294_967_297, + } + .as_event(), + ) + .unwrap(); + assert_eq!( + cleared, + r#"{"event":"stream_status_cleared","version":1,"stream_serial":4294967297}"# + ); + let failed = serde_json::to_string( &AudioExclusionEvent::AecFailed { module_index: 536_870_919, diff --git a/src/host/fanout.rs b/src/host/fanout.rs index 8cc2100..ff7089b 100644 --- a/src/host/fanout.rs +++ b/src/host/fanout.rs @@ -107,6 +107,24 @@ impl PlanIssue { Self::LinkCreationFailed => "link-creation-failed", } } + + /// Whether this is a stable per-stream limitation worth surfacing to the + /// user. Node and Port globals arrive independently on a live PipeWire + /// graph, so a new playback node normally has no output ports for one or + /// more projections. Capture-sink replacement has the same transient + /// shape on the input side. Those states stay fail-closed, but warning on + /// them would report ordinary graph enumeration as lost audio. + fn reports_stream_unsupported(self) -> bool { + matches!( + self, + Self::UnidentifiedChannel + | Self::AmbiguousCaptureChannel + | Self::DuplicateOutputChannel + | Self::IncompatibleChannelLayout + | Self::ExclusiveCapturePort + | Self::LinkCreationFailed + ) + } } #[derive(Clone, Debug, PartialEq, Eq)] @@ -181,6 +199,11 @@ pub(super) struct FanoutController { validator: AecValidator, sticky: StickyState, states: BTreeMap, + /// Streams for which a user-visible unsupported status is currently live. + /// Unlike `states`, this intentionally survives intermediate `Linking` + /// projections so the eventual all-links-ACTIVE transition can clear the + /// exact warning that was emitted earlier. + reported_streams: BTreeSet, foreign_aec_groups: BTreeSet, status_tx: Option>, #[cfg(test)] @@ -251,6 +274,7 @@ impl FanoutController { validator: AecValidator::new(aec, AEC_VALIDATION_TIMEOUT_MILLIS), sticky: StickyState::default(), states: BTreeMap::new(), + reported_streams: BTreeSet::new(), foreign_aec_groups: BTreeSet::new(), status_tx, #[cfg(test)] @@ -371,6 +395,31 @@ impl MutationProjectionSink for FanoutController { (serial, state) }) .collect(); + let mut next_reported_streams: BTreeSet = next_states + .iter() + .filter_map(|(&serial, state)| match state { + StreamCaptureState::Excluded { reason } if reason.reports_stream_unsupported() => { + Some(serial) + } + StreamCaptureState::Unsupported { issue } if issue.reports_stream_unsupported() => { + Some(serial) + } + _ => None, + }) + .collect(); + // Link proxies become ACTIVE asynchronously, and Node/Port globals can + // temporarily become incoherent again while the graph changes. Keep a + // previously reported warning through both shapes; clear it only once + // the stream is captured, intentionally excluded, or gone. + next_reported_streams.extend(self.reported_streams.iter().copied().filter(|serial| { + match next_states.get(serial) { + Some(StreamCaptureState::Linking) => true, + Some(StreamCaptureState::Unsupported { issue }) => { + !issue.reports_stream_unsupported() + } + _ => false, + } + })); let mut stream_statuses = Vec::new(); for (&serial, state) in &next_states { if self.states.get(&serial) == Some(state) { @@ -388,7 +437,7 @@ impl MutationProjectionSink for FanoutController { reason: reason.code(), }); } - StreamCaptureState::Unsupported { issue } => { + StreamCaptureState::Unsupported { issue } if issue.reports_stream_unsupported() => { tracing::warn!( serial = serial.0, reason = issue.code(), @@ -406,7 +455,21 @@ impl MutationProjectionSink for FanoutController { _ => {} } } + for serial in self + .reported_streams + .difference(&next_reported_streams) + .copied() + { + tracing::info!( + serial = serial.0, + "audio fan-out: previously reported stream status cleared" + ); + stream_statuses.push(AudioExclusionEvent::StreamStatusCleared { + stream_serial: serial.0, + }); + } self.states = next_states; + self.reported_streams = next_reported_streams; self.foreign_aec_groups = foreign_aec_groups; // `reconcile` above has already dropped every link forbidden by an AEC @@ -802,6 +865,202 @@ mod tests { assert_eq!(status_rx.try_recv(), Err(mpsc::error::TryRecvError::Empty)); } + #[test] + fn node_before_output_ports_is_fail_closed_without_a_false_warning() { + // PipeWire announces a new Node and its Ports independently. Model the + // ordinary first projection without output ports, then the coherent + // projection after both channels arrive. Both fixtures create their + // nodes in the same order, so the object identities are stable. + let mut pending = Graph::new(); + let pending_app = pending.app_node("music", MediaRole::StreamOutput, 42); + let pending_sink = + pending.native_virtual_node("pixelpass_capture_test", MediaRole::Sink, 43); + pending.port_on_channel(pending_sink, PortDirection::In, false, Some("FL")); + pending.port_on_channel(pending_sink, PortDirection::In, false, Some("FR")); + + let mut ready = Graph::new(); + let ready_app = ready.app_node("music", MediaRole::StreamOutput, 42); + let ready_sink = ready.native_virtual_node("pixelpass_capture_test", MediaRole::Sink, 43); + ready.port_on_channel(ready_app, PortDirection::Out, false, Some("FL")); + ready.port_on_channel(ready_app, PortDirection::Out, false, Some("FR")); + ready.port_on_channel(ready_sink, PortDirection::In, false, Some("FL")); + ready.port_on_channel(ready_sink, PortDirection::In, false, Some("FR")); + assert_eq!((pending_app, pending_sink), (ready_app, ready_sink)); + + let mut links = FakeLinks::default(); + let (mut controller, mut status_rx) = + controller_with_status(pending_sink.serial, AecConfig::Off); + drive(&mut controller, &pending, &mut links); + assert_eq!( + controller.states()[&pending_app.serial], + StreamCaptureState::Unsupported { + issue: PlanIssue::MissingOutputPorts, + } + ); + assert_eq!(links.creates, 0); + assert_eq!( + status_rx.try_recv(), + Err(mpsc::error::TryRecvError::Empty), + "normal Node-before-Port arrival must not warn the user" + ); + + drive(&mut controller, &ready, &mut links); + assert_eq!( + controller.states()[&ready_app.serial], + StreamCaptureState::Captured + ); + assert_eq!(links.creates, 2); + assert_eq!(status_rx.try_recv(), Err(mpsc::error::TryRecvError::Empty)); + } + + #[test] + fn reported_stream_status_clears_only_after_capture_is_active() { + let mut refused = Graph::new(); + let refused_app = refused.app_node("music", MediaRole::StreamOutput, 42); + let refused_sink = + refused.native_virtual_node("pixelpass_capture_test", MediaRole::Sink, 43); + refused.port_on_channel(refused_app, PortDirection::Out, false, None); + refused.port_on_channel(refused_sink, PortDirection::In, false, Some("FL")); + + let mut ready = Graph::new(); + let ready_app = ready.app_node("music", MediaRole::StreamOutput, 42); + let ready_sink = ready.native_virtual_node("pixelpass_capture_test", MediaRole::Sink, 43); + ready.port_on_channel(ready_app, PortDirection::Out, false, Some("FL")); + ready.port_on_channel(ready_sink, PortDirection::In, false, Some("FL")); + assert_eq!((refused_app, refused_sink), (ready_app, ready_sink)); + + let mut links = FakeLinks::default(); + let (mut controller, mut status_rx) = + controller_with_status(refused_sink.serial, AecConfig::Off); + drive(&mut controller, &refused, &mut links); + assert_eq!( + status_rx.try_recv(), + Ok(AudioExclusionEvent::StreamUnsupported { + stream_serial: refused_app.serial.0, + reason: "unidentified-channel", + }) + ); + + let ready_projection = projection(&ready); + let decisions = evaluate( + &ready_projection.snapshot, + &ExclusionCtx { + pipewire_pulse_pid: Some(PULSE_PID), + graph_ready: true, + ..ExclusionCtx::default() + }, + &StickyState::default(), + ) + .0; + let StreamPlan::Capture { links: planned } = + &plan(&ready_projection.snapshot, &decisions, ready_sink.serial)[&ready_app.serial] + else { + panic!("ready fixture must plan one link"); + }; + let pending_link = *planned.iter().next().expect("mono link"); + links + .next_state + .insert(pending_link, ManagedLinkState::Pending); + drive(&mut controller, &ready, &mut links); + assert_eq!( + controller.states()[&ready_app.serial], + StreamCaptureState::Linking + ); + assert_eq!( + status_rx.try_recv(), + Err(mpsc::error::TryRecvError::Empty), + "a warning must remain live while its replacement link is pending" + ); + + links.next_state.clear(); + drive(&mut controller, &ready, &mut links); + assert_eq!( + controller.states()[&ready_app.serial], + StreamCaptureState::Captured + ); + assert_eq!( + status_rx.try_recv(), + Ok(AudioExclusionEvent::StreamStatusCleared { + stream_serial: ready_app.serial.0, + }) + ); + } + + #[test] + fn reported_stream_status_survives_a_transient_port_gap() { + let mut refused = Graph::new(); + let refused_app = refused.app_node("music", MediaRole::StreamOutput, 42); + let refused_sink = + refused.native_virtual_node("pixelpass_capture_test", MediaRole::Sink, 43); + refused.port_on_channel(refused_app, PortDirection::Out, false, None); + refused.port_on_channel(refused_sink, PortDirection::In, false, Some("FL")); + + let mut pending = Graph::new(); + let pending_app = pending.app_node("music", MediaRole::StreamOutput, 42); + let pending_sink = + pending.native_virtual_node("pixelpass_capture_test", MediaRole::Sink, 43); + pending.port_on_channel(pending_sink, PortDirection::In, false, Some("FL")); + assert_eq!((refused_app, refused_sink), (pending_app, pending_sink)); + + let mut links = FakeLinks::default(); + let (mut controller, mut status_rx) = + controller_with_status(refused_sink.serial, AecConfig::Off); + drive(&mut controller, &refused, &mut links); + assert_eq!( + status_rx.try_recv(), + Ok(AudioExclusionEvent::StreamUnsupported { + stream_serial: refused_app.serial.0, + reason: "unidentified-channel", + }) + ); + + drive(&mut controller, &pending, &mut links); + assert_eq!( + controller.states()[&pending_app.serial], + StreamCaptureState::Unsupported { + issue: PlanIssue::MissingOutputPorts, + } + ); + assert_eq!( + status_rx.try_recv(), + Err(mpsc::error::TryRecvError::Empty), + "a transient Port gap must not falsely resolve an existing warning" + ); + } + + #[test] + fn reported_stream_status_clears_when_the_stream_disappears() { + let mut graph = Graph::new(); + let app = graph.app_node("music", MediaRole::StreamOutput, 42); + let sink = graph.native_virtual_node("pixelpass_capture_test", MediaRole::Sink, 43); + graph.port_on_channel(app, PortDirection::Out, false, None); + graph.port_on_channel(sink, PortDirection::In, false, Some("FL")); + + let mut links = FakeLinks::default(); + let (mut controller, mut status_rx) = controller_with_status(sink.serial, AecConfig::Off); + drive(&mut controller, &graph, &mut links); + assert_eq!( + status_rx.try_recv(), + Ok(AudioExclusionEvent::StreamUnsupported { + stream_serial: app.serial.0, + reason: "unidentified-channel", + }) + ); + + controller.on_projection( + &projection_from(graph.build_without(&[app])), + EventKind::Graph, + 2_000, + &mut links, + ); + assert_eq!( + status_rx.try_recv(), + Ok(AudioExclusionEvent::StreamStatusCleared { + stream_serial: app.serial.0, + }) + ); + } + #[test] fn stereo_ports_pair_by_channel_not_enumeration_order() { let (graph, app, sink, [app_fl, app_fr, sink_fl, sink_fr]) = stereo_graph();