diff --git a/crates/punktfunk-host/src/audio.rs b/crates/punktfunk-host/src/audio.rs index 21d1ab68..b37a6bb5 100644 --- a/crates/punktfunk-host/src/audio.rs +++ b/crates/punktfunk-host/src/audio.rs @@ -1,5 +1,7 @@ -//! Desktop audio capture for the GameStream audio stream. On Linux: a PipeWire stream that -//! records the default sink's monitor (i.e. everything playing out of the system), delivered +//! Desktop audio capture for the GameStream audio stream. On Linux: a PipeWire stream that by +//! default registers a host-owned **stream sink** claimed as the session's default output +//! (apps play into it directly — immune to hardware-sink churn; `PUNKTFUNK_STREAM_SINK=0` +//! falls back to recording the default sink's monitor). Either way the capture is delivered //! as interleaved `f32` PCM at 48 kHz in the requested channel count (stereo, 5.1 or 7.1 — //! GameStream surround order FL FR FC LFE RL RR [SL SR]). The audio data plane //! (`gamestream::audio`) reframes this into fixed Opus frames, encodes, and sends it. @@ -28,13 +30,24 @@ pub trait AudioCapturer: Send { } /// Discard any buffered chunks (called when a persistent capturer is reused for a new - /// stream, so the client doesn't hear stale audio captured while idle). Default: no-op. + /// stream, so the client doesn't hear stale audio captured while idle). On Linux this is + /// also the session-start hook: the stream-sink capturer re-claims the default sink here + /// (see [`idle`](Self::idle)). Default: no-op. fn drain(&mut self) {} + + /// Called when a session parks the capturer back into the persistent slot: release + /// session-scoped routing side effects while keeping the capture backend alive. On Linux + /// the stream-sink capturer restores the user's default sink here, so host apps play to + /// the real output again between streams; the claim returns with the next + /// [`drain`](Self::drain) (reuse) or a fresh open. Default: no-op. + fn idle(&mut self) {} } -/// Open a live capturer for the default sink monitor (system output) via PipeWire, asking -/// for `channels` interleaved channels. If the sink has fewer channels than requested, -/// PipeWire's channel-mixer fills the missing positions with silence (zero upmix). +/// Open a live capturer for system output via PipeWire, asking for `channels` interleaved +/// channels. Default: a host-owned stream sink claimed as the default output (the sink +/// advertises exactly `channels`, so apps can produce real surround); with +/// `PUNKTFUNK_STREAM_SINK=0`, the default sink's monitor, where a sink with fewer channels +/// gets the missing positions filled with silence (zero upmix). #[cfg(target_os = "linux")] pub fn open_audio_capture(channels: u32) -> Result> { linux::PwAudioCapturer::open(channels).map(|c| Box::new(c) as Box) diff --git a/crates/punktfunk-host/src/audio/linux/mod.rs b/crates/punktfunk-host/src/audio/linux/mod.rs index d535d33a..534cc29a 100644 --- a/crates/punktfunk-host/src/audio/linux/mod.rs +++ b/crates/punktfunk-host/src/audio/linux/mod.rs @@ -1,17 +1,33 @@ -//! PipeWire audio capture of the default sink's monitor (system output). +//! PipeWire desktop-audio capture — via a **host-owned stream sink** (default), or the legacy +//! default-sink-monitor follower (`PUNKTFUNK_STREAM_SINK=0`). //! -//! Connects to the user's PipeWire daemon (via `XDG_RUNTIME_DIR`, inherited from the Sway -//! session) and opens an input stream with `stream.capture.sink=true`, which routes the -//! default sink's monitor into us — no portal needed (unlike screen capture). The (`!Send`) -//! MainLoop/Stream live on a dedicated thread; interleaved `f32` chunks leave over a bounded -//! channel (dropped if the encoder falls behind, never blocking the PipeWire loop). +//! **Stream-sink mode.** The capture stream registers itself as an `Audio/Sink` node +//! ("Punktfunk Stream Speaker", unique `node.name` per capturer): host apps play *into* it, +//! PipeWire mixes them, and our `process()` callback receives the mix directly — the same +//! stream-node architecture as [`PwMicSource`] below (inverted), and the documented +//! `pw-loopback --capture-props='media.class=Audio/Sink'` virtual-sink recipe. A session-scoped +//! [`stream_sink`] claim makes it the *default* sink so apps route to it (and back) with the +//! session. Why: capture no longer depends on any hardware sink, whose availability is display +//! hardware state — live-diagnosed 2026-07-14 on a bazzite/TV host, every gamescope modeset +//! dropped the HDMI audio endpoint, WirePlumber ping-ponged the default HDMI↔auto_null ~8×/s, +//! and the old monitor-follower relinked on every flip (Paused→renegotiate→Streaming storms = +//! client crackle). Bonus: the sink advertises the session's true channel count, so games can +//! produce real 5.1/7.1 even when the local hardware is stereo. //! -//! The stream is opened at the *session's* channel count (2/6/8). If the sink has fewer -//! channels than requested, PipeWire's channel-mixer fills the extra positions with silence -//! (zero upmix), so a stereo desktop still produces a valid 5.1/7.1 capture. Dropping the -//! capturer quits the loop thread (via a `pipewire::channel` Terminate message), tearing the -//! stream down promptly — required so a surround session can replace a stereo capturer -//! without leaking a PipeWire consumer (see CLAUDE.md: a wedged link head-blocks the daemon). +//! **Legacy mode** connects an input stream with `stream.capture.sink=true`, which routes the +//! *default* sink's monitor into us — no portal needed (unlike screen capture), but coupled to +//! hardware-default churn as above. +//! +//! In both modes the (`!Send`) MainLoop/Stream live on a dedicated thread; interleaved `f32` +//! chunks leave over a bounded channel (dropped if the encoder falls behind, never blocking +//! the PipeWire loop). The stream is opened at the *session's* channel count (2/6/8); in +//! legacy mode PipeWire's channel-mixer fills missing positions with silence (zero upmix). +//! Dropping the capturer quits the loop thread (via a `pipewire::channel` Terminate message), +//! tearing the stream — and in stream-sink mode the sink node itself — down promptly, so a +//! surround session can replace a stereo capturer without leaking a PipeWire consumer (see +//! CLAUDE.md: a wedged link head-blocks the daemon). + +mod stream_sink; use super::{AudioCapturer, VirtualMic, SAMPLE_RATE}; use anyhow::{anyhow, Context, Result}; @@ -25,10 +41,26 @@ use std::time::Duration; /// Message asking the PipeWire loop thread to quit (sent from `Drop`). struct Terminate; +/// Whether the host-owned stream sink is active. **Default ON** — decouples capture (and app +/// routing) from hardware-sink availability; see the module docs for the live-diagnosed +/// crackle this fixes. `PUNKTFUNK_STREAM_SINK=0` (also `false`/`no`/`off`) is the escape hatch +/// back to capturing the default sink's monitor. +fn stream_sink_enabled() -> bool { + std::env::var("PUNKTFUNK_STREAM_SINK") + .map(|v| !matches!(v.trim(), "0" | "false" | "no" | "off")) + .unwrap_or(true) +} + pub struct PwAudioCapturer { chunks: Receiver>, channels: u32, quit: pipewire::channel::Sender, + /// `Some(node.name)` in stream-sink mode; `None` = legacy monitor follower. + sink_name: Option, + /// Whether this capturer currently holds a [`stream_sink`] default-sink claim (session + /// active). Toggled by open/[`drain`](AudioCapturer::drain) (claim) and + /// [`idle`](AudioCapturer::idle)/Drop (release). + claimed: bool, } impl PwAudioCapturer { @@ -37,26 +69,64 @@ impl PwAudioCapturer { matches!(channels, 1 | 2 | 6 | 8), "unsupported audio channel count {channels} (want 2, 6 or 8)" ); + // Unique per capturer: overlapping instances (mid-session reopen, concurrent sessions) + // must never alias in metadata claims, and a fresh name gets fresh (unity) WirePlumber + // volume state instead of whatever a previous run left behind. + let sink_name = stream_sink_enabled().then(|| { + use std::sync::atomic::AtomicU64; + static SEQ: AtomicU64 = AtomicU64::new(0); + format!( + "{}-{}-{}", + stream_sink::SINK_NAME_PREFIX, + std::process::id(), + SEQ.fetch_add(1, Ordering::Relaxed) + ) + }); let (tx, rx) = sync_channel::>(64); let (quit_tx, quit_rx) = pipewire::channel::channel::(); + // Bring-up handshake (mirrors the virtual mic): a PipeWire that isn't running must + // surface as an open ERROR — engaging the callers' reopen backoff — and in stream-sink + // mode the sink node must exist before we claim the default to its name. + let (ready_tx, ready_rx) = sync_channel::>(1); + let thread_sink_name = sink_name.clone(); thread::Builder::new() .name("punktfunk-pw-audio".into()) .spawn(move || { - if let Err(e) = pw_thread(tx, quit_rx, channels) { + if let Err(e) = pw_thread(tx, quit_rx, channels, thread_sink_name, ready_tx) { tracing::error!(error = %format!("{e:#}"), "pipewire audio thread failed"); } }) .context("spawn pipewire audio thread")?; + match ready_rx.recv_timeout(Duration::from_secs(5)) { + Ok(Ok(())) => {} + Ok(Err(e)) => return Err(e), + Err(_) => return Err(anyhow!("pipewire audio init timed out")), + } + // The capturer opens at session start, so the routing claim begins here; the paired + // release is `idle()` (parked between sessions) or Drop. + let claimed = match &sink_name { + Some(name) => { + stream_sink::claim(name); + true + } + None => false, + }; Ok(PwAudioCapturer { chunks: rx, channels, quit: quit_tx, + sink_name, + claimed, }) } } impl Drop for PwAudioCapturer { fn drop(&mut self) { + if self.claimed { + self.claimed = false; + stream_sink::release(); + } // Ask the loop thread to quit; the stream/core/loop unwind there (RAII). A failed // send means the thread already exited — nothing to tear down. let _ = self.quit.send(Terminate); @@ -80,6 +150,19 @@ impl AudioCapturer for PwAudioCapturer { fn drain(&mut self) { while self.chunks.try_recv().is_ok() {} + // A parked capturer being reused = a new session starting: re-claim the default sink + // (released by `idle()` when the previous session parked us). + if let (Some(name), false) = (&self.sink_name, self.claimed) { + stream_sink::claim(name); + self.claimed = true; + } + } + + fn idle(&mut self) { + if self.claimed { + self.claimed = false; + stream_sink::release(); + } } } @@ -487,137 +570,205 @@ fn pw_thread( tx: std::sync::mpsc::SyncSender>, quit_rx: pipewire::channel::Receiver, channels: u32, + sink_name: Option, + ready: std::sync::mpsc::SyncSender>, ) -> Result<()> { use pipewire as pw; use pw::{properties::properties, spa}; use spa::param::audio::{AudioFormat, AudioInfoRaw}; use spa::pod::Pod; - pf_capture::pwinit::ensure_init(); - let mainloop = pw::main_loop::MainLoopRc::new(None).context("pw audio MainLoop")?; - let context = pw::context::ContextRc::new(&mainloop, None).context("pw audio Context")?; - let core = context - .connect_rc(None) - .context("pw audio connect (is PipeWire running in this session?)")?; + // Setup errors funnel through the ready handshake (mirrors mic_pw_thread's IIFE). + let result = (|| -> Result<()> { + pf_capture::pwinit::ensure_init(); + let mainloop = pw::main_loop::MainLoopRc::new(None).context("pw audio MainLoop")?; + let context = pw::context::ContextRc::new(&mainloop, None).context("pw audio Context")?; + let core = context + .connect_rc(None) + .context("pw audio connect (is PipeWire running in this session?)")?; - // Cross-thread teardown: the capturer's Drop sends Terminate; quit the loop here. - let _quit_guard = quit_rx.attach(mainloop.loop_(), { - let mainloop = mainloop.clone(); - move |_| mainloop.quit() - }); + // Cross-thread teardown: the capturer's Drop sends Terminate; quit the loop here. + let _quit_guard = quit_rx.attach(mainloop.loop_(), { + let mainloop = mainloop.clone(); + move |_| mainloop.quit() + }); - let stream = pw::stream::StreamBox::new( - &core, - "punktfunk-audio", - properties! { - *pw::keys::MEDIA_TYPE => "Audio", - *pw::keys::MEDIA_CATEGORY => "Capture", - *pw::keys::MEDIA_ROLE => "Music", - // Capture the default sink's monitor (system output), not a microphone. - *pw::keys::STREAM_CAPTURE_SINK => "true", - // Ask for a ~5ms quantum (= one Opus frame) so buffers arrive smoothly rather than - // in large bursts the client's low-latency jitter buffer would hear as glitching. - *pw::keys::NODE_LATENCY => "240/48000", - }, - ) - .context("pw audio Stream")?; + // Death detection (same contract as the virtual mic below): a core error — the daemon + // restarted/went away — ends this thread, so the chunk channel disconnects and + // `next_chunk` returns Err, engaging the sessions' reopen-with-backoff. Without this, a + // PipeWire restart mid-session left a zombie capture thread whose `next_chunk` returned + // quiet-sink empty chunks forever — audio silently dead for the rest of the session. + let _core_listener = core + .add_listener_local() + .error({ + let mainloop = mainloop.clone(); + move |id, _seq, res, message| { + tracing::warn!(id, res, message, "pipewire core error — audio capture ends"); + mainloop.quit(); + } + }) + .register(); - let _listener = stream - .add_local_listener_with_user_data(tx) - .state_changed(|_s, _ud, old, new| { - tracing::debug!(?old, ?new, "pipewire audio stream state"); - }) - .param_changed(|_stream, _tx, id, param| { - let Some(param) = param else { return }; - if id != pw::spa::param::ParamType::Format.as_raw() { - return; - } - let mut info = AudioInfoRaw::default(); - if info.parse(param).is_ok() { - tracing::info!( - format = ?info.format(), - rate = info.rate(), - channels = info.channels(), - "audio format negotiated" - ); - } - }) - .process(|stream, tx| { - let outcome = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| { - let Some(mut buffer) = stream.dequeue_buffer() else { - return; + let props = match &sink_name { + // Stream-sink mode: this stream IS the sink (media.class + Direction::Input). Apps + // play into it, PipeWire mixes them, process() receives the mix. Mirrors the + // validated PwMicSource recipe (stream node + RT_PROCESS; see its property + // comments) — do NOT "modernize" either into a `support.null-audio-sink` adapter + // without re-running that validation. + Some(name) => { + let mut p = properties! { + *pw::keys::MEDIA_TYPE => "Audio", + *pw::keys::MEDIA_CLASS => "Audio/Sink", + *pw::keys::NODE_DESCRIPTION => "Punktfunk Stream Speaker", + *pw::keys::NODE_VIRTUAL => "true", + // Ask for a ~5ms quantum (= one Opus frame) so buffers arrive smoothly + // rather than in bursts the client's jitter buffer would hear as glitching. + *pw::keys::NODE_LATENCY => "240/48000", + // LOW priority — the opposite of the mic's 3000: between sessions the sink + // node stays alive (parked capturer) but must never win WirePlumber's auto + // default election against real hardware; session routing comes from the + // stream_sink claim, not from priority. + "priority.session" => "50", }; - let datas = buffer.datas_mut(); - if datas.is_empty() { - return; - } - let d = &mut datas[0]; - let (offset, size) = { - let c = d.chunk(); - (c.offset() as usize, c.size() as usize) - }; - let Some(buf) = d.data() else { return }; - if offset > buf.len() { - return; - } - let region = &buf[offset..(offset + size).min(buf.len())]; - // Negotiated as F32LE; reinterpret the byte region as interleaved f32. - let n = region.len() / 4; - static FIRST: std::sync::atomic::AtomicBool = - std::sync::atomic::AtomicBool::new(true); - if FIRST.swap(false, std::sync::atomic::Ordering::Relaxed) { - tracing::info!(samples = n, "audio first capture buffer"); - } - let mut samples = Vec::with_capacity(n); - for i in 0..n { - let b = [ - region[i * 4], - region[i * 4 + 1], - region[i * 4 + 2], - region[i * 4 + 3], - ]; - samples.push(f32::from_le_bytes(b)); - } - let _ = tx.try_send(samples); // drop if the encoder is behind - })); - if outcome.is_err() { - tracing::error!("panic in pipewire audio callback — chunk dropped"); + p.insert(*pw::keys::NODE_NAME, name.as_str()); + p } - }) - .register() - .context("register audio stream listener")?; + // Legacy: capture the default sink's monitor (system output), not a microphone. + None => properties! { + *pw::keys::MEDIA_TYPE => "Audio", + *pw::keys::MEDIA_CATEGORY => "Capture", + *pw::keys::MEDIA_ROLE => "Music", + *pw::keys::STREAM_CAPTURE_SINK => "true", + *pw::keys::NODE_LATENCY => "240/48000", + }, + }; + let stream = pw::stream::StreamBox::new(&core, "punktfunk-audio", props) + .context("pw audio Stream")?; - // Request F32LE, 48 kHz, at the session's channel count with explicit positions — - // PipeWire's channel-mixer up/downmixes the sink monitor to this layout. - let mut info = AudioInfoRaw::new(); - info.set_format(AudioFormat::F32LE); - info.set_rate(SAMPLE_RATE); - info.set_channels(channels); - info.set_position(spa_positions(channels)); - let obj = pw::spa::pod::Object { - type_: pw::spa::utils::SpaTypes::ObjectParamFormat.as_raw(), - id: pw::spa::param::ParamType::EnumFormat.as_raw(), - properties: info.into(), - }; - let values: Vec = pw::spa::pod::serialize::PodSerializer::serialize( - std::io::Cursor::new(Vec::new()), - &pw::spa::pod::Value::Object(obj), - ) - .context("serialize audio format pod")? - .0 - .into_inner(); - let mut params = [Pod::from_bytes(&values).context("audio pod from bytes")?]; + let _listener = stream + .add_local_listener_with_user_data(tx) + .state_changed({ + let mainloop = mainloop.clone(); + move |_s, _ud, old, new| { + tracing::debug!(?old, ?new, "pipewire audio stream state"); + // A stream error is unrecoverable for this instance — exit so the sessions' + // reopen path builds a fresh one (same contract as the core-error path above). + if matches!(new, pw::stream::StreamState::Error(_)) { + mainloop.quit(); + } + } + }) + .param_changed(|_stream, _tx, id, param| { + let Some(param) = param else { return }; + if id != pw::spa::param::ParamType::Format.as_raw() { + return; + } + let mut info = AudioInfoRaw::default(); + if info.parse(param).is_ok() { + tracing::info!( + format = ?info.format(), + rate = info.rate(), + channels = info.channels(), + "audio format negotiated" + ); + } + }) + .process(|stream, tx| { + let outcome = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| { + let Some(mut buffer) = stream.dequeue_buffer() else { + return; + }; + let datas = buffer.datas_mut(); + if datas.is_empty() { + return; + } + let d = &mut datas[0]; + let (offset, size) = { + let c = d.chunk(); + (c.offset() as usize, c.size() as usize) + }; + let Some(buf) = d.data() else { return }; + if offset > buf.len() { + return; + } + let region = &buf[offset..(offset + size).min(buf.len())]; + // Negotiated as F32LE; reinterpret the byte region as interleaved f32. + let n = region.len() / 4; + static FIRST: std::sync::atomic::AtomicBool = + std::sync::atomic::AtomicBool::new(true); + if FIRST.swap(false, std::sync::atomic::Ordering::Relaxed) { + tracing::info!(samples = n, "audio first capture buffer"); + } + let mut samples = Vec::with_capacity(n); + for i in 0..n { + let b = [ + region[i * 4], + region[i * 4 + 1], + region[i * 4 + 2], + region[i * 4 + 3], + ]; + samples.push(f32::from_le_bytes(b)); + } + let _ = tx.try_send(samples); // drop if the encoder is behind + })); + if outcome.is_err() { + tracing::error!("panic in pipewire audio callback — chunk dropped"); + } + }) + .register() + .context("register audio stream listener")?; - stream - .connect( - spa::utils::Direction::Input, - None, // PW_ID_ANY — autoconnect to the default sink monitor - pw::stream::StreamFlags::AUTOCONNECT | pw::stream::StreamFlags::MAP_BUFFERS, - &mut params, + // Request F32LE, 48 kHz, at the session's channel count with explicit positions. In + // legacy mode PipeWire's channel-mixer up/downmixes the sink monitor to this layout; + // in stream-sink mode this IS the sink's advertised layout (apps mix/route to it). + let mut info = AudioInfoRaw::new(); + info.set_format(AudioFormat::F32LE); + info.set_rate(SAMPLE_RATE); + info.set_channels(channels); + info.set_position(spa_positions(channels)); + let obj = pw::spa::pod::Object { + type_: pw::spa::utils::SpaTypes::ObjectParamFormat.as_raw(), + id: pw::spa::param::ParamType::EnumFormat.as_raw(), + properties: info.into(), + }; + let values: Vec = pw::spa::pod::serialize::PodSerializer::serialize( + std::io::Cursor::new(Vec::new()), + &pw::spa::pod::Value::Object(obj), ) - .context("pw audio stream connect")?; + .context("serialize audio format pod")? + .0 + .into_inner(); + let mut params = [Pod::from_bytes(&values).context("audio pod from bytes")?]; - mainloop.run(); - tracing::debug!("pipewire audio loop exited (capturer dropped)"); - Ok(()) + // RT_PROCESS in stream-sink mode for the same reason as the mic: the sink must be a + // *synchronous* graph node that joins its producers' driver group and is actually + // driven (see the mic's connect comment — async device-class stream nodes on a busy + // graph never acquire a driver and their process() never fires). + let mut flags = pw::stream::StreamFlags::AUTOCONNECT | pw::stream::StreamFlags::MAP_BUFFERS; + if sink_name.is_some() { + flags |= pw::stream::StreamFlags::RT_PROCESS; + } + stream + .connect( + spa::utils::Direction::Input, // we CONSUME samples (a sink / a monitor tap) + None, // PW_ID_ANY — legacy mode: the default sink monitor + flags, + &mut params, + ) + .context("pw audio stream connect")?; + + // Setup complete: the daemon connection and stream connect succeeded — report ready, + // then block until quit/death. (The connect is async server-side; if the caller's + // default-sink claim lands a few ms before the node registers, WirePlumber simply + // keeps the configured value and elects it the moment the node appears — verified + // live: configured values persist unelected while their target is absent.) + let _ = ready.send(Ok(())); + mainloop.run(); + tracing::debug!("pipewire audio loop exited (capturer dropped)"); + Ok(()) + })(); + if let Err(e) = &result { + let _ = ready.send(Err(anyhow!("{e:#}"))); + } + result } diff --git a/crates/punktfunk-host/src/audio/linux/stream_sink.rs b/crates/punktfunk-host/src/audio/linux/stream_sink.rs new file mode 100644 index 00000000..982dc6d5 --- /dev/null +++ b/crates/punktfunk-host/src/audio/linux/stream_sink.rs @@ -0,0 +1,367 @@ +//! Session-scoped **default-sink claim** for the host-owned stream sink. +//! +//! In stream-sink mode (see the module docs in [`super`]) the capture stream registers itself +//! as an `Audio/Sink` node; for host apps to actually play into it, it must be the *default* +//! sink for the duration of a stream session. WirePlumber elects the default from the +//! `default.configured.audio.sink` metadata key (what `wpctl set-default` writes), and with +//! `linking.follow-default-target` (default true) moves already-running app streams when it +//! changes. So a claim is: save the current configured value, point it at our sink; a release +//! restores it. Live-diagnosed motivation (bazzite host, 2026-07-14): the *hardware* default +//! (HDMI audio on a TV) vanishes on every gamescope modeset, making WirePlumber ping-pong the +//! default HDMI↔auto_null ~8×/s — a capture stream that follows the default relinks on every +//! flip and the client hears crackle. A claimed stream sink is immune: nothing about it +//! depends on display hardware. +//! +//! **Refcounted, latest-wins.** Concurrent sessions (GameStream + punktfunk/1) each hold a +//! claim on their own capturer's sink; the newest claim points routing at *its* sink, and only +//! the release of the last claim restores the pre-claim value — a session ending must never +//! yank the default from under one still running. The ledger lock is held **across** the +//! metadata round-trip so a racing claim/release pair can't interleave their writes (a stale +//! restore overwriting a fresh claim would silence the surviving session). +//! +//! **Crash self-healing.** If the host dies while claimed, the configured default is left +//! pointing at a `punktfunk-speaker-*` node that no longer exists; WirePlumber then falls back +//! to availability-based election (local audio keeps working) and the next claim overwrites +//! the stale value. A stale punktfunk name is never saved as a restore target — restoring it +//! would wedge routing on a ghost sink forever — the restore degrades to *deleting* the key, +//! i.e. handing the choice back to WirePlumber's automatic election. + +use anyhow::{anyhow, Context, Result}; +use std::sync::Mutex; + +/// `node.name` prefix for every host-owned stream sink (full names are uniqued per capturer: +/// `punktfunk-speaker--`, so overlapping capturers — mid-session reopen, concurrent +/// sessions — never alias in `target`/metadata lookups). The staleness rule below matches on +/// this prefix. +pub(super) const SINK_NAME_PREFIX: &str = "punktfunk-speaker"; + +/// The metadata key WirePlumber reads the user's preferred sink from (subject 0 on the +/// `default` metadata object; value is `{"name":""}` typed `Spa:String:JSON`). +const CONFIGURED_SINK_KEY: &str = "default.configured.audio.sink"; + +/// What the last release writes back. +#[derive(Debug, PartialEq)] +enum Restore { + /// Re-set the saved pre-claim value (the raw `{"name":"..."}` JSON). + Value(String), + /// Remove the key: no pre-claim preference existed, or the saved one was a stale + /// punktfunk claim from a crashed host (see the module docs). + Delete, +} + +/// Pure claim bookkeeping — separated from the PipeWire I/O so the refcount/restore rules are +/// unit-testable on every platform. +struct Ledger { + holders: u32, + restore: Option, +} + +impl Ledger { + const fn new() -> Ledger { + Ledger { + holders: 0, + restore: None, + } + } + + /// Count a new claim; `true` means this is the first holder and the caller must save the + /// pre-claim value via [`note_previous`](Self::note_previous). + fn on_claim(&mut self) -> bool { + self.holders += 1; + self.holders == 1 + } + + /// Record what the first claim found, applying the staleness rule. + fn note_previous(&mut self, prev: Option) { + self.restore = Some(match prev { + Some(v) if !v.contains(SINK_NAME_PREFIX) => Restore::Value(v), + _ => Restore::Delete, + }); + } + + /// Count a release; the last holder gets the restore action to apply. + fn on_release(&mut self) -> Option { + self.holders = self.holders.saturating_sub(1); + if self.holders == 0 { + self.restore.take() + } else { + None + } + } +} + +static LEDGER: Mutex = Mutex::new(Ledger::new()); + +/// Point the configured default sink at `sink_name` (refcounted; see the module docs). Never +/// fails the caller: a box where the metadata write doesn't work (WirePlumber absent) still +/// gets a working capture — apps just aren't rerouted, which is exactly the legacy behaviour. +pub(super) fn claim(sink_name: &str) { + let mut ledger = LEDGER.lock().unwrap(); + let first = ledger.on_claim(); + // Latest claim wins: even with an existing holder, route to the newest session's sink. + match set_configured_sink(Some(&format!(r#"{{"name":"{sink_name}"}}"#))) { + Ok(prev) => { + if first { + ledger.note_previous(prev); + } + tracing::info!( + sink = sink_name, + "claimed default sink for the stream session" + ); + } + Err(e) => { + if first { + // Nothing knowable to restore — the release will hand election back to + // WirePlumber (Delete), which is also correct if IT starts working by then. + ledger.note_previous(None); + } + tracing::warn!(error = %format!("{e:#}"), + "could not claim the default sink — host apps may keep playing to the previous output"); + } + } +} + +/// Release one claim; the last release restores the pre-claim configured default. +pub(super) fn release() { + let mut ledger = LEDGER.lock().unwrap(); + let Some(restore) = ledger.on_release() else { + return; + }; + let value = match &restore { + Restore::Value(v) => Some(v.as_str()), + Restore::Delete => None, + }; + match set_configured_sink(value) { + Ok(_) => tracing::info!( + restored = value.unwrap_or(""), + "restored default sink after the stream session" + ), + Err(e) => tracing::warn!(error = %format!("{e:#}"), + "could not restore the default sink — set it manually (wpctl set-default)"), + } +} + +/// One-shot metadata round-trip: connect, find the `default` metadata object, read the current +/// [`CONFIGURED_SINK_KEY`] value, then set it to `value` (`None` deletes the key). Returns the +/// **previous** value. Runs its own short-lived main loop on the calling thread — claims come +/// from session threads at start/end, never from a PipeWire callback. +fn set_configured_sink(value: Option<&str>) -> Result> { + use pipewire as pw; + use std::cell::RefCell; + use std::rc::Rc; + + pf_capture::pwinit::ensure_init(); + let mainloop = pw::main_loop::MainLoopRc::new(None).context("claim MainLoop")?; + let context = pw::context::ContextRc::new(&mainloop, None).context("claim Context")?; + let core = context + .connect_rc(None) + .context("claim connect (is PipeWire running in this session?)")?; + let registry = core.get_registry_rc().context("claim registry")?; + + /// Round-trip phases: 0 = globals replaying, 1 = metadata properties replaying, + /// 2 = mutation flushing. + struct Op { + metadata: Option, + md_listener: Option, + previous: Option, + phase: u8, + expected: Option, + outcome: Option>, + } + let op = Rc::new(RefCell::new(Op { + metadata: None, + md_listener: None, + previous: None, + phase: 0, + expected: None, + outcome: None, + })); + + let _registry_listener = registry + .add_listener_local() + .global({ + let op = op.clone(); + let registry = registry.clone(); + move |global| { + if global.type_ != pw::types::ObjectType::Metadata + || op.borrow().metadata.is_some() + || global.props.as_ref().and_then(|p| p.get("metadata.name")) != Some("default") + { + return; + } + match registry.bind::(global) { + Ok(md) => { + // The server replays existing properties to a fresh bind; capture the + // current configured sink before mutating it. + let listener = md + .add_listener_local() + .property({ + let op = op.clone(); + move |subject, key, _type, value| { + if subject == 0 && key == Some(CONFIGURED_SINK_KEY) { + op.borrow_mut().previous = value.map(str::to_owned); + } + 0 + } + }) + .register(); + let mut o = op.borrow_mut(); + o.metadata = Some(md); + o.md_listener = Some(listener); + } + Err(e) => { + op.borrow_mut().outcome = Some(Err(anyhow!("bind default metadata: {e}"))); + } + } + } + }) + .register(); + + let value_owned = value.map(str::to_owned); + let _core_listener = core + .add_listener_local() + .done({ + let op = op.clone(); + let core = core.clone(); + let mainloop = mainloop.clone(); + move |id, seq| { + if id != pw::core::PW_ID_CORE { + return; + } + let mut o = op.borrow_mut(); + if o.expected != Some(seq) || o.outcome.is_some() { + return; + } + match o.phase { + 0 => { + // All pre-existing globals replayed. No `default` metadata → no + // session manager to negotiate with. + if o.metadata.is_none() { + o.outcome = Some(Err(anyhow!( + "no 'default' metadata object (is WirePlumber running?)" + ))); + mainloop.quit(); + return; + } + o.phase = 1; + o.expected = core.sync(0).ok(); + } + 1 => { + // Property replay complete — `previous` holds the pre-claim value. + let md = o.metadata.as_ref().unwrap(); + md.set_property( + 0, + CONFIGURED_SINK_KEY, + value_owned.as_ref().map(|_| "Spa:String:JSON"), + value_owned.as_deref(), + ); + o.phase = 2; + o.expected = core.sync(0).ok(); + } + _ => { + o.outcome = Some(Ok(())); + mainloop.quit(); + } + } + } + }) + .error({ + let op = op.clone(); + let mainloop = mainloop.clone(); + move |id, _seq, res, message| { + op.borrow_mut().outcome.get_or_insert(Err(anyhow!( + "pipewire core error id={id} res={res}: {message}" + ))); + mainloop.quit(); + } + }) + .register(); + + // A sick-but-connected daemon must not wedge a session start/end (the ledger lock is held + // across this call) — bail out after a bounded wait. + let timer = mainloop.loop_().add_timer({ + let op = op.clone(); + let mainloop = mainloop.clone(); + move |_| { + op.borrow_mut() + .outcome + .get_or_insert(Err(anyhow!("metadata round-trip timed out"))); + mainloop.quit(); + } + }); + let _ = timer.update_timer(Some(std::time::Duration::from_secs(5)), None); + + op.borrow_mut().expected = core.sync(0).ok(); + mainloop.run(); + + let mut o = op.borrow_mut(); + match o.outcome.take() { + Some(Ok(())) => Ok(o.previous.take()), + Some(Err(e)) => Err(e), + None => Err(anyhow!("metadata loop exited unexpectedly")), + } +} + +#[cfg(test)] +mod tests { + use super::*; + + /// First claim saves the pre-claim value; the last release yields it for restore. + #[test] + fn claim_release_roundtrip() { + let mut l = Ledger::new(); + assert!(l.on_claim(), "first claim must save the previous value"); + l.note_previous(Some(r#"{"name":"alsa_output.hdmi"}"#.into())); + assert_eq!( + l.on_release(), + Some(Restore::Value(r#"{"name":"alsa_output.hdmi"}"#.into())) + ); + } + + /// Nested claims (concurrent sessions): only the FIRST saves, only the LAST restores. + #[test] + fn nested_claims_restore_once() { + let mut l = Ledger::new(); + assert!(l.on_claim()); + l.note_previous(Some(r#"{"name":"alsa_output.hdmi"}"#.into())); + assert!( + !l.on_claim(), + "second claim must not overwrite the saved value" + ); + assert_eq!(l.on_release(), None, "inner release must not restore"); + assert_eq!( + l.on_release(), + Some(Restore::Value(r#"{"name":"alsa_output.hdmi"}"#.into())) + ); + } + + /// A stale punktfunk claim left by a crashed host must NEVER become the restore target — + /// it degrades to deleting the key (automatic election). + #[test] + fn stale_own_claim_degrades_to_delete() { + let mut l = Ledger::new(); + assert!(l.on_claim()); + l.note_previous(Some(r#"{"name":"punktfunk-speaker-4242-0"}"#.into())); + assert_eq!(l.on_release(), Some(Restore::Delete)); + } + + /// No pre-claim preference → restore deletes the key. + #[test] + fn unset_previous_deletes() { + let mut l = Ledger::new(); + assert!(l.on_claim()); + l.note_previous(None); + assert_eq!(l.on_release(), Some(Restore::Delete)); + } + + /// Release without claim (defensive) must not underflow or restore. + #[test] + fn unbalanced_release_is_harmless() { + let mut l = Ledger::new(); + assert_eq!(l.on_release(), None); + assert!( + l.on_claim(), + "ledger must stay usable after an unbalanced release" + ); + } +} diff --git a/crates/punktfunk-host/src/gamestream/audio.rs b/crates/punktfunk-host/src/gamestream/audio.rs index 75fe796d..b86555e8 100644 --- a/crates/punktfunk-host/src/gamestream/audio.rs +++ b/crates/punktfunk-host/src/gamestream/audio.rs @@ -297,6 +297,7 @@ fn run( None => audio::open_audio_capture(want).context("open audio capture")?, }; let result = audio_body(&mut *cap, &sock, gcm_key, rikeyid, params, running); + cap.idle(); // parked between sessions — release the routing claim (Linux stream sink) *audio_cap.lock().unwrap() = Some(cap); result } diff --git a/crates/punktfunk-host/src/native/audio.rs b/crates/punktfunk-host/src/native/audio.rs index 5c894599..3e5b253d 100644 --- a/crates/punktfunk-host/src/native/audio.rs +++ b/crates/punktfunk-host/src/native/audio.rs @@ -71,9 +71,9 @@ pub(super) fn audio_thread( // Reuse the cached capturer ONLY when its channel count matches this session's; a stereo // capturer left by a prior session must not feed a 5.1/7.1 session (the encoder + the client's // decoder are sized for `want`, so a mismatched capturer would garble/desync the audio). - let capturer = match audio_cap.lock().unwrap().take() { + let mut capturer = match audio_cap.lock().unwrap().take() { Some(mut c) if c.channels() == want as u32 => { - c.drain(); // discard audio captured between sessions + c.drain(); // discard audio captured between sessions (also re-claims routing) c } prev => { @@ -91,6 +91,7 @@ pub(super) fn audio_thread( Ok(e) => e, Err(e) => { tracing::warn!(error = %e, "opus encoder init failed — session continues without audio"); + capturer.idle(); // parked, not streaming — release the routing claim *audio_cap.lock().unwrap() = Some(capturer); return; } @@ -172,8 +173,10 @@ pub(super) fn audio_thread( } } } - // Return the live capturer for the next session (None if it died and never reopened). - if let Some(c) = capturer { + // Return the live capturer for the next session (None if it died and never reopened), + // releasing its session-scoped routing claim (Linux: the default sink moves back). + if let Some(mut c) = capturer { + c.idle(); *audio_cap.lock().unwrap() = Some(c); } }