merge(audio): land feat/linux-stream-sink — Linux host-owned stream sink

Merges 44b71e74 (see its message for the full design: the "Punktfunk Stream
Speaker" Audio/Sink claimed as the session's default output — the crackle
root fix for hardware-sink churn — plus the core-error liveness fix and true
5.1/7.1 capture) across the W1–W8 refactor that landed since the branch
forked:

- punktfunk1.rs was split on main: the audio_thread idle()/drain() hunks are
  hand-ported into native/audio.rs (same three sites: reuse-drain comment,
  encoder-fail park, end-of-thread park).
- pwinit moved to pf-capture (W6.2): the branch's crate::pwinit calls in
  pw_thread and stream_sink.rs become pf_capture::pwinit.
- Kept main's log-tier demotions (stream-state + setup lines at debug)
  inside the restructured pw_thread.

Resolution verified on Linux (home-worker-5): clippy -D warnings clean +
173 punktfunk-host tests green at the 691c064a-based equivalent ba5973a2;
re-verified at this base before landing.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
2026-07-17 16:35:02 +02:00
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
//! 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<Box<dyn AudioCapturer>> {
linux::PwAudioCapturer::open(channels).map(|c| Box::new(c) as Box<dyn AudioCapturer>)
+178 -27
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
//! 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<Vec<f32>>,
channels: u32,
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 {
@@ -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::<Vec<f32>>(64);
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()
.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,12 +570,16 @@ fn pw_thread(
tx: std::sync::mpsc::SyncSender<Vec<f32>>,
quit_rx: pipewire::channel::Receiver<Terminate>,
channels: u32,
sink_name: Option<String>,
ready: std::sync::mpsc::SyncSender<Result<()>>,
) -> Result<()> {
use pipewire as pw;
use pw::{properties::properties, spa};
use spa::param::audio::{AudioFormat, AudioInfoRaw};
use spa::pod::Pod;
// 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")?;
@@ -506,26 +593,70 @@ fn pw_thread(
move |_| mainloop.quit()
});
let stream = pw::stream::StreamBox::new(
&core,
"punktfunk-audio",
properties! {
// 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 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",
};
p.insert(*pw::keys::NODE_NAME, name.as_str());
p
}
// 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",
// 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",
},
)
};
let stream = pw::stream::StreamBox::new(&core, "punktfunk-audio", props)
.context("pw audio Stream")?;
let _listener = stream
.add_local_listener_with_user_data(tx)
.state_changed(|_s, _ud, old, new| {
.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 };
@@ -587,8 +718,9 @@ fn pw_thread(
.register()
.context("register audio stream listener")?;
// 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.
// 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);
@@ -608,16 +740,35 @@ fn pw_thread(
.into_inner();
let mut params = [Pod::from_bytes(&values).context("audio pod from bytes")?];
// 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,
None, // PW_ID_ANY — autoconnect to the default sink monitor
pw::stream::StreamFlags::AUTOCONNECT | pw::stream::StreamFlags::MAP_BUFFERS,
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;
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<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")?,
};
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
}
+7 -4
View File
@@ -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);
}
}