fix(audio): close the teardown orphan race by wiring in the ledger
Teardown used to `event_task.abort()` and then read three `Option<u32>`s. The task's work was a synchronous `pactl` call with no await point, so the abort could not land until the load had already returned: teardown saw `None`, unloaded the sink, and the task then stored the new module's id into a mutex nobody would read again. An orphan loopback, pointing at a sink that no longer existed. Three changes close it: - Loads and unloads are `tokio::process::Command` with `kill_on_drop` and a bound, so cancellation is expressible at all. They are deliberately not `select!`ed against a cancel signal — dropping a completed load's index on the floor is the defect, not the fix. Cancellation happens by dropping the future, and the permit's `Drop` turns that into a question. - `Routing::shutdown` is async and *awaits* the event task through `&mut JoinHandle`, falling back to abort-then-await. Dropping the handle would detach the task, which is how a load could still land after teardown believed it had finished. It then runs two reconcile-then-unload rounds: one round can raise exactly one new question, and a second settles it. - `Drop` stays as the narrower synchronous backstop for the paths that never reach `shutdown`. It cannot await or reconcile, so when the ledger is left unexplained it says so and names `--repair`. A load whose outcome cannot be observed is now distinguished from one the server refused: a clean non-zero `pactl` exit abandons the permit (nothing was created), while a signal death, a timeout, an unreadable index or `PA_INVALID_INDEX` all leave it unsettled for reconciliation. Two live gates, both A/B against the real module table: teardown leaves it byte-identical with both modules carrying owner tokens, and a load cancelled mid-flight is reconciled rather than orphaned. The second asserts the slot is pending *before* reconciling, so it cannot pass by aborting before the load ever began. Both mutate global state, so they need `--test-threads=1` — running them in parallel makes each see the other's modules, which is how the first run failed. 273 tests, clippy clean under `-D warnings`, `--doctor` all checks pass. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
+375
-134
@@ -35,26 +35,45 @@ use std::cell::RefCell;
|
|||||||
use std::collections::BTreeMap;
|
use std::collections::BTreeMap;
|
||||||
use std::process::Command;
|
use std::process::Command;
|
||||||
use std::rc::Rc;
|
use std::rc::Rc;
|
||||||
use std::sync::{Arc, Mutex};
|
use std::sync::Arc;
|
||||||
use std::thread::JoinHandle;
|
use std::thread::JoinHandle;
|
||||||
|
use std::time::Duration;
|
||||||
|
|
||||||
use crate::cli::HostOpts;
|
use crate::cli::HostOpts;
|
||||||
use crate::repair::plan::{self as repair_plan, Shape};
|
use crate::host::ledger::{self, LedgerError, ModuleLedger, UnloadOutcome};
|
||||||
|
use crate::repair::plan::{self as repair_plan, Fingerprint, Shape};
|
||||||
|
|
||||||
|
/// How long a single `pactl load-module` / `unload-module` may take.
|
||||||
|
///
|
||||||
|
/// A bound, not a calibration: a local Pulse socket answers in milliseconds, and
|
||||||
|
/// this exists only so a wedged server cannot hang teardown forever. It is
|
||||||
|
/// deliberately generous because exceeding it is no longer destructive — the
|
||||||
|
/// ledger records the attempt, and reconciliation finds whatever the server
|
||||||
|
/// actually did.
|
||||||
|
const PACTL_BUDGET: Duration = Duration::from_secs(5);
|
||||||
|
|
||||||
|
/// How many reconcile-then-unload rounds teardown runs.
|
||||||
|
///
|
||||||
|
/// Two, because one round can create exactly one new question: an ambiguous load
|
||||||
|
/// resolves to a module that then needs unloading, and an uncertain unload
|
||||||
|
/// resolves to a module that is either gone or still there. A second round
|
||||||
|
/// settles either. Anything still unresolved after that is left to `--repair`
|
||||||
|
/// rather than looped over.
|
||||||
|
const TEARDOWN_ROUNDS: usize = 2;
|
||||||
|
|
||||||
/// Owns the pactl-loaded modules plus, when filtering is active, the
|
/// Owns the pactl-loaded modules plus, when filtering is active, the
|
||||||
/// libpipewire stream-router thread. Drop unloads modules as a backstop;
|
/// libpipewire stream-router thread. Drop unloads modules as a backstop;
|
||||||
/// prefer [`Routing::shutdown`] explicitly so failures get logged.
|
/// prefer [`Routing::shutdown`] explicitly, which is the only path that can
|
||||||
|
/// reconcile a load whose outcome was never observed.
|
||||||
pub struct Routing {
|
pub struct Routing {
|
||||||
sink_module: Option<u32>,
|
/// Every module this host has loaded, is loading, or must ask the server
|
||||||
/// Shared with the event task so it can `take()` and unload on the
|
/// about. Shared with the event task, which loads and unloads the two
|
||||||
/// first successful route. `Routing::shutdown` unloads whatever
|
/// loopbacks as the routed app comes and goes.
|
||||||
/// remains.
|
///
|
||||||
loopback_module: Arc<Mutex<Option<u32>>>,
|
/// This replaced three `Option<u32>`s. The reason is in [`crate::host::ledger`]:
|
||||||
/// The `null-sink.monitor → @DEFAULT_SINK@` loopback that lets the sharer
|
/// an `Option` cannot say "a load is in flight", so a cancelled load looked
|
||||||
/// hear the routed app. Shared with the event task, which loads it on the
|
/// exactly like no load at all and its module was left behind.
|
||||||
/// first routed stream and unloads it when the app stops. `None` outside
|
ledger: Arc<ModuleLedger>,
|
||||||
/// app mode and whenever no app is currently routed.
|
|
||||||
local_monitor_module: Arc<Mutex<Option<u32>>>,
|
|
||||||
sink_name: String,
|
sink_name: String,
|
||||||
stream_router: Option<StreamRouter>,
|
stream_router: Option<StreamRouter>,
|
||||||
event_task: Option<tokio::task::JoinHandle<()>>,
|
event_task: Option<tokio::task::JoinHandle<()>>,
|
||||||
@@ -66,13 +85,17 @@ impl Routing {
|
|||||||
pub async fn start(opts: &HostOpts) -> Result<Self> {
|
pub async fn start(opts: &HostOpts) -> Result<Self> {
|
||||||
let pid = std::process::id();
|
let pid = std::process::id();
|
||||||
let sink_name = repair_plan::sink_name_for(pid);
|
let sink_name = repair_plan::sink_name_for(pid);
|
||||||
|
let ledger = ModuleLedger::new();
|
||||||
|
|
||||||
// Every module this host loads carries an ownership token, minted per
|
// Every module this host loads carries an ownership token, minted per
|
||||||
// load, so `--repair` can tell whose pid the name refers to instead of
|
// load, so `--repair` can tell whose pid the name refers to instead of
|
||||||
// assuming the number means the same thing everywhere. Without it a repair
|
// assuming the number means the same thing everywhere. Without it a repair
|
||||||
// run in another pid namespace can unload a live host's audio; see
|
// run in another pid namespace can unload a live host's audio; see
|
||||||
// `repair::plan::OwnerToken`.
|
// `repair::plan::OwnerToken`. That same per-load nonce is what lets
|
||||||
let sink_module = load_module(Shape::LegacyCaptureSink, pid)
|
// reconciliation identify a module whose load was interrupted before its
|
||||||
|
// index was ever read.
|
||||||
|
load_module(&ledger, Shape::LegacyCaptureSink, pid)
|
||||||
|
.await
|
||||||
.context("failed to load module-null-sink")?;
|
.context("failed to load module-null-sink")?;
|
||||||
|
|
||||||
// In strict per-app mode we never mirror the default sink: the viewer
|
// In strict per-app mode we never mirror the default sink: the viewer
|
||||||
@@ -84,29 +107,20 @@ impl Routing {
|
|||||||
// 20ms loopback latency keeps the mirrored audio tight; pactl's
|
// 20ms loopback latency keeps the mirrored audio tight; pactl's
|
||||||
// default of 200ms is enough to be perceptible.
|
// default of 200ms is enough to be perceptible.
|
||||||
let strict_app = opts.app.is_some() && opts.strict_audio;
|
let strict_app = opts.app.is_some() && opts.strict_audio;
|
||||||
let loopback_module = if strict_app {
|
if !strict_app {
|
||||||
None
|
load_module(&ledger, Shape::LoopbackIntoCapture, pid)
|
||||||
} else {
|
.await
|
||||||
Some(
|
.context("failed to load module-loopback (null-sink cleaned up on Drop)")?;
|
||||||
load_module(Shape::LoopbackIntoCapture, pid)
|
}
|
||||||
.context("failed to load module-loopback (null-sink cleaned up on Drop)")?,
|
|
||||||
)
|
|
||||||
};
|
|
||||||
|
|
||||||
tracing::info!(
|
tracing::info!(
|
||||||
sink_module,
|
|
||||||
?loopback_module,
|
|
||||||
strict_app,
|
strict_app,
|
||||||
%sink_name,
|
%sink_name,
|
||||||
"audio routing: null-sink ready (loopback skipped in strict app mode)"
|
"audio routing: null-sink ready (loopback skipped in strict app mode)"
|
||||||
);
|
);
|
||||||
|
|
||||||
let loopback_arc = Arc::new(Mutex::new(loopback_module));
|
|
||||||
let local_monitor_arc = Arc::new(Mutex::new(None));
|
|
||||||
let mut routing = Self {
|
let mut routing = Self {
|
||||||
sink_module: Some(sink_module),
|
ledger: Arc::clone(&ledger),
|
||||||
loopback_module: Arc::clone(&loopback_arc),
|
|
||||||
local_monitor_module: Arc::clone(&local_monitor_arc),
|
|
||||||
sink_name: sink_name.clone(),
|
sink_name: sink_name.clone(),
|
||||||
stream_router: None,
|
stream_router: None,
|
||||||
event_task: None,
|
event_task: None,
|
||||||
@@ -114,21 +128,17 @@ impl Routing {
|
|||||||
|
|
||||||
if let Some(app) = &opts.app {
|
if let Some(app) = &opts.app {
|
||||||
let (router, mut event_rx) = StreamRouter::spawn(app.clone(), sink_name.clone())?;
|
let (router, mut event_rx) = StreamRouter::spawn(app.clone(), sink_name.clone())?;
|
||||||
let loopback_for_task = Arc::clone(&loopback_arc);
|
let ledger_for_task = Arc::clone(&ledger);
|
||||||
let local_monitor_for_task = Arc::clone(&local_monitor_arc);
|
|
||||||
let strict = opts.strict_audio;
|
let strict = opts.strict_audio;
|
||||||
let event_task = tokio::spawn(async move {
|
let event_task = tokio::spawn(async move {
|
||||||
use crate::common::output::{self, AppAudioState};
|
use crate::common::output::{self, AppAudioState};
|
||||||
while let Some(ev) = event_rx.recv().await {
|
while let Some(ev) = event_rx.recv().await {
|
||||||
match ev {
|
match ev {
|
||||||
Event::FirstRoutedStream => {
|
Event::FirstRoutedStream => {
|
||||||
let mid = loopback_for_task.lock().unwrap().take();
|
tracing::info!(
|
||||||
if let Some(id) = mid {
|
"audio routing: first stream routed → unloading default-sink loopback"
|
||||||
tracing::info!(
|
);
|
||||||
"audio routing: first stream routed → unloading default-sink loopback"
|
unload_module(&ledger_for_task, Shape::LoopbackIntoCapture).await;
|
||||||
);
|
|
||||||
unload_module(id);
|
|
||||||
}
|
|
||||||
// Mirror the routed app back to the sharer's own
|
// Mirror the routed app back to the sharer's own
|
||||||
// speakers so they hear the content they're sharing.
|
// speakers so they hear the content they're sharing.
|
||||||
// Loaded *after* the default-sink loopback is gone so
|
// Loaded *after* the default-sink loopback is gone so
|
||||||
@@ -136,20 +146,7 @@ impl Routing {
|
|||||||
// sourced from the null-sink monitor — the chosen app
|
// sourced from the null-sink monitor — the chosen app
|
||||||
// only, never the desktop/call — so it can't echo into
|
// only, never the desktop/call — so it can't echo into
|
||||||
// the capture.
|
// the capture.
|
||||||
if local_monitor_for_task.lock().unwrap().is_none() {
|
ensure_loaded(&ledger_for_task, Shape::LoopbackOutOfCapture, pid).await;
|
||||||
match load_module(Shape::LoopbackOutOfCapture, pid) {
|
|
||||||
Ok(id) => {
|
|
||||||
tracing::info!(
|
|
||||||
module = id,
|
|
||||||
"audio routing: local monitor loaded (sharer hears the shared app)"
|
|
||||||
);
|
|
||||||
*local_monitor_for_task.lock().unwrap() = Some(id);
|
|
||||||
}
|
|
||||||
Err(e) => tracing::warn!(
|
|
||||||
"audio routing: failed to load local monitor loopback: {e:#}"
|
|
||||||
),
|
|
||||||
}
|
|
||||||
}
|
|
||||||
// Tell the front-end the chosen app's audio is live.
|
// Tell the front-end the chosen app's audio is live.
|
||||||
output::emit(output::Event::AppAudio {
|
output::emit(output::Event::AppAudio {
|
||||||
state: AppAudioState::Routed,
|
state: AppAudioState::Routed,
|
||||||
@@ -164,13 +161,7 @@ impl Routing {
|
|||||||
// The shared app is gone, so its null-sink is silent:
|
// The shared app is gone, so its null-sink is silent:
|
||||||
// stop mirroring it to the sharer's speakers. Re-loads
|
// stop mirroring it to the sharer's speakers. Re-loads
|
||||||
// on the next FirstRoutedStream if the app resumes.
|
// on the next FirstRoutedStream if the app resumes.
|
||||||
if let Some(id) = local_monitor_for_task.lock().unwrap().take() {
|
unload_module(&ledger_for_task, Shape::LoopbackOutOfCapture).await;
|
||||||
tracing::info!(
|
|
||||||
module = id,
|
|
||||||
"audio routing: last routed stream gone → unloading local monitor"
|
|
||||||
);
|
|
||||||
unload_module(id);
|
|
||||||
}
|
|
||||||
if strict {
|
if strict {
|
||||||
// Strict mode: do NOT restore the whole-desktop
|
// Strict mode: do NOT restore the whole-desktop
|
||||||
// loopback. Viewers hear silence until the app
|
// loopback. Viewers hear silence until the app
|
||||||
@@ -183,23 +174,12 @@ impl Routing {
|
|||||||
}
|
}
|
||||||
// Best-effort mode: restore the default-sink loopback
|
// Best-effort mode: restore the default-sink loopback
|
||||||
// so the viewer hears system audio again instead of
|
// so the viewer hears system audio again instead of
|
||||||
// silence.
|
// silence. Already loaded is not an error — the ledger
|
||||||
if loopback_for_task.lock().unwrap().is_some() {
|
// refuses the load and `ensure_loaded` says so quietly.
|
||||||
continue;
|
|
||||||
}
|
|
||||||
tracing::info!(
|
tracing::info!(
|
||||||
"audio routing: last routed stream gone → restoring default-sink loopback"
|
"audio routing: last routed stream gone → restoring default-sink loopback"
|
||||||
);
|
);
|
||||||
match load_module(Shape::LoopbackIntoCapture, pid) {
|
ensure_loaded(&ledger_for_task, Shape::LoopbackIntoCapture, pid).await;
|
||||||
Ok(id) => {
|
|
||||||
*loopback_for_task.lock().unwrap() = Some(id);
|
|
||||||
}
|
|
||||||
Err(e) => {
|
|
||||||
tracing::warn!(
|
|
||||||
"audio routing: failed to re-load loopback: {e:#}"
|
|
||||||
);
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -225,42 +205,90 @@ impl Routing {
|
|||||||
&self.sink_name
|
&self.sink_name
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Stop the stream router (if any), then unload loopback (if still
|
/// Stop the stream router and the event task, settle anything the ledger is
|
||||||
/// loaded), then unload the null-sink. Order matters: PipeWire can
|
/// unsure about, then unload every module in shape order — the loopbacks
|
||||||
/// leave zombie links if you destroy a sink with active inputs.
|
/// before the sink they reference, because PipeWire can leave zombie links if
|
||||||
|
/// a sink is destroyed with active inputs.
|
||||||
///
|
///
|
||||||
/// Every step is a `take()`, so this is idempotent — `Drop` calls it again
|
/// The event task is **awaited, not merely aborted**. Aborting and walking
|
||||||
/// as a backstop and the second run is a no-op.
|
/// away is what left orphans behind: the task's load is an await point now,
|
||||||
fn cleanup(&mut self) {
|
/// so dropping its future marks the slot ambiguous rather than losing the
|
||||||
|
/// module — but only a path that then reconciles can actually clean it up.
|
||||||
|
/// `Drop` cannot await, which is why it is the narrower backstop.
|
||||||
|
pub async fn shutdown(mut self) {
|
||||||
|
if let Some(router) = self.stream_router.take() {
|
||||||
|
// ⚠️ Still an unbounded join: a wedged PipeWire thread parks this
|
||||||
|
// task indefinitely. That is the pre-existing defect S3b exists for.
|
||||||
|
// Nothing here makes it worse, and the ledger is what will make
|
||||||
|
// bounding it safe when it lands.
|
||||||
|
router.shutdown();
|
||||||
|
}
|
||||||
|
if let Some(mut task) = self.event_task.take() {
|
||||||
|
// The router's exit drops the event senders, so the task normally
|
||||||
|
// ends by itself. Abort is the fallback, and it is awaited through
|
||||||
|
// `&mut JoinHandle` so the future is genuinely dropped — and with it
|
||||||
|
// any in-flight permit — before reconciliation reads the ledger.
|
||||||
|
// Dropping the handle instead would *detach* the task, which is how a
|
||||||
|
// load could still land after teardown believed it was finished.
|
||||||
|
if tokio::time::timeout(PACTL_BUDGET, &mut task).await.is_err() {
|
||||||
|
tracing::warn!(
|
||||||
|
"audio routing: the event task did not finish within {PACTL_BUDGET:?}; \
|
||||||
|
cancelling it"
|
||||||
|
);
|
||||||
|
task.abort();
|
||||||
|
let _ = task.await;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
for _ in 0..TEARDOWN_ROUNDS {
|
||||||
|
if let Err(e) = ledger::reconcile_pending(&self.ledger).await {
|
||||||
|
tracing::warn!("audio routing: could not reconcile the module ledger: {e:#}");
|
||||||
|
}
|
||||||
|
let loaded = self.ledger.loaded();
|
||||||
|
if loaded.is_empty() {
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
for fp in loaded {
|
||||||
|
unload_module(&self.ledger, fp.shape).await;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
if !self.ledger.is_settled() {
|
||||||
|
tracing::warn!(
|
||||||
|
"audio routing: some audio modules could not be accounted for; \
|
||||||
|
`pixelpass --repair` will clean up anything left behind"
|
||||||
|
);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
impl Drop for Routing {
|
||||||
|
/// Synchronous backstop for the paths that never reach [`Routing::shutdown`]
|
||||||
|
/// — an error on the way up, or a panic. It cannot await, so it can neither
|
||||||
|
/// wait for the event task nor reconcile; it unloads what the ledger can name
|
||||||
|
/// and says so plainly when something is left unexplained.
|
||||||
|
///
|
||||||
|
/// After a completed `shutdown` the ledger holds nothing and this does nothing.
|
||||||
|
fn drop(&mut self) {
|
||||||
if let Some(router) = self.stream_router.take() {
|
if let Some(router) = self.stream_router.take() {
|
||||||
router.shutdown();
|
router.shutdown();
|
||||||
}
|
}
|
||||||
if let Some(task) = self.event_task.take() {
|
if let Some(task) = self.event_task.take() {
|
||||||
task.abort();
|
task.abort();
|
||||||
}
|
}
|
||||||
if let Some(id) = self.loopback_module.lock().unwrap().take() {
|
for fp in self.ledger.loaded() {
|
||||||
unload_module(id);
|
if self.ledger.begin_unload(fp.shape).is_none() {
|
||||||
|
continue;
|
||||||
|
}
|
||||||
|
let outcome = blocking_unload(fp.id);
|
||||||
|
self.ledger.finish_unload(fp.shape, outcome);
|
||||||
}
|
}
|
||||||
// Unload the local monitor before the null-sink it reads from, so the
|
if !self.ledger.is_settled() {
|
||||||
// sink has no active loopback reader when it's destroyed.
|
tracing::warn!(
|
||||||
if let Some(id) = self.local_monitor_module.lock().unwrap().take() {
|
"audio routing: torn down without settling the module ledger; \
|
||||||
unload_module(id);
|
run `pixelpass --repair` to clean up anything left behind"
|
||||||
|
);
|
||||||
}
|
}
|
||||||
if let Some(id) = self.sink_module.take() {
|
|
||||||
unload_module(id);
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
/// Consume the routing and tear it all down now. `Drop` is the backstop;
|
|
||||||
/// the real work lives in [`cleanup`](Self::cleanup).
|
|
||||||
pub fn shutdown(mut self) {
|
|
||||||
self.cleanup();
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
impl Drop for Routing {
|
|
||||||
fn drop(&mut self) {
|
|
||||||
self.cleanup();
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -374,54 +402,143 @@ fn owner_token(pid: u32) -> Result<repair_plan::OwnerToken> {
|
|||||||
/// `--repair`'s exact-form matcher and this loader are one source of truth. A
|
/// `--repair`'s exact-form matcher and this loader are one source of truth. A
|
||||||
/// latency or argument change that moved only one of them would leave repair
|
/// latency or argument change that moved only one of them would leave repair
|
||||||
/// silently unable to recognise the modules this build loads.
|
/// silently unable to recognise the modules this build loads.
|
||||||
fn load_module(shape: Shape, pid: u32) -> Result<u32> {
|
async fn load_module(ledger: &Arc<ModuleLedger>, shape: Shape, pid: u32) -> Result<Fingerprint> {
|
||||||
let owner = owner_token(pid).context("could not build an audio ownership token")?;
|
let owner = owner_token(pid).context("could not build an audio ownership token")?;
|
||||||
let output = Command::new("pactl")
|
let permit = ledger
|
||||||
.arg("load-module")
|
.begin_load(shape, pid, owner.clone())
|
||||||
|
.map_err(anyhow::Error::new)
|
||||||
|
.with_context(|| format!("cannot load the {} module", shape.label()))?;
|
||||||
|
|
||||||
|
let mut cmd = tokio::process::Command::new("pactl");
|
||||||
|
cmd.arg("load-module")
|
||||||
.arg(shape.module_name())
|
.arg(shape.module_name())
|
||||||
.args(shape.render_args(pid, Some(&owner)))
|
.args(shape.render_args(pid, Some(&owner)))
|
||||||
.output()
|
.kill_on_drop(true);
|
||||||
.context("failed to run pactl load-module")?;
|
|
||||||
|
// Deliberately **not** `select!`ed against a cancellation signal: a completed
|
||||||
|
// load whose index was then dropped on the floor is precisely the defect the
|
||||||
|
// ledger exists to prevent. Cancellation here happens by dropping this whole
|
||||||
|
// future, and the permit's `Drop` turns that into a question reconciliation
|
||||||
|
// can answer, rather than into silence.
|
||||||
|
let output = match tokio::time::timeout(PACTL_BUDGET, cmd.output()).await {
|
||||||
|
Ok(Ok(output)) => output,
|
||||||
|
Ok(Err(e)) => {
|
||||||
|
// Spawning or reading failed. Whether the server was ever reached is
|
||||||
|
// not knowable from here, so leave the permit unsettled: an
|
||||||
|
// unnecessary reconcile costs one listing, a missed one costs an
|
||||||
|
// orphan.
|
||||||
|
drop(permit);
|
||||||
|
return Err(e).context("failed to run pactl load-module");
|
||||||
|
}
|
||||||
|
Err(_) => {
|
||||||
|
drop(permit);
|
||||||
|
bail!("pactl load-module did not finish within {PACTL_BUDGET:?}");
|
||||||
|
}
|
||||||
|
};
|
||||||
|
|
||||||
if !output.status.success() {
|
if !output.status.success() {
|
||||||
bail!(
|
let stderr = String::from_utf8_lossy(&output.stderr).trim().to_string();
|
||||||
"pactl load-module failed: {}",
|
if output.status.code().is_some() {
|
||||||
String::from_utf8_lossy(&output.stderr).trim()
|
// pactl exited of its own accord, having reported the server's
|
||||||
);
|
// refusal: nothing was created, so there is nothing to reconcile.
|
||||||
|
permit.abandon();
|
||||||
|
} else {
|
||||||
|
// Killed by a signal, which may have arrived *after* the server
|
||||||
|
// created the module.
|
||||||
|
drop(permit);
|
||||||
|
}
|
||||||
|
bail!("pactl load-module failed: {stderr}");
|
||||||
}
|
}
|
||||||
let id_str = String::from_utf8(output.stdout)
|
|
||||||
.context("pactl returned non-UTF-8")?
|
let id_str = String::from_utf8_lossy(&output.stdout).trim().to_string();
|
||||||
.trim()
|
|
||||||
.to_string();
|
|
||||||
// Genuinely 32-bit, unlike `object.serial`: this is a PulseAudio module
|
// Genuinely 32-bit, unlike `object.serial`: this is a PulseAudio module
|
||||||
// index (`pa_module.index`, `uint32_t`), which `pactl unload-module` takes
|
// index (`pa_module.index`, `uint32_t`), which `pactl unload-module` takes
|
||||||
// back verbatim. Do not widen it.
|
// back verbatim. Do not widen it.
|
||||||
id_str
|
let Ok(index) = id_str.parse::<u32>() else {
|
||||||
.parse::<u32>()
|
// The load may well have succeeded — we simply cannot say which module it
|
||||||
.with_context(|| format!("pactl returned unexpected module ID: {id_str:?}"))
|
// produced, which is exactly what reconciliation is for.
|
||||||
|
drop(permit);
|
||||||
|
bail!("pactl returned unexpected module ID: {id_str:?}");
|
||||||
|
};
|
||||||
|
let fp = permit.commit(index)?;
|
||||||
|
tracing::info!(
|
||||||
|
module = fp.id,
|
||||||
|
shape = shape.label(),
|
||||||
|
"audio routing: loaded pactl module"
|
||||||
|
);
|
||||||
|
Ok(fp)
|
||||||
}
|
}
|
||||||
|
|
||||||
fn unload_module(id: u32) {
|
/// Load `shape` unless the slot already holds it.
|
||||||
let result = Command::new("pactl")
|
///
|
||||||
|
/// A busy slot is not a failure on the oscillation path: `FirstRoutedStream` and
|
||||||
|
/// `LastRoutedStreamGone` can both ask for a module that is already in the state
|
||||||
|
/// they want, and the ledger is what decides that rather than a separate flag.
|
||||||
|
async fn ensure_loaded(ledger: &Arc<ModuleLedger>, shape: Shape, pid: u32) {
|
||||||
|
let Err(e) = load_module(ledger, shape, pid).await else {
|
||||||
|
return;
|
||||||
|
};
|
||||||
|
match e.downcast_ref::<LedgerError>() {
|
||||||
|
Some(LedgerError::Busy { state, .. }) => tracing::debug!(
|
||||||
|
shape = shape.label(),
|
||||||
|
state,
|
||||||
|
"audio routing: nothing to load, the slot is already occupied"
|
||||||
|
),
|
||||||
|
_ => tracing::warn!(
|
||||||
|
"audio routing: failed to load the {} module: {e:#}",
|
||||||
|
shape.label()
|
||||||
|
),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Unload whatever the ledger holds for `shape`, and record how it went.
|
||||||
|
///
|
||||||
|
/// A no-op for a slot holding nothing. An outcome that cannot be confirmed is
|
||||||
|
/// recorded as uncertain rather than assumed done, so the module keeps being
|
||||||
|
/// named until the server is asked about it.
|
||||||
|
async fn unload_module(ledger: &Arc<ModuleLedger>, shape: Shape) {
|
||||||
|
let Some(fp) = ledger.begin_unload(shape) else {
|
||||||
|
return;
|
||||||
|
};
|
||||||
|
let mut cmd = tokio::process::Command::new("pactl");
|
||||||
|
cmd.arg("unload-module")
|
||||||
|
.arg(fp.id.to_string())
|
||||||
|
.kill_on_drop(true);
|
||||||
|
let outcome = match tokio::time::timeout(PACTL_BUDGET, cmd.output()).await {
|
||||||
|
Ok(Ok(output)) if output.status.success() => {
|
||||||
|
tracing::info!(module = fp.id, "audio routing: unloaded pactl module");
|
||||||
|
UnloadOutcome::Confirmed
|
||||||
|
}
|
||||||
|
Ok(Ok(output)) => UnloadOutcome::Uncertain(format!(
|
||||||
|
"pactl unload-module exited {}: {}",
|
||||||
|
output.status,
|
||||||
|
String::from_utf8_lossy(&output.stderr).trim()
|
||||||
|
)),
|
||||||
|
Ok(Err(e)) => UnloadOutcome::Uncertain(format!("could not run pactl unload-module: {e}")),
|
||||||
|
Err(_) => {
|
||||||
|
UnloadOutcome::Uncertain(format!("pactl unload-module exceeded {PACTL_BUDGET:?}"))
|
||||||
|
}
|
||||||
|
};
|
||||||
|
ledger.finish_unload(shape, outcome);
|
||||||
|
}
|
||||||
|
|
||||||
|
/// The blocking unload [`Drop`] uses, since it has no runtime to await on.
|
||||||
|
fn blocking_unload(id: u32) -> UnloadOutcome {
|
||||||
|
match Command::new("pactl")
|
||||||
.arg("unload-module")
|
.arg("unload-module")
|
||||||
.arg(id.to_string())
|
.arg(id.to_string())
|
||||||
.output();
|
.output()
|
||||||
match result {
|
{
|
||||||
Ok(output) if output.status.success() => {
|
Ok(output) if output.status.success() => {
|
||||||
tracing::info!(module = id, "audio routing: unloaded pactl module");
|
tracing::info!(module = id, "audio routing: unloaded pactl module");
|
||||||
|
UnloadOutcome::Confirmed
|
||||||
}
|
}
|
||||||
Ok(output) => {
|
Ok(output) => UnloadOutcome::Uncertain(format!(
|
||||||
tracing::warn!(
|
"pactl unload-module exited {}: {}",
|
||||||
module = id,
|
output.status,
|
||||||
stderr = %String::from_utf8_lossy(&output.stderr).trim(),
|
String::from_utf8_lossy(&output.stderr).trim()
|
||||||
"audio routing: pactl unload-module exited non-zero"
|
)),
|
||||||
);
|
Err(e) => UnloadOutcome::Uncertain(format!("could not run pactl unload-module: {e}")),
|
||||||
}
|
|
||||||
Err(e) => {
|
|
||||||
tracing::warn!(
|
|
||||||
module = id,
|
|
||||||
"audio routing: failed to run pactl unload-module: {e}"
|
|
||||||
);
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -718,6 +835,130 @@ fn try_flush(
|
|||||||
#[cfg(test)]
|
#[cfg(test)]
|
||||||
mod tests {
|
mod tests {
|
||||||
use super::*;
|
use super::*;
|
||||||
|
use crate::repair::plan::{ModuleObservation, classify};
|
||||||
|
|
||||||
|
/// Whole-desktop routing: no app filter, so no PipeWire thread and no event
|
||||||
|
/// task — just the null-sink and its default-sink loopback.
|
||||||
|
fn whole_desktop_opts() -> HostOpts {
|
||||||
|
HostOpts {
|
||||||
|
window: false,
|
||||||
|
app: None,
|
||||||
|
strict_audio: false,
|
||||||
|
display_server: None,
|
||||||
|
quality: crate::cli::Quality::Auto,
|
||||||
|
bitrate: None,
|
||||||
|
framerate: None,
|
||||||
|
max_height: None,
|
||||||
|
no_hwencode: false,
|
||||||
|
max_viewers: None,
|
||||||
|
interactive: false,
|
||||||
|
relay: None,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// The module table exactly as `--repair` observes it.
|
||||||
|
fn module_snapshot() -> Vec<(u32, String, String)> {
|
||||||
|
let mut session =
|
||||||
|
crate::repair::introspect::PulseSession::connect().expect("a local Pulse server");
|
||||||
|
session
|
||||||
|
.list_modules()
|
||||||
|
.expect("the server lists its modules")
|
||||||
|
.into_iter()
|
||||||
|
.map(|m| (m.id, m.name, m.args))
|
||||||
|
.collect()
|
||||||
|
}
|
||||||
|
|
||||||
|
/// A/B against the live graph: routing must leave the module table exactly
|
||||||
|
/// as it found it. The same shape as `--repair`'s field gate, because the
|
||||||
|
/// property is the same one — nothing of ours outlives the session.
|
||||||
|
#[tokio::test]
|
||||||
|
#[ignore = "loads real Pulse modules; run with --ignored --test-threads=1"]
|
||||||
|
async fn live_teardown_leaves_the_module_table_as_it_found_it() {
|
||||||
|
let before = module_snapshot();
|
||||||
|
let routing = Routing::start(&whole_desktop_opts())
|
||||||
|
.await
|
||||||
|
.expect("routing starts");
|
||||||
|
|
||||||
|
let during = module_snapshot();
|
||||||
|
let ours: Vec<_> = during
|
||||||
|
.iter()
|
||||||
|
.filter(|m| !before.iter().any(|b| b.0 == m.0))
|
||||||
|
.collect();
|
||||||
|
assert_eq!(
|
||||||
|
ours.len(),
|
||||||
|
2,
|
||||||
|
"the null-sink and its default-sink loopback must both be loaded"
|
||||||
|
);
|
||||||
|
for (id, name, args) in ours {
|
||||||
|
let fp = classify(&ModuleObservation::new(*id, name, args))
|
||||||
|
.expect("a module we loaded must match one of our canonical forms");
|
||||||
|
assert!(
|
||||||
|
fp.owner.is_some(),
|
||||||
|
"every module we load carries an owner token, or --repair cannot \
|
||||||
|
attribute it to this host's pid namespace"
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
|
routing.shutdown().await;
|
||||||
|
assert_eq!(
|
||||||
|
module_snapshot(),
|
||||||
|
before,
|
||||||
|
"teardown must leave the module table byte-identical"
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
|
/// The orphan race, staged against a real server: a load cancelled while
|
||||||
|
/// `pactl` is in flight must still be findable and removable.
|
||||||
|
///
|
||||||
|
/// Whether the server got as far as creating the module is genuinely racy,
|
||||||
|
/// and that is the point — the gate does not care which way it went, only
|
||||||
|
/// that the ledger can account for both. With the permit's `Drop` disarmed
|
||||||
|
/// and the module created, the final comparison fails.
|
||||||
|
#[tokio::test]
|
||||||
|
#[ignore = "loads real Pulse modules; run with --ignored --test-threads=1"]
|
||||||
|
async fn live_a_cancelled_load_is_reconciled_not_orphaned() {
|
||||||
|
let before = module_snapshot();
|
||||||
|
let ledger = ModuleLedger::new();
|
||||||
|
let pid = std::process::id();
|
||||||
|
|
||||||
|
let ledger_for_task = Arc::clone(&ledger);
|
||||||
|
let task = tokio::spawn(async move {
|
||||||
|
let _ = load_module(&ledger_for_task, Shape::LegacyCaptureSink, pid).await;
|
||||||
|
});
|
||||||
|
// Let the task run up to its first await — the spawned `pactl` — so the
|
||||||
|
// abort lands mid-flight rather than before the load ever started, which
|
||||||
|
// would make this gate vacuous.
|
||||||
|
tokio::task::yield_now().await;
|
||||||
|
task.abort();
|
||||||
|
let _ = task.await;
|
||||||
|
|
||||||
|
// Non-vacuity: the permit is taken *before* `pactl` is spawned, so a
|
||||||
|
// cancelled load must leave a question behind whichever side of the spawn
|
||||||
|
// the abort landed on. Without this the gate could pass while the abort
|
||||||
|
// fired before the load ever began, proving nothing.
|
||||||
|
assert_eq!(
|
||||||
|
ledger.pending().len(),
|
||||||
|
1,
|
||||||
|
"the cancelled load must have left exactly one question behind"
|
||||||
|
);
|
||||||
|
|
||||||
|
ledger::reconcile_pending(&ledger)
|
||||||
|
.await
|
||||||
|
.expect("the ledger reconciles against the server");
|
||||||
|
for fp in ledger.loaded() {
|
||||||
|
unload_module(&ledger, fp.shape).await;
|
||||||
|
}
|
||||||
|
|
||||||
|
assert!(
|
||||||
|
ledger.is_settled(),
|
||||||
|
"every slot must end in a state we can explain"
|
||||||
|
);
|
||||||
|
assert_eq!(
|
||||||
|
module_snapshot(),
|
||||||
|
before,
|
||||||
|
"a cancelled load must leave nothing behind"
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn object_serial_parses_past_u32() {
|
fn object_serial_parses_past_u32() {
|
||||||
|
|||||||
+7
-12
@@ -191,6 +191,13 @@ impl ModuleLedger {
|
|||||||
}
|
}
|
||||||
|
|
||||||
/// The current state of one slot. Absent keys read as [`SlotState::Vacant`].
|
/// The current state of one slot. Absent keys read as [`SlotState::Vacant`].
|
||||||
|
///
|
||||||
|
/// Test-only: production code never needs to look a slot up, because every
|
||||||
|
/// decision that depends on one is made *by* the ledger — `begin_load`
|
||||||
|
/// refuses a busy slot and says which state refused, `begin_unload` returns
|
||||||
|
/// nothing for a slot holding nothing. An accessor callers could branch on
|
||||||
|
/// would invite exactly the check-then-act races the permit removes.
|
||||||
|
#[cfg(test)]
|
||||||
pub fn state(&self, shape: Shape) -> SlotState {
|
pub fn state(&self, shape: Shape) -> SlotState {
|
||||||
self.slots
|
self.slots
|
||||||
.lock()
|
.lock()
|
||||||
@@ -200,10 +207,6 @@ impl ModuleLedger {
|
|||||||
.unwrap_or(SlotState::Vacant)
|
.unwrap_or(SlotState::Vacant)
|
||||||
}
|
}
|
||||||
|
|
||||||
fn set(&self, shape: Shape, state: SlotState) {
|
|
||||||
self.slots.lock().unwrap().insert(shape, state);
|
|
||||||
}
|
|
||||||
|
|
||||||
/// Take permission to load `shape`, moving the slot to
|
/// Take permission to load `shape`, moving the slot to
|
||||||
/// [`SlotState::Loading`].
|
/// [`SlotState::Loading`].
|
||||||
///
|
///
|
||||||
@@ -372,14 +375,6 @@ pub struct LoadPermit {
|
|||||||
}
|
}
|
||||||
|
|
||||||
impl LoadPermit {
|
impl LoadPermit {
|
||||||
pub fn shape(&self) -> Shape {
|
|
||||||
self.shape
|
|
||||||
}
|
|
||||||
|
|
||||||
pub fn token(&self) -> &OwnerToken {
|
|
||||||
&self.token
|
|
||||||
}
|
|
||||||
|
|
||||||
/// Record that the server created the module at `index`.
|
/// Record that the server created the module at `index`.
|
||||||
///
|
///
|
||||||
/// Fails on [`PA_INVALID_INDEX`], leaving the permit unsettled so that
|
/// Fails on [`PA_INVALID_INDEX`], leaving the permit unsettled so that
|
||||||
|
|||||||
@@ -47,7 +47,7 @@ impl CaptureHandle {
|
|||||||
let _ = child.start_kill();
|
let _ = child.start_kill();
|
||||||
}
|
}
|
||||||
if let Some(audio) = self.audio.take() {
|
if let Some(audio) = self.audio.take() {
|
||||||
audio.shutdown();
|
audio.shutdown().await;
|
||||||
}
|
}
|
||||||
if let Some(serve) = self.serve.take() {
|
if let Some(serve) = self.serve.take() {
|
||||||
serve.shutdown().await;
|
serve.shutdown().await;
|
||||||
|
|||||||
Reference in New Issue
Block a user