fix: clear resolved audio exclusion statuses

This commit is contained in:
2026-08-22 02:41:26 -04:00
parent 360d7112e6
commit ce909afc4b
2 changed files with 282 additions and 1 deletions
+22
View File
@@ -77,6 +77,9 @@ pub enum Event<'a> {
stream_serial: u64, stream_serial: u64,
reason: &'a str, 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 /// The configured AEC identity never appeared before its validation
/// deadline. Fan-out remains fail-closed. /// deadline. Fan-out remains fail-closed.
AecFailed { version: u8, module_index: u64 }, AecFailed { version: u8, module_index: u64 },
@@ -111,6 +114,9 @@ pub(crate) enum AudioExclusionEvent {
stream_serial: u64, stream_serial: u64,
reason: &'static str, reason: &'static str,
}, },
StreamStatusCleared {
stream_serial: u64,
},
AecFailed { AecFailed {
module_index: u64, module_index: u64,
}, },
@@ -133,6 +139,10 @@ impl AudioExclusionEvent {
stream_serial: *stream_serial, stream_serial: *stream_serial,
reason, reason,
}, },
Self::StreamStatusCleared { stream_serial } => Event::StreamStatusCleared {
version: AUDIO_EXCLUSION_EVENT_VERSION,
stream_serial: *stream_serial,
},
Self::AecFailed { module_index } => Event::AecFailed { Self::AecFailed { module_index } => Event::AecFailed {
version: AUDIO_EXCLUSION_EVENT_VERSION, version: AUDIO_EXCLUSION_EVENT_VERSION,
module_index: *module_index, module_index: *module_index,
@@ -209,6 +219,18 @@ mod tests {
r#"{"event":"stream_unsupported","version":1,"stream_serial":4294967297,"reason":"link-creation-failed"}"# 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( let failed = serde_json::to_string(
&AudioExclusionEvent::AecFailed { &AudioExclusionEvent::AecFailed {
module_index: 536_870_919, module_index: 536_870_919,
+260 -1
View File
@@ -107,6 +107,24 @@ impl PlanIssue {
Self::LinkCreationFailed => "link-creation-failed", 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)] #[derive(Clone, Debug, PartialEq, Eq)]
@@ -181,6 +199,11 @@ pub(super) struct FanoutController {
validator: AecValidator, validator: AecValidator,
sticky: StickyState, sticky: StickyState,
states: BTreeMap<Serial, StreamCaptureState>, states: BTreeMap<Serial, StreamCaptureState>,
/// 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<Serial>,
foreign_aec_groups: BTreeSet<String>, foreign_aec_groups: BTreeSet<String>,
status_tx: Option<mpsc::UnboundedSender<AudioExclusionEvent>>, status_tx: Option<mpsc::UnboundedSender<AudioExclusionEvent>>,
#[cfg(test)] #[cfg(test)]
@@ -251,6 +274,7 @@ impl FanoutController {
validator: AecValidator::new(aec, AEC_VALIDATION_TIMEOUT_MILLIS), validator: AecValidator::new(aec, AEC_VALIDATION_TIMEOUT_MILLIS),
sticky: StickyState::default(), sticky: StickyState::default(),
states: BTreeMap::new(), states: BTreeMap::new(),
reported_streams: BTreeSet::new(),
foreign_aec_groups: BTreeSet::new(), foreign_aec_groups: BTreeSet::new(),
status_tx, status_tx,
#[cfg(test)] #[cfg(test)]
@@ -371,6 +395,31 @@ impl MutationProjectionSink for FanoutController {
(serial, state) (serial, state)
}) })
.collect(); .collect();
let mut next_reported_streams: BTreeSet<Serial> = 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(); let mut stream_statuses = Vec::new();
for (&serial, state) in &next_states { for (&serial, state) in &next_states {
if self.states.get(&serial) == Some(state) { if self.states.get(&serial) == Some(state) {
@@ -388,7 +437,7 @@ impl MutationProjectionSink for FanoutController {
reason: reason.code(), reason: reason.code(),
}); });
} }
StreamCaptureState::Unsupported { issue } => { StreamCaptureState::Unsupported { issue } if issue.reports_stream_unsupported() => {
tracing::warn!( tracing::warn!(
serial = serial.0, serial = serial.0,
reason = issue.code(), 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.states = next_states;
self.reported_streams = next_reported_streams;
self.foreign_aec_groups = foreign_aec_groups; self.foreign_aec_groups = foreign_aec_groups;
// `reconcile` above has already dropped every link forbidden by an AEC // `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)); 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] #[test]
fn stereo_ports_pair_by_channel_not_enumeration_order() { fn stereo_ports_pair_by_channel_not_enumeration_order() {
let (graph, app, sink, [app_fl, app_fr, sink_fl, sink_fr]) = stereo_graph(); let (graph, app, sink, [app_fl, app_fr, sink_fl, sink_fr]) = stereo_graph();