diff --git a/src/host/audio.rs b/src/host/audio.rs index 675a3f7..00541c8 100644 --- a/src/host/audio.rs +++ b/src/host/audio.rs @@ -92,7 +92,7 @@ impl Routing { // by this actor connection, not a module owned by pipewire-pulse. The // actor also absorbs the per-app router so every sink-owning mode has one // graph lifetime and one readiness handshake. - let (graph_owner, mut event_rx) = AudioGraphOwner::start( + let (graph_owner, mut event_rx, _identity_rx) = AudioGraphOwner::start( opts.app.clone(), CaptureSinkSpec::for_pid(pid), health.clone(), diff --git a/src/host/audio_plan.rs b/src/host/audio_plan.rs index 0977cbe..21da94d 100644 --- a/src/host/audio_plan.rs +++ b/src/host/audio_plan.rs @@ -321,12 +321,16 @@ mod tests { } fn sink_serial(name: &str) -> Result { + sink_global_identity(name).map(|(_, serial)| serial) + } + + fn sink_global_identity(name: &str) -> Result<(u32, 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 + let object = objects .as_array() .context("pw-dump root was not an array")? .iter() @@ -336,8 +340,14 @@ mod tests { .and_then(serde_json::Value::as_str) == Some(name) }) - .and_then(object_serial) - .context("capture sink had no object.serial") + .context("capture sink was absent from pw-dump")?; + let id = object + .get("id") + .and_then(value_u64) + .and_then(|id| u32::try_from(id).ok()) + .context("capture sink had no usable global id")?; + let serial = object_serial(object).context("capture sink had no object.serial")?; + Ok((id, serial)) } /// Subprocess half of the SIGKILL gate. The outer test kills this process, @@ -637,4 +647,146 @@ mod tests { "the owned fan-out link survived its output stream" ); } + + /// Phase-6 row 7: remove the live connection-owned sink from the server, + /// then prove the owner recreates it with a new serial and the fan-out + /// controller drops both stale proxies before returning both replacement + /// channel links to ACTIVE. + #[tokio::test] + #[ignore = "live: destroys and recreates the shared PipeWire capture sink; run alone with --test-threads=1"] + async fn live_capture_sink_replacement_relinks_every_channel() { + let opts = opts(false, false, CaptureMode::DesktopExcluding, false); + let sink_name = crate::repair::plan::sink_name_for(std::process::id()); + let fixture_name = format!("pixelpass_phase6_replacement_{}", std::process::id()); + assert!(!pulse_source_exists(&format!("{sink_name}.monitor"))); + + let (health, _) = health::channel(); + let plan = CapturePlan::start(&opts, health.clone()) + .await + .expect("start desktop-excluding replacement fixture"); + 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 replacement audio fixture"); + + let initial_deadline = Instant::now() + Duration::from_secs(5); + let old_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() < initial_deadline => { + tokio::time::sleep(Duration::from_millis(20)).await; + } + _ => panic!("initial sink never reached two ACTIVE links"), + } + }; + let old_link_serials: Vec = old_links + .iter() + .map(|link| object_serial(link).expect("old link object.serial")) + .collect(); + let (old_global_id, old_sink_serial) = + sink_global_identity(&sink_name).expect("read initial sink identity"); + + let destroy = Command::new("pw-cli") + .args(["destroy", &old_global_id.to_string()]) + .status() + .expect("run pw-cli destroy for the owned sink"); + assert!( + destroy.success(), + "pw-cli refused to destroy the owned sink" + ); + + let replacement_deadline = Instant::now() + Duration::from_secs(5); + let (new_sink_serial, new_links) = loop { + let identity = sink_global_identity(&sink_name); + let links = native_links(&fixture_name, &sink_name); + if let (Ok((_, serial)), Ok(links)) = (identity, links) + && serial != old_sink_serial + && links.len() == 2 + && links.iter().all(|link| { + link.pointer("/info/state") + .and_then(serde_json::Value::as_str) + == Some("active") + }) + { + break (serial, links); + } + if Instant::now() >= replacement_deadline { + panic!( + "replacement sink never returned every channel link to ACTIVE; identity={:?}, links={:?}, health={:?}", + sink_global_identity(&sink_name), + native_links(&fixture_name, &sink_name).map(|links| { + links + .iter() + .map(|link| { + ( + object_serial(link), + link.pointer("/info/state") + .and_then(serde_json::Value::as_str) + .map(str::to_string), + ) + }) + .collect::>() + }), + health.fault(), + ); + } + tokio::time::sleep(Duration::from_millis(20)).await; + }; + assert_ne!(new_sink_serial, old_sink_serial); + let new_link_serials: Vec = new_links + .iter() + .map(|link| object_serial(link).expect("replacement link object.serial")) + .collect(); + assert!( + old_link_serials + .iter() + .all(|serial| !object_serial_is_live(*serial).unwrap_or(true)), + "a stale old-sink link proxy remained live" + ); + assert!( + new_link_serials + .iter() + .all(|serial| !old_link_serials.contains(serial)), + "replacement must bind fresh non-lingering link objects" + ); + + fixture + .kill() + .await + .expect("stop replacement audio fixture"); + plan.shutdown().await; + let residue_deadline = Instant::now() + Duration::from_secs(2); + while pulse_source_exists(&format!("{sink_name}.monitor")) + && Instant::now() < residue_deadline + { + tokio::time::sleep(Duration::from_millis(20)).await; + } + assert!(!pulse_source_exists(&format!("{sink_name}.monitor"))); + assert!(health.fault().is_none()); + } } diff --git a/src/host/fanout.rs b/src/host/fanout.rs index 474f58f..30010ba 100644 --- a/src/host/fanout.rs +++ b/src/host/fanout.rs @@ -133,6 +133,19 @@ pub(super) trait MutationProjectionSink: Send { now_us: u64, links: &mut dyn LinkMutation, ); + + /// Retarget a running controller to the exact serial of a recreated + /// connection-owned capture sink, then reconcile immediately against the + /// observer's current coherent projection. The adapter invokes this only + /// on its PipeWire loop thread, keeping old-link revocation and new-link + /// creation ordered with ordinary registry observations. + fn replace_capture_sink( + &mut self, + capture_sink: Serial, + projection: &Projection, + now_us: u64, + links: &mut dyn LinkMutation, + ); } #[derive(Clone, Debug, PartialEq, Eq)] @@ -339,6 +352,24 @@ impl MutationProjectionSink for FanoutController { self.emit_status(event); } } + + fn replace_capture_sink( + &mut self, + capture_sink: Serial, + projection: &Projection, + now_us: u64, + links: &mut dyn LinkMutation, + ) { + if self.capture_sink == capture_sink { + return; + } + self.capture_sink = capture_sink; + // This is an explicit owner control message rather than a model event, + // so it must not be coalesced behind the next registry callback. The + // normal projection path is still the single place that evaluates, + // plans, revokes stale proxies, and determines all-links-ACTIVE. + self.on_projection(projection, EventKind::Graph, now_us, links); + } } /// Plan every candidate in the decision universe against one capture sink. @@ -712,6 +743,36 @@ mod tests { ); } + #[test] + fn replacing_the_capture_sink_drops_stale_links_and_recaptures() { + let (mut graph, app, old_sink, _) = stereo_graph(); + let new_sink = graph.native_virtual_node("pixelpass_capture_new", MediaRole::Sink, 46); + let new_fl = graph.port_on_channel(new_sink, PortDirection::In, false, Some("FL")); + let new_fr = graph.port_on_channel(new_sink, PortDirection::In, false, Some("FR")); + let projection = projection(&graph); + let mut links = FakeLinks::default(); + let mut controller = FanoutController::new(old_sink, AecConfig::Off); + + controller.on_projection(&projection, EventKind::Graph, 1_000, &mut links); + assert_eq!(links.creates, 2); + assert_eq!(controller.states()[&app], StreamCaptureState::Captured); + + controller.replace_capture_sink(new_sink.serial, &projection, 2_000, &mut links); + + assert_eq!(links.drops, 2, "every old-sink proxy must be dropped"); + assert_eq!(links.creates, 4, "both replacement links must be created"); + assert_eq!(controller.states()[&app], StreamCaptureState::Captured); + assert_eq!( + links + .held + .iter() + .map(|link| link.input.port_id) + .collect::>(), + BTreeSet::from([new_fl, new_fr]), + "only replacement-sink ports may remain held" + ); + } + #[test] fn serial_revalidation_rejects_a_recycled_node_id_before_mutation() { let (mut graph, app, sink, _) = stereo_graph(); diff --git a/src/host/graph.rs b/src/host/graph.rs index fe3fcfe..2c64bd9 100644 --- a/src/host/graph.rs +++ b/src/host/graph.rs @@ -12,7 +12,7 @@ use super::audio::parse_object_serial; use super::fanout::FanoutController; use super::health; use super::observer::Readiness; -use super::observer::adapter::RegistryObserverHandle; +use super::observer::adapter::{FanoutControl, RegistryObserverHandle}; use super::owned_thread::OwnedThread; use super::taint::snapshot::{GlobalId, Serial}; use crate::common::output::AudioExclusionEvent; @@ -123,6 +123,9 @@ enum Cmd { /// the main loop. Dropping the connection removes the native non-lingering /// sink only after those restoration writes have reached the server. Stop, + /// Recreate an unexpectedly removed connection-owned sink on this same + /// PipeWire connection. Only the actor's own callbacks enqueue this. + RecreateSink, } /// Tokio-side owner of the PipeWire actor thread. @@ -143,9 +146,11 @@ pub(super) struct AudioGraphOwner { /// 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), then detach the status task, before - // the sink-owning connection. + // Shutdown order is load-bearing: stop identity forwarding, close the + // fan-out observer connection (and therefore every non-lingering link), + // then drain status before closing the sink-owning connection. + replacement_stop: Option>, + replacement_forwarder: Option>, fanout: Option, status_forwarder: Option>, graph_owner: Option, @@ -155,9 +160,10 @@ pub(super) struct BareCaptureSink { impl BareCaptureSink { pub(super) async fn start(health: health::Reporter) -> Result { let spec = CaptureSinkSpec::for_pid(std::process::id()); - let (graph_owner, _event_rx) = AudioGraphOwner::start(None, spec, health.clone()) - .await - .context("failed to start the bare connection-owned capture sink")?; + let (graph_owner, _event_rx, identity_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 (status_tx, status_rx) = mpsc::unbounded_channel(); let status_forwarder = tokio::spawn(forward_audio_exclusion_status(status_rx)); @@ -167,7 +173,7 @@ impl BareCaptureSink { AecConfig::Off, status_tx, )), - health, + health.clone(), ) .context("failed to start the desktop-excluding fan-out observer") { @@ -184,7 +190,17 @@ impl BareCaptureSink { graph_owner.shutdown().await; return Err(error).context("desktop-excluding fan-out did not become ready"); } + let (replacement_stop, replacement_stop_rx) = oneshot::channel(); + let replacement_forwarder = tokio::spawn(forward_capture_sink_replacements( + identity_rx, + fanout.fanout_control(), + identity.serial, + replacement_stop_rx, + health, + )); Ok(Self { + replacement_stop: Some(replacement_stop), + replacement_forwarder: Some(replacement_forwarder), fanout: Some(fanout), status_forwarder: Some(status_forwarder), graph_owner: Some(graph_owner), @@ -204,6 +220,14 @@ impl BareCaptureSink { } pub(super) async fn shutdown(mut self) { + if let Some(stop) = self.replacement_stop.take() { + let _ = stop.send(()); + } + if let Some(replacement_forwarder) = self.replacement_forwarder.take() + && let Err(error) = replacement_forwarder.await + { + tracing::warn!(%error, "audio fan-out: replacement forwarder task failed"); + } // 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() { @@ -222,6 +246,39 @@ impl BareCaptureSink { } } +async fn forward_capture_sink_replacements( + mut identity_rx: mpsc::UnboundedReceiver, + fanout: FanoutControl, + mut current_serial: u64, + mut stop_rx: oneshot::Receiver<()>, + health: health::Reporter, +) { + loop { + tokio::select! { + biased; + _ = &mut stop_rx => break, + identity = identity_rx.recv() => { + let Some(identity) = identity else { + break; + }; + if identity.serial == current_serial { + continue; + } + current_serial = identity.serial; + tracing::info!( + global_id = identity.global_id, + serial = identity.serial, + "audio fan-out: retargeting to recreated capture sink" + ); + if !fanout.replace_capture_sink(Serial(identity.serial)) { + health.poison("audio fan-out observer rejected capture-sink replacement"); + break; + } + } + } + } +} + async fn forward_audio_exclusion_status( mut status_rx: mpsc::UnboundedReceiver, ) { @@ -263,9 +320,14 @@ impl AudioGraphOwner { filter_name: Option, spec: CaptureSinkSpec, health: health::Reporter, - ) -> Result<(Self, mpsc::UnboundedReceiver)> { + ) -> Result<( + Self, + mpsc::UnboundedReceiver, + mpsc::UnboundedReceiver, + )> { let (cmd_tx, cmd_rx) = pw::channel::channel::(); let (event_tx, event_rx) = mpsc::unbounded_channel::(); + let (identity_tx, identity_rx) = mpsc::unbounded_channel::(); let (ready_tx, ready_rx) = oneshot::channel::(); let phase = Arc::new(AtomicU8::new(GRAPH_STARTING)); let phase_for_thread = Arc::clone(&phase); @@ -273,6 +335,7 @@ impl AudioGraphOwner { let shutdown_for_thread = Arc::clone(&shutdown_observed); let health_for_thread = health.clone(); let spec_for_thread = spec.clone(); + let cmd_tx_for_thread = cmd_tx.clone(); let thread = std::thread::Builder::new() .name("pixelpass-audio-graph".to_string()) @@ -282,7 +345,9 @@ impl AudioGraphOwner { filter_name, spec_for_thread, cmd_rx, + cmd_tx_for_thread, event_tx, + identity_tx, ready_tx, Arc::clone(&phase_for_thread), Arc::clone(&shutdown_for_thread), @@ -350,7 +415,7 @@ impl AudioGraphOwner { monitor = %identity.monitor_name, "audio graph: connection-owned capture sink ready" ); - Ok((owner, event_rx)) + Ok((owner, event_rx, identity_rx)) } pub(super) fn identity(&self) -> &SinkIdentity { @@ -503,7 +568,10 @@ struct ActorState { filter_lower: Option, readiness: NativeReadiness, ready_tx: Option>, + identity_tx: Option>, identity: Option, + replacement_pending: bool, + replacement_prior_serial: Option, epoch: u64, pending: BTreeMap, routed: BTreeMap, @@ -525,6 +593,10 @@ impl ActorState { return; }; self.identity = Some(identity.clone()); + self.replacement_pending = false; + if let Some(tx) = &self.identity_tx { + let _ = tx.send(identity.clone()); + } if let Some(tx) = self.ready_tx.take() { let _ = tx.send(identity); } @@ -573,7 +645,21 @@ impl ActorState { { let _ = tx.send(GraphEvent::LastRoutedStreamGone); } - self.readiness.bound_global_id == Some(id) + self.identity + .as_ref() + .is_some_and(|identity| identity.global_id == id) + || self.readiness.bound_global_id == Some(id) + } + + fn begin_sink_replacement(&mut self) -> bool { + if self.closing || self.replacement_pending { + return false; + } + self.replacement_prior_serial = self.identity.take().map(|identity| identity.serial); + self.readiness.bound_global_id = None; + self.readiness.core_synced = false; + self.replacement_pending = true; + true } fn update_metadata( @@ -669,6 +755,69 @@ fn try_flush(state: &Rc>) { } } +/// Move routes that still point at the removed owned serial directly to the +/// replacement serial. A route changed by another actor during the gap is left +/// alone, preserving the same compare-before-restore ownership rule used at +/// shutdown. Bare Phase-6 capture has no routed records, but keeping the graph +/// actor internally coherent avoids making sink recovery mode-dependent. +fn retarget_routed_after_replacement(state: &Rc>) { + let (metadata, actions, replacement_target) = { + let mut state = state.borrow_mut(); + let Some(prior_serial) = state.replacement_prior_serial else { + return; + }; + let Some(identity) = state.identity.as_ref() else { + return; + }; + let replacement_target = identity.serial.to_string(); + let prior_target = prior_serial.to_string(); + let Some(metadata) = state.default_metadata.clone() else { + if state.routed.is_empty() { + state.replacement_prior_serial = None; + } + return; + }; + let actions: Vec = state + .routed + .values() + .filter(|record| { + observation_is_current(&state.readiness.globals, &record.observed) + && state + .current_targets + .get(&record.observed.global_id) + .and_then(|target| target.value.as_deref()) + == Some(prior_target.as_str()) + }) + .map(|record| record.observed.global_id) + .collect(); + for id in &actions { + state.current_targets.insert( + *id, + MetadataValue { + type_: Some("Spa:Id".to_string()), + value: Some(replacement_target.clone()), + }, + ); + } + state.replacement_prior_serial = None; + (metadata, actions, replacement_target) + }; + + for id in actions { + metadata.metadata.set_property( + id, + "target.object", + Some("Spa:Id"), + Some(&replacement_target), + ); + tracing::info!( + node_id = id, + sink_serial = %replacement_target, + "audio graph: stream retargeted to replacement sink" + ); + } +} + fn restoration_for( record: &RouteRecord, current: Option<&MetadataValue>, @@ -707,12 +856,83 @@ fn prepare_quiesce( (metadata, actions) } +fn request_sink_replacement( + state: &Rc>, + cmd_tx: &pw::channel::Sender, + health: &health::Reporter, + main_loop: &pw::main_loop::MainLoopRc, +) { + if !state.borrow_mut().begin_sink_replacement() { + return; + } + if cmd_tx.send(Cmd::RecreateSink).is_err() { + health.poison("audio graph could not enqueue capture-sink recreation"); + main_loop.quit(); + } +} + +#[allow(clippy::too_many_arguments)] +fn create_owned_sink( + core: &pw::core::CoreRc, + spec: &CaptureSinkSpec, + state: &Rc>, + cmd_tx: &pw::channel::Sender, + health: &health::Reporter, + main_loop: &pw::main_loop::MainLoopRc, + shutdown_observed: &Arc, +) -> Result { + let sink = core + .create_object::("adapter", &spec.properties()) + .context("could not create the connection-owned capture sink")?; + let state_for_bound = Rc::clone(state); + let state_for_proxy_remove = Rc::clone(state); + let cmd_for_proxy_remove = cmd_tx.clone(); + let main_loop_for_proxy_remove = main_loop.clone(); + let health_for_proxy_remove = health.clone(); + let shutdown_for_proxy_remove = Arc::clone(shutdown_observed); + let main_loop_for_proxy_error = main_loop.clone(); + let health_for_proxy_error = health.clone(); + let shutdown_for_proxy_error = Arc::clone(shutdown_observed); + let listener = sink + .upcast_ref() + .add_listener_local() + .bound(move |global_id| { + state_for_bound.borrow_mut().observe_bound(global_id); + }) + .removed(move || { + if !shutdown_for_proxy_remove.load(Ordering::Acquire) { + request_sink_replacement( + &state_for_proxy_remove, + &cmd_for_proxy_remove, + &health_for_proxy_remove, + &main_loop_for_proxy_remove, + ); + } + }) + .error(move |seq, res, message| { + tracing::warn!(seq, result = res, %message, "audio graph: capture-sink proxy error"); + if !shutdown_for_proxy_error.load(Ordering::Acquire) { + health_for_proxy_error.poison(format!( + "connection-owned capture-sink proxy error {res}: {message}" + )); + main_loop_for_proxy_error.quit(); + } + }) + .register(); + Ok(OwnedSink { + _listener: listener, + _proxy: sink, + }) +} + #[allow(clippy::too_many_arguments)] fn run_graph( filter_name: Option, spec: CaptureSinkSpec, cmd_rx: pw::channel::Receiver, + cmd_tx: pw::channel::Sender, event_tx: mpsc::UnboundedSender, + identity_tx: mpsc::UnboundedSender, ready_tx: oneshot::Sender, phase: Arc, shutdown_observed: Arc, @@ -732,7 +952,10 @@ fn run_graph( filter_lower: filter_name.map(|name| name.to_ascii_lowercase()), readiness: NativeReadiness::default(), ready_tx: Some(ready_tx), + identity_tx: Some(identity_tx), identity: None, + replacement_pending: false, + replacement_prior_serial: None, epoch: 0, pending: BTreeMap::new(), routed: BTreeMap::new(), @@ -744,10 +967,15 @@ fn run_graph( stop_sync: None, closing: false, })); + let owned_sink = Rc::new(RefCell::new(None::)); let main_loop_for_cmd = main_loop.clone(); let core_for_cmd = core.clone(); let state_for_cmd = Rc::clone(&state); + let sink_for_cmd = Rc::clone(&owned_sink); + let spec_for_cmd = spec.clone(); + let cmd_tx_for_cmd = cmd_tx.clone(); + let health_for_cmd = health.clone(); let shutdown_for_cmd = Arc::clone(&shutdown_observed); let _cmd_receiver = cmd_rx.attach(main_loop.loop_(), move |cmd| match cmd { Cmd::Quiesce { ack } => { @@ -796,6 +1024,41 @@ fn run_graph( } } } + Cmd::RecreateSink => { + if shutdown_for_cmd.load(Ordering::Acquire) || state_for_cmd.borrow().closing { + return; + } + // The old proxy is already server-removed. Drop its callbacks + // before installing the next generation so no stale listener can + // affect the replacement's identity handshake. + drop(sink_for_cmd.borrow_mut().take()); + match create_owned_sink( + &core_for_cmd, + &spec_for_cmd, + &state_for_cmd, + &cmd_tx_for_cmd, + &health_for_cmd, + &main_loop_for_cmd, + &shutdown_for_cmd, + ) { + Ok(sink) => { + *sink_for_cmd.borrow_mut() = Some(sink); + match core_for_cmd.sync(0) { + Ok(seq) => state_for_cmd.borrow_mut().initial_sync = Some(seq), + Err(error) => { + health_for_cmd.poison(format!( + "capture-sink replacement core.sync failed: {error}" + )); + main_loop_for_cmd.quit(); + } + } + } + Err(error) => { + health_for_cmd.poison(format!("capture-sink recreation failed: {error:#}")); + main_loop_for_cmd.quit(); + } + } + } }); let state_for_done = Rc::clone(&state); @@ -834,6 +1097,7 @@ fn run_graph( // Existing matching streams may have been observed before the // initial round-trip. Readiness becoming true is itself the // event that makes those observations routable. + retarget_routed_after_replacement(&state_for_done); try_flush(&state_for_done); } if stop_completed { @@ -855,6 +1119,7 @@ fn run_graph( let state_for_remove = Rc::clone(&state); let main_loop_for_remove = main_loop.clone(); let health_for_remove = health.clone(); + let cmd_tx_for_remove = cmd_tx.clone(); let shutdown_for_remove = Arc::clone(&shutdown_observed); let _registry_listener = registry .add_listener_local() @@ -927,48 +1192,25 @@ fn run_graph( .global_remove(move |id| { let own_sink_removed = state_for_remove.borrow_mut().remove_global(id); if own_sink_removed && !shutdown_for_remove.load(Ordering::Acquire) { - health_for_remove.poison("connection-owned capture sink disappeared unexpectedly"); - main_loop_for_remove.quit(); + request_sink_replacement( + &state_for_remove, + &cmd_tx_for_remove, + &health_for_remove, + &main_loop_for_remove, + ); } }) .register(); - let sink = core - .create_object::("adapter", &spec.properties()) - .context("could not create the connection-owned capture sink")?; - let state_for_bound = Rc::clone(&state); - let main_loop_for_proxy_remove = main_loop.clone(); - let health_for_proxy_remove = health.clone(); - let shutdown_for_proxy_remove = Arc::clone(&shutdown_observed); - let main_loop_for_proxy_error = main_loop.clone(); - let health_for_proxy_error = health.clone(); - let shutdown_for_proxy_error = Arc::clone(&shutdown_observed); - let listener = sink - .upcast_ref() - .add_listener_local() - .bound(move |global_id| { - state_for_bound.borrow_mut().observe_bound(global_id); - }) - .removed(move || { - if !shutdown_for_proxy_remove.load(Ordering::Acquire) { - health_for_proxy_remove.poison("connection-owned capture-sink proxy was removed"); - main_loop_for_proxy_remove.quit(); - } - }) - .error(move |seq, res, message| { - tracing::warn!(seq, result = res, %message, "audio graph: capture-sink proxy error"); - if !shutdown_for_proxy_error.load(Ordering::Acquire) { - health_for_proxy_error.poison(format!( - "connection-owned capture-sink proxy error {res}: {message}" - )); - main_loop_for_proxy_error.quit(); - } - }) - .register(); - let _owned_sink = OwnedSink { - _listener: listener, - _proxy: sink, - }; + *owned_sink.borrow_mut() = Some(create_owned_sink( + &core, + &spec, + &state, + &cmd_tx, + &health, + &main_loop, + &shutdown_observed, + )?); let initial_sync = core .sync(0) @@ -1508,9 +1750,10 @@ mod tests { "live fixture name is already present" ); let (health, _) = health::channel(); - let (mut owner, _events) = AudioGraphOwner::start(None, spec.clone(), health.clone()) - .await - .expect("connection-owned sink starts"); + let (mut owner, _events, _identities) = + AudioGraphOwner::start(None, spec.clone(), health.clone()) + .await + .expect("connection-owned sink starts"); assert_ne!(owner.identity().global_id, u32::MAX); assert!(owner.identity().serial > 0); assert!(pulse_source_exists(spec.monitor_name()).await); @@ -1532,4 +1775,45 @@ mod tests { ); assert!(health.fault().is_none()); } + + #[tokio::test] + #[ignore = "live: destroys and recreates a real connection-owned PipeWire sink; serialize live audio tests"] + async fn live_removed_sink_is_recreated_with_a_fresh_identity() { + let spec = CaptureSinkSpec::for_pid(std::process::id()); + assert!(!pulse_source_exists(spec.monitor_name()).await); + let (health, _) = health::channel(); + let (owner, _events, mut identities) = + AudioGraphOwner::start(None, spec.clone(), health.clone()) + .await + .expect("connection-owned sink starts"); + let initial = owner.identity().clone(); + let announced_initial = tokio::time::timeout(Duration::from_secs(1), identities.recv()) + .await + .expect("initial identity announcement timed out") + .expect("identity channel closed"); + assert_eq!(announced_initial, initial); + + let destroy = tokio::process::Command::new("pw-cli") + .args(["destroy", &initial.global_id.to_string()]) + .status() + .await + .expect("run pw-cli destroy"); + assert!(destroy.success()); + let replacement = tokio::time::timeout(Duration::from_secs(5), identities.recv()) + .await + .expect("replacement identity announcement timed out") + .expect("identity channel closed during replacement"); + assert_ne!(replacement.serial, initial.serial); + assert_eq!(replacement.name, initial.name); + assert_eq!(replacement.monitor_name, initial.monitor_name); + assert!(pulse_source_exists(spec.monitor_name()).await); + assert!(health.fault().is_none()); + + assert!(owner.shutdown().await); + let deadline = Instant::now() + Duration::from_secs(2); + while pulse_source_exists(spec.monitor_name()).await && Instant::now() < deadline { + tokio::time::sleep(Duration::from_millis(20)).await; + } + assert!(!pulse_source_exists(spec.monitor_name()).await); + } } diff --git a/src/host/observer/adapter.rs b/src/host/observer/adapter.rs index bda8d97..6f29575 100644 --- a/src/host/observer/adapter.rs +++ b/src/host/observer/adapter.rs @@ -29,6 +29,16 @@ use std::time::{Duration, Instant}; const READINESS_TIMEOUT_MILLIS: u64 = 2_000; const TICK_INTERVAL: Duration = Duration::from_millis(250); +fn recoverable_core_error(result: i32) -> bool { + // A global can disappear after registry enumeration but before (or while) + // this observer's bound proxy finishes an operation. PipeWire reports that + // ordinary churn as asynchronous -ENOENT on the Core. The model's removal + // event and serial revalidation still fail closed, so terminating the + // mutation owner here would make sink recreation impossible without + // adding safety. Other Core failures remain fatal. + result == -(nix::errno::Errno::ENOENT as i32) +} + /// A consumer that sees **every** projection, one per applied registry event, /// on the observer thread. /// @@ -53,10 +63,31 @@ pub trait ProjectionSink: Send { /// Tokio-side access to the observer's most recent coherent projection. pub struct RegistryObserverHandle { latest: Arc>>, - shutdown_tx: pw::channel::Sender<()>, + command_tx: pw::channel::Sender, thread: Option>, } +enum ObserverCommand { + ReplaceCaptureSink(Serial), + Shutdown, +} + +/// Cloneable Tokio-side capability for the one mutation-controller command +/// needed during capture-sink recreation. The serial is consumed on the +/// observer's PipeWire thread; callers never receive or reuse a global id. +#[derive(Clone)] +pub(in crate::host) struct FanoutControl { + command_tx: pw::channel::Sender, +} + +impl FanoutControl { + pub(in crate::host) fn replace_capture_sink(&self, capture_sink: Serial) -> bool { + self.command_tx + .send(ObserverCommand::ReplaceCaptureSink(capture_sink)) + .is_ok() + } +} + impl RegistryObserverHandle { /// Spawn the read-only PipeWire registry observer. pub fn spawn() -> Result { @@ -86,11 +117,11 @@ impl RegistryObserverHandle { 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 (command_tx, command_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, consumer) { + if let Err(e) = run_observer(latest_for_thread, command_rx, consumer) { tracing::warn!( "registry observer: libpipewire thread exited with error: {e:#}" ); @@ -103,7 +134,7 @@ impl RegistryObserverHandle { Ok(Self { latest, - shutdown_tx, + command_tx, thread: Some(thread), }) } @@ -116,6 +147,12 @@ impl RegistryObserverHandle { .unwrap_or_else(|poisoned| poisoned.into_inner()) .clone() } + + pub(in crate::host) fn fanout_control(&self) -> FanoutControl { + FanoutControl { + command_tx: self.command_tx.clone(), + } + } } enum ObserverConsumer { @@ -137,7 +174,7 @@ impl ObserverConsumer { impl Drop for RegistryObserverHandle { fn drop(&mut self) { - let _ = self.shutdown_tx.send(()); + let _ = self.command_tx.send(ObserverCommand::Shutdown); if let Some(thread) = self.thread.take() && let Err(e) = thread.join() { @@ -424,6 +461,14 @@ impl ObserverState { .unwrap_or_else(|poisoned| poisoned.into_inner()) = Some(projection); } + fn replace_capture_sink(&mut self, capture_sink: Serial) { + let projection = self.model.project(); + let now_us = u64::try_from(self.started_at.elapsed().as_micros()).unwrap_or(u64::MAX); + if let ObserverConsumer::Mutating { sink, .. } = &mut self.consumer { + sink.replace_capture_sink(capture_sink, &projection, now_us, &mut self.link_mutation); + } + } + /// Record the global's id and apply its add event as one step, so the /// bound-proxy FIFO stays provably lockstep with the model's own `live_ids` /// index. Recording only on *applied* adds (never on unknown object types @@ -489,7 +534,7 @@ impl ObserverState { fn run_observer( latest: Arc>>, - shutdown_rx: pw::channel::Receiver<()>, + command_rx: pw::channel::Receiver, consumer: ObserverConsumer, ) -> Result<()> { let started_at = Instant::now(); @@ -512,10 +557,18 @@ fn run_observer( 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 main_loop_for_command = main_loop.clone(); + let state_for_command = Rc::clone(&state); + let _command_receiver = command_rx.attach(main_loop.loop_(), move |command| match command { + ObserverCommand::ReplaceCaptureSink(capture_sink) => { + state_for_command + .borrow_mut() + .replace_capture_sink(capture_sink); + } + ObserverCommand::Shutdown => { + shutdown_for_receiver.set(true); + main_loop_for_command.quit(); + } }); let pending_sync = Rc::new(Cell::new(None)); @@ -539,7 +592,15 @@ fn run_observer( %message, "registry observer: PipeWire core error" ); - if let Some(health) = &mutation_health_for_error { + if recoverable_core_error(res) { + tracing::info!( + id, + seq, + result = res, + %message, + "registry observer: stale resource disappeared during graph churn" + ); + } else if let Some(health) = &mutation_health_for_error { health.poison(format!( "audio fan-out PipeWire core error {res}: {message}" )); @@ -992,6 +1053,13 @@ mod tests { use super::*; use std::process::Command; + #[test] + fn only_stale_resource_core_errors_are_recoverable() { + assert!(recoverable_core_error(-(nix::errno::Errno::ENOENT as i32))); + assert!(!recoverable_core_error(-(nix::errno::Errno::EPIPE as i32))); + assert!(!recoverable_core_error(0)); + } + /// **R10-4.** The ownership carrier is matched exactly; the lenient /// [`truthy`] spelling is wrong for it. ///