feat(audio/linux): host-owned stream sink decouples capture from hardware-sink churn

The Linux desktop-audio capture stream now registers itself as an Audio/Sink
node ("Punktfunk Stream Speaker", default-on, PUNKTFUNK_STREAM_SINK=0 =
legacy escape hatch) and claims the configured default sink for the duration
of a session (saved and restored around it, refcounted across concurrent
sessions, crash-stale claims degrade to automatic election). Host apps play
directly into the capture stream, so the capture link no longer depends on
any hardware sink.

Root cause this fixes (live-diagnosed on a bazzite/LG-TV host): gamescope
modesets drop the NVIDIA HDMI audio endpoint, WirePlumber ping-pongs the
default sink HDMI<->auto_null ~8x/s, and the old monitor-follower relinked
its capture on every flip - Paused/renegotiate/Streaming storms (~1300 log
lines/min) heard as crackle on the client.

Also fixes a latent liveness bug in both modes: the capture thread had no
core-error listener, so a PipeWire daemon restart mid-session left a zombie
thread returning quiet-sink empty chunks forever (same class as the historic
virtual-mic death bug). Now the thread exits and sessions reopen with backoff.

Bonus: the sink advertises the session's true channel count, so games can
produce real 5.1/7.1 even when local hardware is stereo. New AudioCapturer::
idle() hook releases the routing claim when a capturer is parked between
sessions; drain() re-claims on reuse.

