Compare commits
2
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
  molluskandClaude Opus 5
|
fa792b9927 | ||
|
molluskandClaude Opus 5
|
cc9694c13f |
+375
-134
@@ -35,26 +35,45 @@ use std::cell::RefCell;
|
||||
use std::collections::BTreeMap;
|
||||
use std::process::Command;
|
||||
use std::rc::Rc;
|
||||
use std::sync::{Arc, Mutex};
|
||||
use std::sync::Arc;
|
||||
use std::thread::JoinHandle;
|
||||
use std::time::Duration;
|
||||
|
||||
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
|
||||
/// 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 {
|
||||
sink_module: Option<u32>,
|
||||
/// Shared with the event task so it can `take()` and unload on the
|
||||
/// first successful route. `Routing::shutdown` unloads whatever
|
||||
/// remains.
|
||||
loopback_module: Arc<Mutex<Option<u32>>>,
|
||||
/// The `null-sink.monitor → @DEFAULT_SINK@` loopback that lets the sharer
|
||||
/// hear the routed app. Shared with the event task, which loads it on the
|
||||
/// first routed stream and unloads it when the app stops. `None` outside
|
||||
/// app mode and whenever no app is currently routed.
|
||||
local_monitor_module: Arc<Mutex<Option<u32>>>,
|
||||
/// Every module this host has loaded, is loading, or must ask the server
|
||||
/// about. Shared with the event task, which loads and unloads the two
|
||||
/// loopbacks as the routed app comes and goes.
|
||||
///
|
||||
/// This replaced three `Option<u32>`s. The reason is in [`crate::host::ledger`]:
|
||||
/// an `Option` cannot say "a load is in flight", so a cancelled load looked
|
||||
/// exactly like no load at all and its module was left behind.
|
||||
ledger: Arc<ModuleLedger>,
|
||||
sink_name: String,
|
||||
stream_router: Option<StreamRouter>,
|
||||
event_task: Option<tokio::task::JoinHandle<()>>,
|
||||
@@ -66,13 +85,17 @@ impl Routing {
|
||||
pub async fn start(opts: &HostOpts) -> Result<Self> {
|
||||
let pid = std::process::id();
|
||||
let sink_name = repair_plan::sink_name_for(pid);
|
||||
let ledger = ModuleLedger::new();
|
||||
|
||||
// Every module this host loads carries an ownership token, minted per
|
||||
// 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
|
||||
// run in another pid namespace can unload a live host's audio; see
|
||||
// `repair::plan::OwnerToken`.
|
||||
let sink_module = load_module(Shape::LegacyCaptureSink, pid)
|
||||
// `repair::plan::OwnerToken`. That same per-load nonce is what lets
|
||||
// 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")?;
|
||||
|
||||
// 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
|
||||
// default of 200ms is enough to be perceptible.
|
||||
let strict_app = opts.app.is_some() && opts.strict_audio;
|
||||
let loopback_module = if strict_app {
|
||||
None
|
||||
} else {
|
||||
Some(
|
||||
load_module(Shape::LoopbackIntoCapture, pid)
|
||||
.context("failed to load module-loopback (null-sink cleaned up on Drop)")?,
|
||||
)
|
||||
};
|
||||
if !strict_app {
|
||||
load_module(&ledger, Shape::LoopbackIntoCapture, pid)
|
||||
.await
|
||||
.context("failed to load module-loopback (null-sink cleaned up on Drop)")?;
|
||||
}
|
||||
|
||||
tracing::info!(
|
||||
sink_module,
|
||||
?loopback_module,
|
||||
strict_app,
|
||||
%sink_name,
|
||||
"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 {
|
||||
sink_module: Some(sink_module),
|
||||
loopback_module: Arc::clone(&loopback_arc),
|
||||
local_monitor_module: Arc::clone(&local_monitor_arc),
|
||||
ledger: Arc::clone(&ledger),
|
||||
sink_name: sink_name.clone(),
|
||||
stream_router: None,
|
||||
event_task: None,
|
||||
@@ -114,21 +128,17 @@ impl Routing {
|
||||
|
||||
if let Some(app) = &opts.app {
|
||||
let (router, mut event_rx) = StreamRouter::spawn(app.clone(), sink_name.clone())?;
|
||||
let loopback_for_task = Arc::clone(&loopback_arc);
|
||||
let local_monitor_for_task = Arc::clone(&local_monitor_arc);
|
||||
let ledger_for_task = Arc::clone(&ledger);
|
||||
let strict = opts.strict_audio;
|
||||
let event_task = tokio::spawn(async move {
|
||||
use crate::common::output::{self, AppAudioState};
|
||||
while let Some(ev) = event_rx.recv().await {
|
||||
match ev {
|
||||
Event::FirstRoutedStream => {
|
||||
let mid = loopback_for_task.lock().unwrap().take();
|
||||
if let Some(id) = mid {
|
||||
tracing::info!(
|
||||
"audio routing: first stream routed → unloading default-sink loopback"
|
||||
);
|
||||
unload_module(id);
|
||||
}
|
||||
tracing::info!(
|
||||
"audio routing: first stream routed → unloading default-sink loopback"
|
||||
);
|
||||
unload_module(&ledger_for_task, Shape::LoopbackIntoCapture).await;
|
||||
// Mirror the routed app back to the sharer's own
|
||||
// speakers so they hear the content they're sharing.
|
||||
// Loaded *after* the default-sink loopback is gone so
|
||||
@@ -136,20 +146,7 @@ impl Routing {
|
||||
// sourced from the null-sink monitor — the chosen app
|
||||
// only, never the desktop/call — so it can't echo into
|
||||
// the capture.
|
||||
if local_monitor_for_task.lock().unwrap().is_none() {
|
||||
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:#}"
|
||||
),
|
||||
}
|
||||
}
|
||||
ensure_loaded(&ledger_for_task, Shape::LoopbackOutOfCapture, pid).await;
|
||||
// Tell the front-end the chosen app's audio is live.
|
||||
output::emit(output::Event::AppAudio {
|
||||
state: AppAudioState::Routed,
|
||||
@@ -164,13 +161,7 @@ impl Routing {
|
||||
// The shared app is gone, so its null-sink is silent:
|
||||
// stop mirroring it to the sharer's speakers. Re-loads
|
||||
// on the next FirstRoutedStream if the app resumes.
|
||||
if let Some(id) = local_monitor_for_task.lock().unwrap().take() {
|
||||
tracing::info!(
|
||||
module = id,
|
||||
"audio routing: last routed stream gone → unloading local monitor"
|
||||
);
|
||||
unload_module(id);
|
||||
}
|
||||
unload_module(&ledger_for_task, Shape::LoopbackOutOfCapture).await;
|
||||
if strict {
|
||||
// Strict mode: do NOT restore the whole-desktop
|
||||
// loopback. Viewers hear silence until the app
|
||||
@@ -183,23 +174,12 @@ impl Routing {
|
||||
}
|
||||
// Best-effort mode: restore the default-sink loopback
|
||||
// so the viewer hears system audio again instead of
|
||||
// silence.
|
||||
if loopback_for_task.lock().unwrap().is_some() {
|
||||
continue;
|
||||
}
|
||||
// silence. Already loaded is not an error — the ledger
|
||||
// refuses the load and `ensure_loaded` says so quietly.
|
||||
tracing::info!(
|
||||
"audio routing: last routed stream gone → restoring default-sink loopback"
|
||||
);
|
||||
match load_module(Shape::LoopbackIntoCapture, pid) {
|
||||
Ok(id) => {
|
||||
*loopback_for_task.lock().unwrap() = Some(id);
|
||||
}
|
||||
Err(e) => {
|
||||
tracing::warn!(
|
||||
"audio routing: failed to re-load loopback: {e:#}"
|
||||
);
|
||||
}
|
||||
}
|
||||
ensure_loaded(&ledger_for_task, Shape::LoopbackIntoCapture, pid).await;
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -225,42 +205,90 @@ impl Routing {
|
||||
&self.sink_name
|
||||
}
|
||||
|
||||
/// Stop the stream router (if any), then unload loopback (if still
|
||||
/// loaded), then unload the null-sink. Order matters: PipeWire can
|
||||
/// leave zombie links if you destroy a sink with active inputs.
|
||||
/// Stop the stream router and the event task, settle anything the ledger is
|
||||
/// unsure about, then unload every module in shape order — the loopbacks
|
||||
/// 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
|
||||
/// as a backstop and the second run is a no-op.
|
||||
fn cleanup(&mut self) {
|
||||
/// The event task is **awaited, not merely aborted**. Aborting and walking
|
||||
/// away is what left orphans behind: the task's load is an await point now,
|
||||
/// 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() {
|
||||
router.shutdown();
|
||||
}
|
||||
if let Some(task) = self.event_task.take() {
|
||||
task.abort();
|
||||
}
|
||||
if let Some(id) = self.loopback_module.lock().unwrap().take() {
|
||||
unload_module(id);
|
||||
for fp in self.ledger.loaded() {
|
||||
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
|
||||
// sink has no active loopback reader when it's destroyed.
|
||||
if let Some(id) = self.local_monitor_module.lock().unwrap().take() {
|
||||
unload_module(id);
|
||||
if !self.ledger.is_settled() {
|
||||
tracing::warn!(
|
||||
"audio routing: torn down without settling the module ledger; \
|
||||
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
|
||||
/// latency or argument change that moved only one of them would leave repair
|
||||
/// 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 output = Command::new("pactl")
|
||||
.arg("load-module")
|
||||
let permit = ledger
|
||||
.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())
|
||||
.args(shape.render_args(pid, Some(&owner)))
|
||||
.output()
|
||||
.context("failed to run pactl load-module")?;
|
||||
.kill_on_drop(true);
|
||||
|
||||
// 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() {
|
||||
bail!(
|
||||
"pactl load-module failed: {}",
|
||||
String::from_utf8_lossy(&output.stderr).trim()
|
||||
);
|
||||
let stderr = String::from_utf8_lossy(&output.stderr).trim().to_string();
|
||||
if output.status.code().is_some() {
|
||||
// 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")?
|
||||
.trim()
|
||||
.to_string();
|
||||
|
||||
let id_str = String::from_utf8_lossy(&output.stdout).trim().to_string();
|
||||
// Genuinely 32-bit, unlike `object.serial`: this is a PulseAudio module
|
||||
// index (`pa_module.index`, `uint32_t`), which `pactl unload-module` takes
|
||||
// back verbatim. Do not widen it.
|
||||
id_str
|
||||
.parse::<u32>()
|
||||
.with_context(|| format!("pactl returned unexpected module ID: {id_str:?}"))
|
||||
let Ok(index) = id_str.parse::<u32>() else {
|
||||
// The load may well have succeeded — we simply cannot say which module it
|
||||
// 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) {
|
||||
let result = Command::new("pactl")
|
||||
/// Load `shape` unless the slot already holds it.
|
||||
///
|
||||
/// 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(id.to_string())
|
||||
.output();
|
||||
match result {
|
||||
.output()
|
||||
{
|
||||
Ok(output) if output.status.success() => {
|
||||
tracing::info!(module = id, "audio routing: unloaded pactl module");
|
||||
UnloadOutcome::Confirmed
|
||||
}
|
||||
Ok(output) => {
|
||||
tracing::warn!(
|
||||
module = id,
|
||||
stderr = %String::from_utf8_lossy(&output.stderr).trim(),
|
||||
"audio routing: pactl unload-module exited non-zero"
|
||||
);
|
||||
}
|
||||
Err(e) => {
|
||||
tracing::warn!(
|
||||
module = id,
|
||||
"audio routing: failed to run pactl unload-module: {e}"
|
||||
);
|
||||
}
|
||||
Ok(output) => UnloadOutcome::Uncertain(format!(
|
||||
"pactl unload-module exited {}: {}",
|
||||
output.status,
|
||||
String::from_utf8_lossy(&output.stderr).trim()
|
||||
)),
|
||||
Err(e) => UnloadOutcome::Uncertain(format!("could not run pactl unload-module: {e}")),
|
||||
}
|
||||
}
|
||||
|
||||
@@ -718,6 +835,130 @@ fn try_flush(
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
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]
|
||||
fn object_serial_parses_past_u32() {
|
||||
|
||||
@@ -0,0 +1,890 @@
|
||||
//! What this host has put into the Pulse module table — and, where it cannot be
|
||||
//! sure, what it must go and find out before doing anything else.
|
||||
//!
|
||||
//! # The defect this exists for
|
||||
//!
|
||||
//! Module ids used to live in three `Option<u32>`s, two of them shared with the
|
||||
//! event task behind an `Arc<Mutex<…>>`. That representation cannot express *a
|
||||
//! load is in flight*, and the gap is reachable today: teardown calls
|
||||
//! `event_task.abort()` and then reads the ids, but the task's work is a
|
||||
//! synchronous `pactl` call with no await point inside it, so the abort cannot
|
||||
//! land until the load has already returned. Teardown therefore sees `None`,
|
||||
//! unloads the sink, and the still-running task stores the new module's id into a
|
||||
//! mutex nobody will ever read again — an orphan loopback pointing at a sink that
|
||||
//! no longer exists.
|
||||
//!
|
||||
//! A slot is consequently not an `Option<u32>` but a small state machine, and the
|
||||
//! transitions that matter are the ones that admit *we do not know*:
|
||||
//!
|
||||
//! ```text
|
||||
//! begin_load commit
|
||||
//! Vacant ───────────────► Loading ───────────────► Loaded
|
||||
//! ▲ │ │ │
|
||||
//! │ abandon │ │ permit dropped │ begin_unload
|
||||
//! └────────────────────┘ │ unsettled ▼
|
||||
//! ▲ ▼ Unloading
|
||||
//! │ Resolution::Nothing Ambiguous ◄──────────────┘
|
||||
//! └───────────────────────► │ ▲ uncertain outcome
|
||||
//! │ │
|
||||
//! Resolution::Conflict▼ └── Resolution::Adopt ──► Loaded
|
||||
//! Poisoned
|
||||
//! ```
|
||||
//!
|
||||
//! # Why the permit is affine
|
||||
//!
|
||||
//! [`LoadPermit`] is not `Clone`, is consumed by value to settle, and its [`Drop`]
|
||||
//! marks the slot [`Reconcile::Load`] when it was never settled. That is what makes
|
||||
//! the guarantee structural rather than a discipline: a cancelled task drops its
|
||||
//! locals, so an aborted load *cannot* silently forget a module the server may
|
||||
//! already have created. Two permitted loads for one slot cannot both commit,
|
||||
//! because [`ModuleLedger::begin_load`] issues a permit only for a `Vacant` slot
|
||||
//! and every other state — including the ambiguous one — refuses.
|
||||
//!
|
||||
//! # Why an ambiguous slot blocks the next load
|
||||
//!
|
||||
//! An unresolved load may or may not have created a module carrying our sink name.
|
||||
//! Starting another one on top of it risks two sinks sharing a `node.name`, which
|
||||
//! is measurably not an error the server reports: it accepts both, and
|
||||
//! `pulsesrc device=<name>.monitor` attaches to the *older* one. A second capture
|
||||
//! session would then be silently stolen by the debris of the first. Refusing to
|
||||
//! proceed until the question is answered is the whole point.
|
||||
|
||||
use std::collections::BTreeMap;
|
||||
use std::sync::atomic::{AtomicU64, Ordering};
|
||||
use std::sync::{Arc, Mutex};
|
||||
|
||||
use anyhow::{Context as _, Result};
|
||||
|
||||
use crate::repair::introspect::PulseSession;
|
||||
use crate::repair::plan::{
|
||||
Fingerprint, ModuleObservation, OwnerToken, Shape, classify, recorded_argument,
|
||||
};
|
||||
|
||||
/// PulseAudio's "no such index" — `PA_INVALID_INDEX`, `(uint32_t) -1`.
|
||||
///
|
||||
/// Every Pulse load callback reports failure by handing back this value rather
|
||||
/// than an index. We load through `pactl`, which reports failure by exiting
|
||||
/// non-zero instead, so seeing it on stdout would mean the tool printed a
|
||||
/// sentinel we must not mistake for a module: unloading it would be a request
|
||||
/// about something that cannot exist. Treated as a load whose outcome is unknown,
|
||||
/// never as a successful one.
|
||||
pub const PA_INVALID_INDEX: u32 = u32::MAX;
|
||||
|
||||
/// What an unresolved slot needs looked up before it can be trusted again.
|
||||
#[derive(Debug, Clone, PartialEq, Eq)]
|
||||
pub enum Reconcile {
|
||||
/// A load whose outcome is unknown: `pactl` was killed, timed out, or printed
|
||||
/// something we could not read as an index. The server may or may not have
|
||||
/// created the module, so it is looked for **by owner token** — the nonce is
|
||||
/// minted per load, so it names this attempt and no other.
|
||||
Load {
|
||||
shape: Shape,
|
||||
pid: u32,
|
||||
token: OwnerToken,
|
||||
},
|
||||
/// An unload whose outcome is unknown. The module is looked for by its full
|
||||
/// fingerprint: still present means the unload did not happen, absent means it
|
||||
/// did. An id alone would not do, because Pulse reuses module indices verbatim.
|
||||
Unload { fp: Fingerprint },
|
||||
}
|
||||
|
||||
/// One shape's slot in the module table.
|
||||
#[derive(Debug, Clone, PartialEq, Eq)]
|
||||
pub enum SlotState {
|
||||
/// Nothing loaded, and nothing outstanding.
|
||||
Vacant,
|
||||
/// A load is in flight under `permit`. The id is meaningless until it commits.
|
||||
Loading { token: OwnerToken, permit: u64 },
|
||||
/// A module we loaded and can name exactly.
|
||||
Loaded { fp: Fingerprint },
|
||||
/// An unload is in flight. The fingerprint is retained deliberately: an unload
|
||||
/// that times out must not leave the module unrecorded.
|
||||
Unloading { fp: Fingerprint },
|
||||
/// The slot's real state is unknown and must be resolved against the server.
|
||||
Ambiguous(Reconcile),
|
||||
/// Resolution found something we refuse to act on. Terminal.
|
||||
Poisoned { reason: String },
|
||||
}
|
||||
|
||||
impl SlotState {
|
||||
/// A short, stable label for logs and refusal messages.
|
||||
fn label(&self) -> &'static str {
|
||||
match self {
|
||||
SlotState::Vacant => "vacant",
|
||||
SlotState::Loading { .. } => "loading",
|
||||
SlotState::Loaded { .. } => "loaded",
|
||||
SlotState::Unloading { .. } => "unloading",
|
||||
SlotState::Ambiguous(_) => "ambiguous",
|
||||
SlotState::Poisoned { .. } => "poisoned",
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Why the ledger refused an operation.
|
||||
#[derive(Debug, Clone, PartialEq, Eq)]
|
||||
pub enum LedgerError {
|
||||
/// The slot was not free. Carries the state that refused, for the log line.
|
||||
Busy { shape: Shape, state: &'static str },
|
||||
/// The slot is poisoned and will not be used again this session.
|
||||
Poisoned { shape: Shape, reason: String },
|
||||
/// `pactl` reported `PA_INVALID_INDEX` where an index was expected.
|
||||
InvalidIndex { shape: Shape },
|
||||
}
|
||||
|
||||
impl std::fmt::Display for LedgerError {
|
||||
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
||||
match self {
|
||||
LedgerError::Busy { shape, state } => {
|
||||
write!(f, "the {} slot is {state}, not free", shape.label())
|
||||
}
|
||||
LedgerError::Poisoned { shape, reason } => {
|
||||
write!(f, "the {} slot is poisoned: {reason}", shape.label())
|
||||
}
|
||||
LedgerError::InvalidIndex { shape } => write!(
|
||||
f,
|
||||
"pactl reported PA_INVALID_INDEX for the {} module",
|
||||
shape.label()
|
||||
),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl std::error::Error for LedgerError {}
|
||||
|
||||
/// The outcome of an attempted unload, as the caller observed it.
|
||||
#[derive(Debug, Clone, PartialEq, Eq)]
|
||||
pub enum UnloadOutcome {
|
||||
/// The server acknowledged the unload.
|
||||
Confirmed,
|
||||
/// The unload may or may not have happened: the command failed, was killed,
|
||||
/// or its result could not be read.
|
||||
Uncertain(String),
|
||||
}
|
||||
|
||||
/// What a reconciliation found.
|
||||
#[derive(Debug, Clone, PartialEq, Eq)]
|
||||
pub enum Resolution {
|
||||
/// Exactly one module answers the question. Adopt it.
|
||||
Adopt(Fingerprint),
|
||||
/// No module answers it: the load never happened, or the unload did.
|
||||
Nothing,
|
||||
/// More than one module answers it. Refuse to guess — see [`Resolution`]'s use
|
||||
/// in [`ModuleLedger::apply`], which poisons the slot rather than picking.
|
||||
Conflict(usize),
|
||||
}
|
||||
|
||||
/// The ledger proper: one slot per [`Shape`], plus the permit counter.
|
||||
///
|
||||
/// Held behind an `Arc` because the permit needs a way back to it from [`Drop`],
|
||||
/// which is what makes a cancelled load impossible to lose.
|
||||
pub struct ModuleLedger {
|
||||
slots: Mutex<BTreeMap<Shape, SlotState>>,
|
||||
next_permit: AtomicU64,
|
||||
}
|
||||
|
||||
impl ModuleLedger {
|
||||
pub fn new() -> Arc<Self> {
|
||||
Arc::new(Self {
|
||||
slots: Mutex::new(BTreeMap::new()),
|
||||
next_permit: AtomicU64::new(1),
|
||||
})
|
||||
}
|
||||
|
||||
/// 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 {
|
||||
self.slots
|
||||
.lock()
|
||||
.unwrap()
|
||||
.get(&shape)
|
||||
.cloned()
|
||||
.unwrap_or(SlotState::Vacant)
|
||||
}
|
||||
|
||||
/// Take permission to load `shape`, moving the slot to
|
||||
/// [`SlotState::Loading`].
|
||||
///
|
||||
/// Refuses anything but a `Vacant` slot. In particular an *ambiguous* slot
|
||||
/// refuses, so an unresolved load blocks the next one until it is reconciled.
|
||||
pub fn begin_load(
|
||||
self: &Arc<Self>,
|
||||
shape: Shape,
|
||||
pid: u32,
|
||||
token: OwnerToken,
|
||||
) -> Result<LoadPermit, LedgerError> {
|
||||
let mut slots = self.slots.lock().unwrap();
|
||||
let current = slots.get(&shape).cloned().unwrap_or(SlotState::Vacant);
|
||||
match current {
|
||||
SlotState::Vacant => {}
|
||||
SlotState::Poisoned { reason } => {
|
||||
return Err(LedgerError::Poisoned { shape, reason });
|
||||
}
|
||||
other => {
|
||||
return Err(LedgerError::Busy {
|
||||
shape,
|
||||
state: other.label(),
|
||||
});
|
||||
}
|
||||
}
|
||||
let permit = self.next_permit.fetch_add(1, Ordering::Relaxed);
|
||||
slots.insert(
|
||||
shape,
|
||||
SlotState::Loading {
|
||||
token: token.clone(),
|
||||
permit,
|
||||
},
|
||||
);
|
||||
drop(slots);
|
||||
Ok(LoadPermit {
|
||||
ledger: Arc::clone(self),
|
||||
shape,
|
||||
pid,
|
||||
token,
|
||||
permit,
|
||||
settled: false,
|
||||
})
|
||||
}
|
||||
|
||||
/// Move a `Loaded` slot to `Unloading` and hand back what to unload.
|
||||
///
|
||||
/// `None` for any other state: there is nothing to unload, or the slot is not
|
||||
/// in a condition to be acted on.
|
||||
pub fn begin_unload(&self, shape: Shape) -> Option<Fingerprint> {
|
||||
let mut slots = self.slots.lock().unwrap();
|
||||
let SlotState::Loaded { fp } = slots.get(&shape).cloned()? else {
|
||||
return None;
|
||||
};
|
||||
slots.insert(shape, SlotState::Unloading { fp: fp.clone() });
|
||||
Some(fp)
|
||||
}
|
||||
|
||||
/// Record how an unload turned out. An uncertain one becomes ambiguous rather
|
||||
/// than being assumed done — the module is not forgotten either way.
|
||||
pub fn finish_unload(&self, shape: Shape, outcome: UnloadOutcome) {
|
||||
let mut slots = self.slots.lock().unwrap();
|
||||
let Some(SlotState::Unloading { fp }) = slots.get(&shape).cloned() else {
|
||||
return;
|
||||
};
|
||||
let next = match outcome {
|
||||
UnloadOutcome::Confirmed => SlotState::Vacant,
|
||||
UnloadOutcome::Uncertain(why) => {
|
||||
tracing::warn!(
|
||||
shape = fp.shape.label(),
|
||||
module = fp.id,
|
||||
"audio ledger: unload outcome uncertain ({why}); slot needs reconciling"
|
||||
);
|
||||
SlotState::Ambiguous(Reconcile::Unload { fp })
|
||||
}
|
||||
};
|
||||
slots.insert(shape, next);
|
||||
}
|
||||
|
||||
/// Every slot currently holding a module, in [`Shape`] declaration order —
|
||||
/// which is unload order: the loopbacks that reference the capture sink come
|
||||
/// before the sink itself.
|
||||
pub fn loaded(&self) -> Vec<Fingerprint> {
|
||||
self.slots
|
||||
.lock()
|
||||
.unwrap()
|
||||
.values()
|
||||
.filter_map(|state| match state {
|
||||
SlotState::Loaded { fp } => Some(fp.clone()),
|
||||
_ => None,
|
||||
})
|
||||
.collect()
|
||||
}
|
||||
|
||||
/// Every question outstanding against the server, in shape order.
|
||||
pub fn pending(&self) -> Vec<(Shape, Reconcile)> {
|
||||
self.slots
|
||||
.lock()
|
||||
.unwrap()
|
||||
.iter()
|
||||
.filter_map(|(shape, state)| match state {
|
||||
SlotState::Ambiguous(r) => Some((*shape, r.clone())),
|
||||
_ => None,
|
||||
})
|
||||
.collect()
|
||||
}
|
||||
|
||||
/// Is every slot in a state we can explain? False while anything is ambiguous
|
||||
/// or poisoned.
|
||||
pub fn is_settled(&self) -> bool {
|
||||
self.slots
|
||||
.lock()
|
||||
.unwrap()
|
||||
.values()
|
||||
.all(|state| !matches!(state, SlotState::Ambiguous(_) | SlotState::Poisoned { .. }))
|
||||
}
|
||||
|
||||
/// Apply a reconciliation result to an ambiguous slot.
|
||||
///
|
||||
/// A slot that is no longer ambiguous is left alone: the answer is stale, and
|
||||
/// overwriting a live state with it would be worse than ignoring it.
|
||||
pub fn apply(&self, shape: Shape, resolution: Resolution) {
|
||||
let mut slots = self.slots.lock().unwrap();
|
||||
if !matches!(slots.get(&shape), Some(SlotState::Ambiguous(_))) {
|
||||
return;
|
||||
}
|
||||
let next = match resolution {
|
||||
Resolution::Adopt(fp) => {
|
||||
tracing::info!(
|
||||
shape = shape.label(),
|
||||
module = fp.id,
|
||||
"audio ledger: reconciled — adopting the module the server actually has"
|
||||
);
|
||||
SlotState::Loaded { fp }
|
||||
}
|
||||
Resolution::Nothing => {
|
||||
tracing::info!(
|
||||
shape = shape.label(),
|
||||
"audio ledger: reconciled — the server has no such module"
|
||||
);
|
||||
SlotState::Vacant
|
||||
}
|
||||
Resolution::Conflict(n) => {
|
||||
let reason =
|
||||
format!("{n} modules answer to this slot's token; refusing to choose one");
|
||||
tracing::error!(shape = shape.label(), "audio ledger: {reason}");
|
||||
SlotState::Poisoned { reason }
|
||||
}
|
||||
};
|
||||
slots.insert(shape, next);
|
||||
}
|
||||
}
|
||||
|
||||
/// Permission to perform exactly one load, which must be settled by value.
|
||||
///
|
||||
/// Dropping it unsettled is not an error — it is the *reporting* path for a
|
||||
/// cancelled or panicking load, and it marks the slot ambiguous so the module the
|
||||
/// server may have created is looked for rather than forgotten.
|
||||
#[must_use = "an unsettled permit marks the slot ambiguous when dropped"]
|
||||
pub struct LoadPermit {
|
||||
ledger: Arc<ModuleLedger>,
|
||||
shape: Shape,
|
||||
pid: u32,
|
||||
token: OwnerToken,
|
||||
permit: u64,
|
||||
settled: bool,
|
||||
}
|
||||
|
||||
impl LoadPermit {
|
||||
/// Record that the server created the module at `index`.
|
||||
///
|
||||
/// Fails on [`PA_INVALID_INDEX`], leaving the permit unsettled so that
|
||||
/// dropping it marks the slot ambiguous — a sentinel where an index belongs
|
||||
/// means the load's outcome is precisely what we do not know.
|
||||
pub fn commit(mut self, index: u32) -> Result<Fingerprint, LedgerError> {
|
||||
if index == PA_INVALID_INDEX {
|
||||
return Err(LedgerError::InvalidIndex { shape: self.shape });
|
||||
}
|
||||
let fp = expected_fingerprint(self.shape, self.pid, &self.token, index);
|
||||
// A permit outlives its slot's `Loading` state only if something else has
|
||||
// already moved the slot on — in which case this answer is stale and the
|
||||
// live state wins.
|
||||
let mut slots = self.ledger.slots.lock().unwrap();
|
||||
match slots.get(&self.shape) {
|
||||
Some(SlotState::Loading { permit, .. }) if *permit == self.permit => {
|
||||
slots.insert(self.shape, SlotState::Loaded { fp: fp.clone() });
|
||||
}
|
||||
other => {
|
||||
let state = other.map(SlotState::label).unwrap_or("vacant");
|
||||
tracing::warn!(
|
||||
shape = self.shape.label(),
|
||||
module = index,
|
||||
"audio ledger: a load committed against a slot that is now {state}; \
|
||||
leaving the live state alone"
|
||||
);
|
||||
}
|
||||
}
|
||||
drop(slots);
|
||||
self.settled = true;
|
||||
Ok(fp)
|
||||
}
|
||||
|
||||
/// Record that the server definitively created nothing.
|
||||
///
|
||||
/// Only for a load that failed *cleanly* — `pactl` exiting non-zero of its own
|
||||
/// accord, having reported the server's refusal. A killed or timed-out load is
|
||||
/// not this: drop the permit instead and let the slot go ambiguous.
|
||||
pub fn abandon(mut self) {
|
||||
let mut slots = self.ledger.slots.lock().unwrap();
|
||||
if let Some(SlotState::Loading { permit, .. }) = slots.get(&self.shape)
|
||||
&& *permit == self.permit
|
||||
{
|
||||
slots.insert(self.shape, SlotState::Vacant);
|
||||
}
|
||||
drop(slots);
|
||||
self.settled = true;
|
||||
}
|
||||
}
|
||||
|
||||
impl Drop for LoadPermit {
|
||||
fn drop(&mut self) {
|
||||
if self.settled {
|
||||
return;
|
||||
}
|
||||
let mut slots = self.ledger.slots.lock().unwrap();
|
||||
// Only claim the slot if it is still *our* load. Anything else already
|
||||
// moved past this permit.
|
||||
if let Some(SlotState::Loading { permit, .. }) = slots.get(&self.shape)
|
||||
&& *permit == self.permit
|
||||
{
|
||||
tracing::warn!(
|
||||
shape = self.shape.label(),
|
||||
"audio ledger: a load was cancelled before its outcome was known; \
|
||||
the slot needs reconciling"
|
||||
);
|
||||
slots.insert(
|
||||
self.shape,
|
||||
SlotState::Ambiguous(Reconcile::Load {
|
||||
shape: self.shape,
|
||||
pid: self.pid,
|
||||
token: self.token.clone(),
|
||||
}),
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// The fingerprint a successful load of `shape` for `pid` under `token` must have.
|
||||
///
|
||||
/// Built from the shape's own renderer, exactly as [`classify`] rebuilds it from
|
||||
/// an observation, so an adopted module and a committed one are the same value.
|
||||
fn expected_fingerprint(shape: Shape, pid: u32, token: &OwnerToken, id: u32) -> Fingerprint {
|
||||
Fingerprint {
|
||||
id,
|
||||
module_name: shape.module_name().to_string(),
|
||||
args: recorded_argument(&shape.render_args(pid, Some(token))),
|
||||
pid,
|
||||
shape,
|
||||
owner: Some(token.clone()),
|
||||
}
|
||||
}
|
||||
|
||||
/// Answer one outstanding question against a snapshot of the module table.
|
||||
///
|
||||
/// Pure: the snapshot is the only input, so every branch is testable without a
|
||||
/// Pulse server.
|
||||
pub fn resolve(reconcile: &Reconcile, observations: &[ModuleObservation]) -> Resolution {
|
||||
let matches: Vec<Fingerprint> = match reconcile {
|
||||
// The nonce is minted per load, so token equality names this attempt and
|
||||
// no other — including a previous load of the same shape by the same pid.
|
||||
Reconcile::Load { shape, token, .. } => observations
|
||||
.iter()
|
||||
.filter_map(classify)
|
||||
.filter(|fp| fp.shape == *shape && fp.owner.as_ref() == Some(token))
|
||||
.collect(),
|
||||
// A full fingerprint match, not an id: Pulse reuses module indices
|
||||
// verbatim, so "something is at that index" is not "our module is".
|
||||
Reconcile::Unload { fp } => observations
|
||||
.iter()
|
||||
.filter(|obs| fp.still_matches(obs))
|
||||
.filter_map(classify)
|
||||
.collect(),
|
||||
};
|
||||
match matches.len() {
|
||||
0 => Resolution::Nothing,
|
||||
1 => Resolution::Adopt(matches.into_iter().next().expect("length checked")),
|
||||
n => Resolution::Conflict(n),
|
||||
}
|
||||
}
|
||||
|
||||
/// Resolve every outstanding question against one snapshot of the module table.
|
||||
///
|
||||
/// One listing answers all of them, so the slots are reconciled against a single
|
||||
/// server response rather than several that could disagree.
|
||||
///
|
||||
/// ⚠️ The session is deliberately short-lived — connect, list, disconnect — which
|
||||
/// is `--repair`'s pattern and not the long-lived host session round 19 sketched.
|
||||
/// `repair::introspect` documents why: on a request timeout the binding leaks the
|
||||
/// boxed callback until the context disconnects, which is bounded and harmless for
|
||||
/// a session that ends immediately, and is not acceptable for one held open for
|
||||
/// the life of a share. Reconciliation is rare and off the hot path, so paying a
|
||||
/// connection for it costs nothing that matters.
|
||||
pub async fn reconcile_pending(ledger: &Arc<ModuleLedger>) -> Result<()> {
|
||||
let pending = ledger.pending();
|
||||
if pending.is_empty() {
|
||||
return Ok(());
|
||||
}
|
||||
tracing::info!(
|
||||
n = pending.len(),
|
||||
"audio ledger: reconciling unresolved module slots against the server"
|
||||
);
|
||||
let observations = tokio::task::spawn_blocking(|| -> Result<Vec<ModuleObservation>> {
|
||||
let mut session = PulseSession::connect()?;
|
||||
session.list_modules()
|
||||
})
|
||||
.await
|
||||
.context("the Pulse listing task failed to run")?
|
||||
.context("could not list Pulse modules to reconcile the audio ledger")?;
|
||||
|
||||
for (shape, reconcile) in pending {
|
||||
ledger.apply(shape, resolve(&reconcile, &observations));
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
fn token(nonce: u64) -> OwnerToken {
|
||||
OwnerToken {
|
||||
machine: "abc123".to_string(),
|
||||
boot: "def456".to_string(),
|
||||
pid_ns: 4_026_531_836,
|
||||
nonce,
|
||||
}
|
||||
}
|
||||
|
||||
/// A module observation as the server would report it for one of our loads.
|
||||
fn observed(id: u32, shape: Shape, pid: u32, token: &OwnerToken) -> ModuleObservation {
|
||||
ModuleObservation::new(
|
||||
id,
|
||||
shape.module_name(),
|
||||
&recorded_argument(&shape.render_args(pid, Some(token))),
|
||||
)
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_committed_load_becomes_loaded() {
|
||||
let ledger = ModuleLedger::new();
|
||||
let permit = ledger
|
||||
.begin_load(Shape::LegacyCaptureSink, 42, token(7))
|
||||
.expect("a vacant slot issues a permit");
|
||||
assert!(matches!(
|
||||
ledger.state(Shape::LegacyCaptureSink),
|
||||
SlotState::Loading { .. }
|
||||
));
|
||||
let fp = permit.commit(9).expect("9 is a real index");
|
||||
assert_eq!(fp.id, 9);
|
||||
assert_eq!(
|
||||
ledger.state(Shape::LegacyCaptureSink),
|
||||
SlotState::Loaded { fp }
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn dropping_a_permit_unsettled_marks_the_slot_ambiguous() {
|
||||
// The orphan race in one test: a load that is cancelled between spawning
|
||||
// pactl and reading its id must leave the slot asking a question, never
|
||||
// looking empty.
|
||||
let ledger = ModuleLedger::new();
|
||||
let permit = ledger
|
||||
.begin_load(Shape::LoopbackOutOfCapture, 42, token(7))
|
||||
.expect("a vacant slot issues a permit");
|
||||
drop(permit);
|
||||
assert_eq!(
|
||||
ledger.state(Shape::LoopbackOutOfCapture),
|
||||
SlotState::Ambiguous(Reconcile::Load {
|
||||
shape: Shape::LoopbackOutOfCapture,
|
||||
pid: 42,
|
||||
token: token(7),
|
||||
}),
|
||||
"a cancelled load must be remembered as a question, not as a vacancy"
|
||||
);
|
||||
assert!(!ledger.is_settled());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn abandoning_a_cleanly_failed_load_returns_the_slot_to_vacant() {
|
||||
let ledger = ModuleLedger::new();
|
||||
ledger
|
||||
.begin_load(Shape::LoopbackIntoCapture, 42, token(7))
|
||||
.expect("a vacant slot issues a permit")
|
||||
.abandon();
|
||||
assert_eq!(
|
||||
ledger.state(Shape::LoopbackIntoCapture),
|
||||
SlotState::Vacant,
|
||||
"a load the server refused created nothing to reconcile"
|
||||
);
|
||||
assert!(ledger.is_settled());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_second_permit_is_refused_while_a_load_is_in_flight() {
|
||||
let ledger = ModuleLedger::new();
|
||||
let _first = ledger
|
||||
.begin_load(Shape::LegacyCaptureSink, 42, token(7))
|
||||
.expect("a vacant slot issues a permit");
|
||||
let second = ledger.begin_load(Shape::LegacyCaptureSink, 42, token(8));
|
||||
assert_eq!(
|
||||
second.err(),
|
||||
Some(LedgerError::Busy {
|
||||
shape: Shape::LegacyCaptureSink,
|
||||
state: "loading"
|
||||
}),
|
||||
"two loads for one slot must not both be permitted"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn an_ambiguous_slot_refuses_the_next_load_until_it_is_reconciled() {
|
||||
// Why this matters: two sinks may share a node.name, and pulsesrc attaches
|
||||
// to the older one — so loading over unresolved debris silently steals the
|
||||
// next session's capture.
|
||||
let ledger = ModuleLedger::new();
|
||||
drop(
|
||||
ledger
|
||||
.begin_load(Shape::LegacyCaptureSink, 42, token(7))
|
||||
.expect("a vacant slot issues a permit"),
|
||||
);
|
||||
assert_eq!(
|
||||
ledger
|
||||
.begin_load(Shape::LegacyCaptureSink, 42, token(8))
|
||||
.err(),
|
||||
Some(LedgerError::Busy {
|
||||
shape: Shape::LegacyCaptureSink,
|
||||
state: "ambiguous"
|
||||
})
|
||||
);
|
||||
|
||||
// Reconciling to "nothing was created" frees it again.
|
||||
ledger.apply(Shape::LegacyCaptureSink, Resolution::Nothing);
|
||||
assert!(
|
||||
ledger
|
||||
.begin_load(Shape::LegacyCaptureSink, 42, token(9))
|
||||
.is_ok()
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn an_invalid_index_is_not_adopted_and_leaves_a_question_behind() {
|
||||
let ledger = ModuleLedger::new();
|
||||
let permit = ledger
|
||||
.begin_load(Shape::LegacyCaptureSink, 42, token(7))
|
||||
.expect("a vacant slot issues a permit");
|
||||
assert_eq!(
|
||||
permit.commit(PA_INVALID_INDEX).err(),
|
||||
Some(LedgerError::InvalidIndex {
|
||||
shape: Shape::LegacyCaptureSink
|
||||
})
|
||||
);
|
||||
assert!(
|
||||
matches!(
|
||||
ledger.state(Shape::LegacyCaptureSink),
|
||||
SlotState::Ambiguous(_)
|
||||
),
|
||||
"a sentinel where an index belongs is exactly the unknown outcome"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn an_uncertain_unload_is_not_forgotten() {
|
||||
let ledger = ModuleLedger::new();
|
||||
let fp = ledger
|
||||
.begin_load(Shape::LoopbackIntoCapture, 42, token(7))
|
||||
.expect("a vacant slot issues a permit")
|
||||
.commit(3)
|
||||
.expect("3 is a real index");
|
||||
assert_eq!(
|
||||
ledger.begin_unload(Shape::LoopbackIntoCapture),
|
||||
Some(fp.clone())
|
||||
);
|
||||
ledger.finish_unload(
|
||||
Shape::LoopbackIntoCapture,
|
||||
UnloadOutcome::Uncertain("pactl was killed".to_string()),
|
||||
);
|
||||
assert_eq!(
|
||||
ledger.state(Shape::LoopbackIntoCapture),
|
||||
SlotState::Ambiguous(Reconcile::Unload { fp }),
|
||||
"an unload whose outcome is unknown must keep naming the module"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_confirmed_unload_empties_the_slot() {
|
||||
let ledger = ModuleLedger::new();
|
||||
ledger
|
||||
.begin_load(Shape::LoopbackIntoCapture, 42, token(7))
|
||||
.expect("a vacant slot issues a permit")
|
||||
.commit(3)
|
||||
.expect("3 is a real index");
|
||||
ledger.begin_unload(Shape::LoopbackIntoCapture);
|
||||
ledger.finish_unload(Shape::LoopbackIntoCapture, UnloadOutcome::Confirmed);
|
||||
assert_eq!(ledger.state(Shape::LoopbackIntoCapture), SlotState::Vacant);
|
||||
assert!(ledger.is_settled());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn begin_unload_only_acts_on_a_loaded_slot() {
|
||||
let ledger = ModuleLedger::new();
|
||||
assert_eq!(ledger.begin_unload(Shape::LegacyCaptureSink), None);
|
||||
let _permit = ledger
|
||||
.begin_load(Shape::LegacyCaptureSink, 42, token(7))
|
||||
.expect("a vacant slot issues a permit");
|
||||
assert_eq!(
|
||||
ledger.begin_unload(Shape::LegacyCaptureSink),
|
||||
None,
|
||||
"a load in flight has no id to unload yet"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn resolve_adopts_the_one_module_carrying_our_token() {
|
||||
let mine = token(7);
|
||||
let reconcile = Reconcile::Load {
|
||||
shape: Shape::LegacyCaptureSink,
|
||||
pid: 42,
|
||||
token: mine.clone(),
|
||||
};
|
||||
let observations = vec![
|
||||
// A previous load by the same pid and shape: same everything but the
|
||||
// per-load nonce, which is the whole reason the nonce exists.
|
||||
observed(4, Shape::LegacyCaptureSink, 42, &token(6)),
|
||||
observed(5, Shape::LegacyCaptureSink, 42, &mine),
|
||||
// Somebody else's module entirely.
|
||||
ModuleObservation::new(6, "module-null-sink", "sink_name=other"),
|
||||
];
|
||||
assert_eq!(
|
||||
resolve(&reconcile, &observations),
|
||||
Resolution::Adopt(expected_fingerprint(Shape::LegacyCaptureSink, 42, &mine, 5))
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn resolve_reports_nothing_when_the_server_never_created_it() {
|
||||
let reconcile = Reconcile::Load {
|
||||
shape: Shape::LegacyCaptureSink,
|
||||
pid: 42,
|
||||
token: token(7),
|
||||
};
|
||||
let observations = vec![observed(4, Shape::LegacyCaptureSink, 42, &token(6))];
|
||||
assert_eq!(resolve(&reconcile, &observations), Resolution::Nothing);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn resolve_fails_closed_when_more_than_one_module_answers() {
|
||||
let mine = token(7);
|
||||
let reconcile = Reconcile::Load {
|
||||
shape: Shape::LegacyCaptureSink,
|
||||
pid: 42,
|
||||
token: mine.clone(),
|
||||
};
|
||||
// Two modules carrying the same token should be impossible. If it ever
|
||||
// happens, guessing which to keep is how a live sink gets unloaded.
|
||||
let observations = vec![
|
||||
observed(4, Shape::LegacyCaptureSink, 42, &mine),
|
||||
observed(5, Shape::LegacyCaptureSink, 42, &mine),
|
||||
];
|
||||
assert_eq!(resolve(&reconcile, &observations), Resolution::Conflict(2));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_conflict_poisons_the_slot_and_it_stays_poisoned() {
|
||||
let ledger = ModuleLedger::new();
|
||||
drop(
|
||||
ledger
|
||||
.begin_load(Shape::LegacyCaptureSink, 42, token(7))
|
||||
.expect("a vacant slot issues a permit"),
|
||||
);
|
||||
ledger.apply(Shape::LegacyCaptureSink, Resolution::Conflict(2));
|
||||
let SlotState::Poisoned { reason } = ledger.state(Shape::LegacyCaptureSink) else {
|
||||
panic!("a conflict must poison the slot");
|
||||
};
|
||||
assert!(!ledger.is_settled());
|
||||
assert_eq!(
|
||||
ledger
|
||||
.begin_load(Shape::LegacyCaptureSink, 42, token(8))
|
||||
.err(),
|
||||
Some(LedgerError::Poisoned {
|
||||
shape: Shape::LegacyCaptureSink,
|
||||
reason
|
||||
}),
|
||||
"poisoning is terminal for the session"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn resolve_finds_a_module_an_uncertain_unload_left_behind() {
|
||||
let mine = token(7);
|
||||
let fp = expected_fingerprint(Shape::LoopbackIntoCapture, 42, &mine, 5);
|
||||
let observations = vec![observed(5, Shape::LoopbackIntoCapture, 42, &mine)];
|
||||
assert_eq!(
|
||||
resolve(&Reconcile::Unload { fp: fp.clone() }, &observations),
|
||||
Resolution::Adopt(fp.clone()),
|
||||
"still present means the unload did not happen"
|
||||
);
|
||||
assert_eq!(
|
||||
resolve(&Reconcile::Unload { fp }, &[]),
|
||||
Resolution::Nothing,
|
||||
"absent means it did"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn an_unload_reconcile_ignores_a_stranger_at_the_same_index() {
|
||||
// Pulse reuses module indices verbatim, so "something is at index 5" must
|
||||
// not be read as "our module is still at index 5".
|
||||
//
|
||||
// The stranger has to be a module that *classifies* — another pixelpass
|
||||
// host's canonical loopback — or this gate is vacuous: a comparator using
|
||||
// the id alone would still return `Nothing` for junk, because junk is
|
||||
// discarded by `classify` regardless of how the match was made.
|
||||
let fp = expected_fingerprint(Shape::LoopbackIntoCapture, 42, &token(7), 5);
|
||||
let another_hosts = observed(5, Shape::LoopbackIntoCapture, 99, &token(3));
|
||||
assert_eq!(
|
||||
resolve(
|
||||
&Reconcile::Unload { fp: fp.clone() },
|
||||
std::slice::from_ref(&another_hosts)
|
||||
),
|
||||
Resolution::Nothing,
|
||||
"another host's module at our old index is not our module"
|
||||
);
|
||||
// And plain junk at that index is ignored too.
|
||||
let junk = ModuleObservation::new(5, "module-loopback", "source=some_mic sink=theirs");
|
||||
assert_eq!(
|
||||
resolve(&Reconcile::Unload { fp }, std::slice::from_ref(&junk)),
|
||||
Resolution::Nothing
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_stale_answer_does_not_overwrite_a_live_slot() {
|
||||
let ledger = ModuleLedger::new();
|
||||
// Nothing is ambiguous, so an answer arriving late is meaningless.
|
||||
let fp = ledger
|
||||
.begin_load(Shape::LegacyCaptureSink, 42, token(7))
|
||||
.expect("a vacant slot issues a permit")
|
||||
.commit(3)
|
||||
.expect("3 is a real index");
|
||||
ledger.apply(Shape::LegacyCaptureSink, Resolution::Nothing);
|
||||
assert_eq!(
|
||||
ledger.state(Shape::LegacyCaptureSink),
|
||||
SlotState::Loaded { fp },
|
||||
"a live state must win over a stale reconciliation"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn loaded_lists_modules_in_shape_declaration_order() {
|
||||
// Declaration order is unload order: the loopbacks that reference the
|
||||
// capture sink must come before the sink itself.
|
||||
let ledger = ModuleLedger::new();
|
||||
for (shape, id) in [
|
||||
(Shape::LegacyCaptureSink, 1),
|
||||
(Shape::LoopbackIntoCapture, 2),
|
||||
(Shape::LoopbackOutOfCapture, 3),
|
||||
] {
|
||||
ledger
|
||||
.begin_load(shape, 42, token(u64::from(id)))
|
||||
.expect("a vacant slot issues a permit")
|
||||
.commit(id)
|
||||
.expect("a real index");
|
||||
}
|
||||
let order: Vec<Shape> = ledger.loaded().into_iter().map(|fp| fp.shape).collect();
|
||||
assert_eq!(order, crate::repair::plan::ALL_SHAPES.to_vec());
|
||||
assert_eq!(
|
||||
order.last(),
|
||||
Some(&Shape::LegacyCaptureSink),
|
||||
"the sink must be unloaded after everything that references it"
|
||||
);
|
||||
}
|
||||
}
|
||||
@@ -2,6 +2,7 @@ pub mod aec;
|
||||
pub mod audio;
|
||||
pub mod audit;
|
||||
mod capture;
|
||||
pub mod ledger;
|
||||
mod observer;
|
||||
mod pipeline;
|
||||
mod quality;
|
||||
|
||||
@@ -47,7 +47,7 @@ impl CaptureHandle {
|
||||
let _ = child.start_kill();
|
||||
}
|
||||
if let Some(audio) = self.audio.take() {
|
||||
audio.shutdown();
|
||||
audio.shutdown().await;
|
||||
}
|
||||
if let Some(serve) = self.serve.take() {
|
||||
serve.shutdown().await;
|
||||
|
||||
Reference in New Issue
Block a user