Verified: cargo clippy -D warnings + 295 tests green on Linux (.21).
On-glass validation on the bazzite host pending.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
2026-07-14 09:34:36 +02:00
parent 57d89217fb
commit 44b71e7460
5 changed files with 676 additions and 141 deletions
+19 -6
View File
@@ -1,5 +1,7 @@
//! Desktop audio capture for the GameStream audio stream. On Linux: a PipeWire stream that //! Desktop audio capture for the GameStream audio stream. On Linux: a PipeWire stream that by
//! records the default sink's monitor (i.e. everything playing out of the system), delivered //! 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 — //! 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 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. //! (`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 /// 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) {} 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 /// Open a live capturer for system output via PipeWire, asking for `channels` interleaved
/// for `channels` interleaved channels. If the sink has fewer channels than requested, /// channels. Default: a host-owned stream sink claimed as the default output (the sink
/// PipeWire's channel-mixer fills the missing positions with silence (zero upmix). /// 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")] #[cfg(target_os = "linux")]
pub fn open_audio_capture(channels: u32) -> Result<Box<dyn AudioCapturer>> { pub fn open_audio_capture(channels: u32) -> Result<Box<dyn AudioCapturer>> {
linux::PwAudioCapturer::open(channels).map(|c| Box::new(c) as Box<dyn AudioCapturer>) linux::PwAudioCapturer::open(channels).map(|c| Box::new(c) as Box<dyn AudioCapturer>)
+282 -131
View File
@@ -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 //! **Stream-sink mode.** The capture stream registers itself as an `Audio/Sink` node
//! session) and opens an input stream with `stream.capture.sink=true`, which routes the //! ("Punktfunk Stream Speaker", unique `node.name` per capturer): host apps play *into* it,
//! default sink's monitor into us — no portal needed (unlike screen capture). The (`!Send`) //! PipeWire mixes them, and our `process()` callback receives the mix directly — the same
//! MainLoop/Stream live on a dedicated thread; interleaved `f32` chunks leave over a bounded //! stream-node architecture as [`PwMicSource`] below (inverted), and the documented
//! channel (dropped if the encoder falls behind, never blocking the PipeWire loop). //! `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 //! **Legacy mode** connects an input stream with `stream.capture.sink=true`, which routes the
//! channels than requested, PipeWire's channel-mixer fills the extra positions with silence //! *default* sink's monitor into us — no portal needed (unlike screen capture), but coupled to
//! (zero upmix), so a stereo desktop still produces a valid 5.1/7.1 capture. Dropping the //! hardware-default churn as above.
//! 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 //! In both modes the (`!Send`) MainLoop/Stream live on a dedicated thread; interleaved `f32`
//! without leaking a PipeWire consumer (see CLAUDE.md: a wedged link head-blocks the daemon). //! 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 super::{AudioCapturer, VirtualMic, SAMPLE_RATE};
use anyhow::{anyhow, Context, Result}; use anyhow::{anyhow, Context, Result};
@@ -25,10 +41,26 @@ use std::time::Duration;
/// Message asking the PipeWire loop thread to quit (sent from `Drop`). /// Message asking the PipeWire loop thread to quit (sent from `Drop`).
struct Terminate; 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 { pub struct PwAudioCapturer {
chunks: Receiver<Vec<f32>>, chunks: Receiver<Vec<f32>>,
channels: u32, channels: u32,
quit: pipewire::channel::Sender<Terminate>, quit: pipewire::channel::Sender<Terminate>,
/// `Some(node.name)` in stream-sink mode; `None` = legacy monitor follower.
sink_name: Option<String>,
/// 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 { impl PwAudioCapturer {
@@ -37,26 +69,64 @@ impl PwAudioCapturer {
matches!(channels, 1 | 2 | 6 | 8), matches!(channels, 1 | 2 | 6 | 8),
"unsupported audio channel count {channels} (want 2, 6 or 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::<Vec<f32>>(64); let (tx, rx) = sync_channel::<Vec<f32>>(64);
let (quit_tx, quit_rx) = pipewire::channel::channel::<Terminate>(); let (quit_tx, quit_rx) = pipewire::channel::channel::<Terminate>();
// 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::<Result<()>>(1);
let thread_sink_name = sink_name.clone();
thread::Builder::new() thread::Builder::new()
.name("punktfunk-pw-audio".into()) .name("punktfunk-pw-audio".into())
.spawn(move || { .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"); tracing::error!(error = %format!("{e:#}"), "pipewire audio thread failed");
} }
}) })
.context("spawn pipewire audio thread")?; .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 { Ok(PwAudioCapturer {
chunks: rx, chunks: rx,
channels, channels,
quit: quit_tx, quit: quit_tx,
sink_name,
claimed,
}) })
} }
} }
impl Drop for PwAudioCapturer { impl Drop for PwAudioCapturer {
fn drop(&mut self) { 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 // 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. // send means the thread already exited — nothing to tear down.
let _ = self.quit.send(Terminate); let _ = self.quit.send(Terminate);
@@ -80,6 +150,19 @@ impl AudioCapturer for PwAudioCapturer {
fn drain(&mut self) { fn drain(&mut self) {
while self.chunks.try_recv().is_ok() {} 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();
}
} }
} }
@@ -484,137 +567,205 @@ fn pw_thread(
tx: std::sync::mpsc::SyncSender<Vec<f32>>, tx: std::sync::mpsc::SyncSender<Vec<f32>>,
quit_rx: pipewire::channel::Receiver<Terminate>, quit_rx: pipewire::channel::Receiver<Terminate>,
channels: u32, channels: u32,
sink_name: Option<String>,
ready: std::sync::mpsc::SyncSender<Result<()>>,
) -> Result<()> { ) -> Result<()> {
use pipewire as pw; use pipewire as pw;
use pw::{properties::properties, spa}; use pw::{properties::properties, spa};
use spa::param::audio::{AudioFormat, AudioInfoRaw}; use spa::param::audio::{AudioFormat, AudioInfoRaw};
use spa::pod::Pod; use spa::pod::Pod;
crate::pwinit::ensure_init(); // Setup errors funnel through the ready handshake (mirrors mic_pw_thread's IIFE).
let mainloop = pw::main_loop::MainLoopRc::new(None).context("pw audio MainLoop")?; let result = (|| -> Result<()> {
let context = pw::context::ContextRc::new(&mainloop, None).context("pw audio Context")?; crate::pwinit::ensure_init();
let core = context let mainloop = pw::main_loop::MainLoopRc::new(None).context("pw audio MainLoop")?;
.connect_rc(None) let context = pw::context::ContextRc::new(&mainloop, None).context("pw audio Context")?;
.context("pw audio connect (is PipeWire running in this session?)")?; 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. // Cross-thread teardown: the capturer's Drop sends Terminate; quit the loop here.
let _quit_guard = quit_rx.attach(mainloop.loop_(), { let _quit_guard = quit_rx.attach(mainloop.loop_(), {
let mainloop = mainloop.clone(); let mainloop = mainloop.clone();
move |_| mainloop.quit() move |_| mainloop.quit()
}); });
let stream = pw::stream::StreamBox::new( // Death detection (same contract as the virtual mic below): a core error — the daemon
&core, // restarted/went away — ends this thread, so the chunk channel disconnects and
"punktfunk-audio", // `next_chunk` returns Err, engaging the sessions' reopen-with-backoff. Without this, a
properties! { // PipeWire restart mid-session left a zombie capture thread whose `next_chunk` returned
*pw::keys::MEDIA_TYPE => "Audio", // quiet-sink empty chunks forever — audio silently dead for the rest of the session.
*pw::keys::MEDIA_CATEGORY => "Capture", let _core_listener = core
*pw::keys::MEDIA_ROLE => "Music", .add_listener_local()
// Capture the default sink's monitor (system output), not a microphone. .error({
*pw::keys::STREAM_CAPTURE_SINK => "true", let mainloop = mainloop.clone();
// Ask for a ~5ms quantum (= one Opus frame) so buffers arrive smoothly rather than move |id, _seq, res, message| {
// in large bursts the client's low-latency jitter buffer would hear as glitching. tracing::warn!(id, res, message, "pipewire core error — audio capture ends");
*pw::keys::NODE_LATENCY => "240/48000", mainloop.quit();
}, }
) })
.context("pw audio Stream")?; .register();
let _listener = stream let props = match &sink_name {
.add_local_listener_with_user_data(tx) // Stream-sink mode: this stream IS the sink (media.class + Direction::Input). Apps
.state_changed(|_s, _ud, old, new| { // play into it, PipeWire mixes them, process() receives the mix. Mirrors the
tracing::info!(?old, ?new, "pipewire audio stream state"); // validated PwMicSource recipe (stream node + RT_PROCESS; see its property
}) // comments) — do NOT "modernize" either into a `support.null-audio-sink` adapter
.param_changed(|_stream, _tx, id, param| { // without re-running that validation.
let Some(param) = param else { return }; Some(name) => {
if id != pw::spa::param::ParamType::Format.as_raw() { let mut p = properties! {
return; *pw::keys::MEDIA_TYPE => "Audio",
} *pw::keys::MEDIA_CLASS => "Audio/Sink",
let mut info = AudioInfoRaw::default(); *pw::keys::NODE_DESCRIPTION => "Punktfunk Stream Speaker",
if info.parse(param).is_ok() { *pw::keys::NODE_VIRTUAL => "true",
tracing::info!( // Ask for a ~5ms quantum (= one Opus frame) so buffers arrive smoothly
format = ?info.format(), // rather than in bursts the client's jitter buffer would hear as glitching.
rate = info.rate(), *pw::keys::NODE_LATENCY => "240/48000",
channels = info.channels(), // LOW priority — the opposite of the mic's 3000: between sessions the sink
"audio format negotiated" // 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",
.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(); p.insert(*pw::keys::NODE_NAME, name.as_str());
if datas.is_empty() { p
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");
} }
}) // Legacy: capture the default sink's monitor (system output), not a microphone.
.register() None => properties! {
.context("register audio stream listener")?; *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 — let _listener = stream
// PipeWire's channel-mixer up/downmixes the sink monitor to this layout. .add_local_listener_with_user_data(tx)
let mut info = AudioInfoRaw::new(); .state_changed({
info.set_format(AudioFormat::F32LE); let mainloop = mainloop.clone();
info.set_rate(SAMPLE_RATE); move |_s, _ud, old, new| {
info.set_channels(channels); tracing::info!(?old, ?new, "pipewire audio stream state");
info.set_position(spa_positions(channels)); // A stream error is unrecoverable for this instance — exit so the sessions'
let obj = pw::spa::pod::Object { // reopen path builds a fresh one (same contract as the core-error path above).
type_: pw::spa::utils::SpaTypes::ObjectParamFormat.as_raw(), if matches!(new, pw::stream::StreamState::Error(_)) {
id: pw::spa::param::ParamType::EnumFormat.as_raw(), mainloop.quit();
properties: info.into(), }
}; }
let values: Vec<u8> = pw::spa::pod::serialize::PodSerializer::serialize( })
std::io::Cursor::new(Vec::new()), .param_changed(|_stream, _tx, id, param| {
&pw::spa::pod::Value::Object(obj), let Some(param) = param else { return };
) if id != pw::spa::param::ParamType::Format.as_raw() {
.context("serialize audio format pod")? return;
.0 }
.into_inner(); let mut info = AudioInfoRaw::default();
let mut params = [Pod::from_bytes(&values).context("audio pod from bytes")?]; 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 // Request F32LE, 48 kHz, at the session's channel count with explicit positions. In
.connect( // legacy mode PipeWire's channel-mixer up/downmixes the sink monitor to this layout;
spa::utils::Direction::Input, // in stream-sink mode this IS the sink's advertised layout (apps mix/route to it).
None, // PW_ID_ANY — autoconnect to the default sink monitor let mut info = AudioInfoRaw::new();
pw::stream::StreamFlags::AUTOCONNECT | pw::stream::StreamFlags::MAP_BUFFERS, info.set_format(AudioFormat::F32LE);
&mut params, 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<u8> = 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(); // RT_PROCESS in stream-sink mode for the same reason as the mic: the sink must be a
tracing::debug!("pipewire audio loop exited (capturer dropped)"); // *synchronous* graph node that joins its producers' driver group and is actually
Ok(()) // 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
} }
@@ -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-<pid>-<seq>`, 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":"<node.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<Restore>,
}
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<String>) {
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<Restore> {
self.holders = self.holders.saturating_sub(1);
if self.holders == 0 {
self.restore.take()
} else {
None
}
}
}
static LEDGER: Mutex<Ledger> = 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("<automatic>"),
"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<Option<String>> {
use pipewire as pw;
use std::cell::RefCell;
use std::rc::Rc;
crate::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<pw::metadata::Metadata>,
md_listener: Option<pw::metadata::MetadataListener>,
previous: Option<String>,
phase: u8,
expected: Option<pw::spa::utils::result::AsyncSeq>,
outcome: Option<Result<()>>,
}
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::<pw::metadata::Metadata, _>(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"
);
}
}
@@ -297,6 +297,7 @@ fn run(
None => audio::open_audio_capture(want).context("open audio capture")?, None => audio::open_audio_capture(want).context("open audio capture")?,
}; };
let result = audio_body(&mut *cap, &sock, gcm_key, rikeyid, params, running); 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); *audio_cap.lock().unwrap() = Some(cap);
result result
} }
+7 -4
View File
@@ -2501,9 +2501,9 @@ fn audio_thread(
// Reuse the cached capturer ONLY when its channel count matches this session's; a stereo // 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 // 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). // 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 => { 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 c
} }
prev => { prev => {
@@ -2521,6 +2521,7 @@ fn audio_thread(
Ok(e) => e, Ok(e) => e,
Err(e) => { Err(e) => {
tracing::error!(error = %e, "opus encoder"); tracing::error!(error = %e, "opus encoder");
capturer.idle(); // parked, not streaming — release the routing claim
*audio_cap.lock().unwrap() = Some(capturer); *audio_cap.lock().unwrap() = Some(capturer);
return; return;
} }
@@ -2590,8 +2591,10 @@ fn audio_thread(
} }
} }
} }
// Return the live capturer for the next session (None if it died and never reopened). // Return the live capturer for the next session (None if it died and never reopened),
if let Some(c) = capturer { // 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); *audio_cap.lock().unwrap() = Some(c);
} }
} }