diff --git a/src/host/audio.rs b/src/host/audio.rs index a2d3030..3b4c1e0 100644 --- a/src/host/audio.rs +++ b/src/host/audio.rs @@ -33,28 +33,41 @@ use anyhow::{Context, Result, bail}; use std::cell::RefCell; use std::collections::BTreeMap; -use std::process::Command; +use std::io::{self, Read}; +use std::process::{Child, Command, ExitStatus, Stdio}; use std::rc::Rc; -use std::sync::{Arc, Mutex}; +use std::sync::Arc; use std::thread::JoinHandle; +use std::time::{Duration, Instant}; 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 `pactl load-module` worker may run before it is killed and reaped. +/// +/// 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. Unloads use PulseSession's independently bounded native +/// connect/list/unload requests instead of a second pactl connection. +const PACTL_BUDGET: Duration = Duration::from_secs(5); +const PACTL_REAP_BUDGET: Duration = Duration::from_secs(1); /// 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, - /// Shared with the event task so it can `take()` and unload on the - /// first successful route. `Routing::shutdown` unloads whatever - /// remains. - loopback_module: Arc>>, - /// 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>>, + /// 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`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, sink_name: String, stream_router: Option, event_task: Option>, @@ -66,13 +79,28 @@ impl Routing { pub async fn start(opts: &HostOpts) -> Result { let pid = std::process::id(); let sink_name = repair_plan::sink_name_for(pid); + let ledger = ModuleLedger::new(); + // Construct the owner before the first mutation. Any error or cancellation + // below now drops a real `Routing`, whose backstop closes, quiesces, + // reconciles, and unloads this ledger. Previously the owner did not exist + // until both initial modules had loaded, so constructor failure leaked + // everything loaded up to that point. + let mut routing = Self { + ledger: Arc::clone(&ledger), + sink_name: sink_name.clone(), + stream_router: None, + event_task: None, + }; // 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,51 +112,32 @@ 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), - sink_name: sink_name.clone(), - stream_router: None, - event_task: None, - }; - 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" + ); + let mirror_absent = + 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,19 +145,14 @@ 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:#}" - ), - } + if mirror_absent { + ensure_loaded(&ledger_for_task, Shape::LoopbackOutOfCapture, pid) + .await; + } else { + tracing::warn!( + "audio routing: default-sink mirror absence was not confirmed; \ + refusing to load the inverse local monitor" + ); } // Tell the front-end the chosen app's audio is live. output::emit(output::Event::AppAudio { @@ -164,13 +168,8 @@ 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); - } + let local_monitor_absent = + 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 @@ -181,25 +180,21 @@ impl Routing { ); continue; } - // 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() { + if !local_monitor_absent { + tracing::warn!( + "audio routing: local-monitor absence was not confirmed; \ + refusing to restore the inverse default-sink mirror" + ); continue; } + // Best-effort mode: restore the default-sink loopback + // so the viewer hears system audio again instead of + // 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 +220,92 @@ 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) { + // Closing is synchronous and happens first: after this point the event + // task cannot register another mutation even if it receives one last + // router event while shutdown is in progress. + self.ledger.close(); + 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; + } + } + + // A cancelled `spawn_blocking` await detaches its worker. Every worker is + // registered before it can be spawned, so this is a real ordering + // boundary: reconciliation cannot overtake late module creation/removal. + let ledger_for_wait = Arc::clone(&self.ledger); + if let Err(e) = tokio::task::spawn_blocking(move || { + ledger_for_wait.wait_for_operations(); + }) + .await + { + tracing::warn!("audio routing: module-operation wait task failed: {e}"); + } + cleanup_modules(&self.ledger).await; + + if !self.ledger.is_clean() { + tracing::warn!( + settled = self.ledger.is_settled(), + "audio routing: some audio modules could not be removed safely; \ + `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) { + self.ledger.close(); 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); + self.ledger.close_and_wait(); + cleanup_modules_blocking(&self.ledger); + if !self.ledger.is_clean() { + tracing::warn!( + settled = self.ledger.is_settled(), + "audio routing: torn down with modules that could not be removed safely; \ + run `pixelpass --repair` to clean up anything left behind" + ); } - // 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 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,53 +419,362 @@ fn owner_token(pid: u32) -> Result { /// `--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 { +async fn load_module(ledger: &Arc, shape: Shape, pid: u32) -> Result { let owner = owner_token(pid).context("could not build an audio ownership token")?; - let output = Command::new("pactl") - .arg("load-module") - .arg(shape.module_name()) - .args(shape.render_args(pid, Some(&owner))) - .output() - .context("failed to run pactl load-module")?; - if !output.status.success() { - bail!( - "pactl load-module failed: {}", - String::from_utf8_lossy(&output.stderr).trim() + let permit = ledger + .begin_load(shape, pid, owner.clone()) + .map_err(anyhow::Error::new) + .with_context(|| format!("cannot load the {} module", shape.label()))?; + let args = shape.render_args(pid, Some(&owner)); + + // The affine permit moves into the blocking worker. Dropping this await does + // not cancel `spawn_blocking`; the worker remains registered, owns and reaps + // its child, and settles the slot before teardown's quiescence barrier opens. + tokio::task::spawn_blocking(move || -> Result { + let output = permit.with_server_operation(|| { + let mut command = Command::new("pactl"); + command + .arg("load-module") + .arg(shape.module_name()) + .args(args); + bounded_output(&mut command, PACTL_BUDGET) + }); + let output = match output { + Ok(output) => output, + Err(e) => { + // Whether the server was reached is unknown. Leaving the permit + // unsettled makes its Drop create a reconciliation question. + drop(permit); + return Err(e).context("failed to run pactl load-module"); + } + }; + + if output.timed_out { + drop(permit); + bail!("pactl load-module did not finish within {PACTL_BUDGET:?}"); + } + if !output.status.success() { + let stderr = String::from_utf8_lossy(&output.stderr).trim().to_string(); + if output.status.code().is_some() { + // pactl exited normally and reported the server's refusal. + permit.abandon(); + } else { + drop(permit); + } + bail!("pactl load-module failed: {stderr}"); + } + + let id_str = String::from_utf8_lossy(&output.stdout).trim().to_string(); + // Genuinely 32-bit: this is a Pulse module index, not object.serial. + let Ok(index) = id_str.parse::() else { + 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" ); - } - let id_str = String::from_utf8(output.stdout) - .context("pactl returned non-UTF-8")? - .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::() - .with_context(|| format!("pactl returned unexpected module ID: {id_str:?}")) + Ok(fp) + }) + .await + .context("the pactl load worker failed")? } -fn unload_module(id: u32) { - let result = Command::new("pactl") - .arg("unload-module") - .arg(id.to_string()) - .output(); - match result { - Ok(output) if output.status.success() => { - tracing::info!(module = id, "audio routing: unloaded pactl module"); - } - Ok(output) => { - tracing::warn!( - module = id, - stderr = %String::from_utf8_lossy(&output.stderr).trim(), - "audio routing: pactl unload-module exited non-zero" - ); - } +/// 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, shape: Shape, pid: u32) { + let Err(e) = load_module(ledger, shape, pid).await else { + return; + }; + match e.downcast_ref::() { + 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, shape: Shape) -> bool { + unload_module_inner(ledger, shape, false).await +} + +async fn unload_module_inner(ledger: &Arc, shape: Shape, cleanup: bool) -> bool { + let begun = if cleanup { + ledger.begin_cleanup_unload(shape) + } else { + ledger.begin_unload(shape) + }; + let permit = match begun { + Ok(Some(permit)) => permit, + Ok(None) => return true, Err(e) => { tracing::warn!( - module = id, - "audio routing: failed to run pactl unload-module: {e}" + shape = shape.label(), + "audio routing: refusing module unload: {e}" ); + return false; + } + }; + match tokio::task::spawn_blocking(move || finish_verified_unload(permit)).await { + Ok(confirmed_absent) => confirmed_absent, + Err(e) => { + // A panicking worker drops its affine permit and therefore leaves an + // unload reconciliation question behind. + tracing::warn!( + shape = shape.label(), + "audio routing: unload worker failed: {e}" + ); + false + } + } +} + +fn finish_verified_unload(permit: ledger::UnloadPermit) -> bool { + let fp = permit.fingerprint().clone(); + let result = permit.with_server_operation(|| verified_unload(&fp)); + match result { + Ok(()) => { + permit.finish(UnloadOutcome::Confirmed); + true + } + Err(e) => { + permit.finish(UnloadOutcome::Uncertain(format!("{e:#}"))); + false + } + } +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum UnloadPresence { + Exact, + Absent, + Replaced, +} + +fn unload_presence(fp: &Fingerprint, current: &[repair_plan::ModuleObservation]) -> UnloadPresence { + match current.iter().find(|module| module.id == fp.id) { + None => UnloadPresence::Absent, + Some(observed) if fp.still_matches(observed) => UnloadPresence::Exact, + Some(_) => UnloadPresence::Replaced, + } +} + +/// Re-list and unload through one verified-local Pulse connection. The exact +/// fingerprint is checked immediately before destruction; if the index vanished +/// or was reused, our module is already absent and the replacement is left alone. +fn verified_unload(fp: &Fingerprint) -> Result<()> { + let mut session = crate::repair::introspect::PulseSession::connect() + .context("could not connect to verify a module unload")?; + let current = session + .list_modules() + .context("could not list modules immediately before unload")?; + match unload_presence(fp, ¤t) { + UnloadPresence::Exact => {} + UnloadPresence::Absent => { + tracing::info!( + module = fp.id, + shape = fp.shape.label(), + "audio routing: tracked module was already absent" + ); + return Ok(()); + } + UnloadPresence::Replaced => { + tracing::warn!( + module = fp.id, + shape = fp.shape.label(), + "audio routing: module index was reused; leaving the replacement alone" + ); + return Ok(()); + } + } + session + .unload_module(fp.id) + .with_context(|| format!("the server did not confirm unloading module #{}", fp.id))?; + tracing::info!( + module = fp.id, + shape = fp.shape.label(), + "audio routing: unloaded verified Pulse module" + ); + Ok(()) +} + +/// One bounded child result. On deadline the child is killed and synchronously +/// reaped before this returns, so server reconciliation cannot overtake a late +/// `pactl` request merely because its async waiter was cancelled. +struct BoundedOutput { + status: ExitStatus, + stdout: Vec, + stderr: Vec, + timed_out: bool, +} + +/// Ensures every early-return and panic after spawn kills and reaps the child. +/// The normal path marks it reaped after `try_wait`/`wait` obtained the status. +struct ReapedChild { + child: Child, + reaped: bool, +} + +impl Drop for ReapedChild { + fn drop(&mut self) { + if self.reaped { + return; + } + let _ = self.child.kill(); + match self.reap_within(PACTL_REAP_BUDGET) { + Ok(Some(_)) => {} + Ok(None) => tracing::warn!( + "audio routing: killed pactl child was not reaped within {PACTL_REAP_BUDGET:?}" + ), + Err(e) => tracing::warn!("audio routing: could not reap killed pactl child: {e}"), + } + } +} + +impl ReapedChild { + fn reap_within(&mut self, budget: Duration) -> io::Result> { + let deadline = Instant::now() + budget; + loop { + if let Some(status) = self.child.try_wait()? { + self.reaped = true; + return Ok(Some(status)); + } + if Instant::now() >= deadline { + return Ok(None); + } + std::thread::sleep(Duration::from_millis(5)); + } + } +} + +fn bounded_output(command: &mut Command, budget: Duration) -> io::Result { + command.stdout(Stdio::piped()).stderr(Stdio::piped()); + let mut child = ReapedChild { + child: command.spawn()?, + reaped: false, + }; + let stdout = child + .child + .stdout + .take() + .ok_or_else(|| io::Error::other("pactl stdout was not piped"))?; + let stderr = child + .child + .stderr + .take() + .ok_or_else(|| io::Error::other("pactl stderr was not piped"))?; + let stdout_reader = std::thread::Builder::new() + .name("pixelpass-pactl-stdout".to_string()) + .spawn(move || { + let mut bytes = Vec::new(); + let mut stdout = stdout; + stdout.read_to_end(&mut bytes).map(|_| bytes) + })?; + let stderr_reader = std::thread::Builder::new() + .name("pixelpass-pactl-stderr".to_string()) + .spawn(move || { + let mut bytes = Vec::new(); + let mut stderr = stderr; + stderr.read_to_end(&mut bytes).map(|_| bytes) + })?; + + let deadline = Instant::now() + budget; + let (status, timed_out) = loop { + if let Some(status) = child.child.try_wait()? { + child.reaped = true; + break (status, false); + } + if Instant::now() >= deadline { + let _ = child.child.kill(); + let Some(status) = child.reap_within(PACTL_REAP_BUDGET)? else { + return Err(io::Error::new( + io::ErrorKind::TimedOut, + format!("pactl did not exit within {PACTL_REAP_BUDGET:?} after SIGKILL"), + )); + }; + break (status, true); + } + std::thread::sleep(Duration::from_millis(5)); + }; + let stdout = stdout_reader + .join() + .map_err(|_| io::Error::other("pactl stdout reader panicked"))??; + let stderr = stderr_reader + .join() + .map_err(|_| io::Error::other("pactl stderr reader panicked"))??; + Ok(BoundedOutput { + status, + stdout, + stderr, + timed_out, + }) +} + +/// Async teardown: keep reconciling and unloading while a pass changes state. +/// This replaces the arbitrary two-round count. With loads closed, the state +/// graph is monotonic except for `Ambiguous(Unload) -> Loaded -> Ambiguous` when +/// the same unload remains uncertain; that produces an identical snapshot and +/// stops here for `--repair` rather than spinning. +async fn cleanup_modules(ledger: &Arc) { + loop { + let before = ledger.snapshot(); + if ledger.is_clean() { + break; + } + if let Err(e) = ledger::reconcile_pending(ledger).await { + tracing::warn!("audio routing: could not reconcile the module ledger: {e:#}"); + } + for fp in ledger.loaded() { + unload_module_inner(ledger, fp.shape, true).await; + } + if ledger.is_clean() || ledger.snapshot() == before { + break; + } + } +} + +/// Synchronous teardown backstop. PulseSession bounds every connect/list/unload +/// request, and `close_and_wait` has already drained any registered pactl worker. +fn cleanup_modules_blocking(ledger: &Arc) { + loop { + let before = ledger.snapshot(); + if ledger.is_clean() { + break; + } + if let Err(e) = ledger::reconcile_pending_blocking(ledger) { + tracing::warn!("audio routing: could not reconcile the module ledger: {e:#}"); + } + for fp in ledger.loaded() { + let permit = match ledger.begin_cleanup_unload(fp.shape) { + Ok(Some(permit)) => permit, + Ok(None) => continue, + Err(e) => { + tracing::warn!( + shape = fp.shape.label(), + "audio routing: refusing blocking module unload: {e}" + ); + continue; + } + }; + finish_verified_unload(permit); + } + if ledger.is_clean() || ledger.snapshot() == before { + break; } } } @@ -718,6 +1072,175 @@ fn try_flush( #[cfg(test)] mod tests { use super::*; + use crate::host::ledger::SlotState; + 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() + } + + #[test] + fn normal_unload_requires_the_full_fingerprint_not_only_the_index() { + fn observation(id: u32, pid: u32, nonce: u64) -> ModuleObservation { + let token = repair_plan::OwnerToken { + machine: "abc123".to_string(), + boot: "def456".to_string(), + pid_ns: 4_026_531_836, + nonce, + }; + ModuleObservation::new( + id, + Shape::LoopbackIntoCapture.module_name(), + &repair_plan::recorded_argument( + &Shape::LoopbackIntoCapture.render_args(pid, Some(&token)), + ), + ) + } + + let ours = observation(5, 42, 7); + let fp = classify(&ours).expect("the fixture is canonical"); + assert_eq!( + unload_presence(&fp, std::slice::from_ref(&ours)), + UnloadPresence::Exact + ); + assert_eq!(unload_presence(&fp, &[]), UnloadPresence::Absent); + + // Same live index, but another perfectly canonical host module. This is + // the non-vacuous reuse case: an id-only normal unload would destroy it. + let replacement = observation(5, 99, 8); + assert_eq!( + unload_presence(&fp, std::slice::from_ref(&replacement)), + UnloadPresence::Replaced + ); + } + + /// 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; + }); + // Wait until the affine permit is registered. A single yield is not a + // scheduling guarantee and made the old version of this gate capable of + // aborting before the task had started. + tokio::time::timeout(PACTL_BUDGET, async { + while matches!(ledger.state(Shape::LegacyCaptureSink), SlotState::Vacant) { + tokio::task::yield_now().await; + } + }) + .await + .expect("the load registers its operation"); + task.abort(); + let _ = task.await; + + // Cancellation detaches `spawn_blocking`; closing plus this registered- + // operation wait is the ordering boundary that prevents reconciliation + // from overtaking the worker's late server mutation. + ledger.close(); + let ledger_for_wait = Arc::clone(&ledger); + tokio::task::spawn_blocking(move || ledger_for_wait.wait_for_operations()) + .await + .expect("the operation wait runs"); + + // The worker either committed a known module or left a reconciliation + // question. Both are correct; vacancy here would mean the live operation + // disappeared from the ledger. + assert_eq!( + ledger.pending().len() + ledger.loaded().len(), + 1, + "the cancelled load must retain exactly one tracked outcome" + ); + + cleanup_modules(&ledger).await; + + assert!( + ledger.is_clean(), + "every tracked module must be removed, not merely explained" + ); + assert_eq!( + module_snapshot(), + before, + "a cancelled load must leave nothing behind" + ); + } #[test] fn object_serial_parses_past_u32() { diff --git a/src/host/ledger.rs b/src/host/ledger.rs new file mode 100644 index 0000000..c3b0980 --- /dev/null +++ b/src/host/ledger.rs @@ -0,0 +1,1349 @@ +//! 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`s, two of them shared with the +//! event task behind an `Arc>`. 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` 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. The permit moves +//! into a registered blocking worker before any await can detach that work; either +//! the worker settles it, or dropping the worker marks the question ambiguous. +//! Teardown waits for all such permits before reconciling. 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=.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::{AtomicBool, AtomicU64, Ordering}; +use std::sync::{Arc, Condvar, 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, permit: u64 }, + /// 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 }, + /// Teardown has begun, so event-driven work may no longer start. + Closed { shape: Shape }, + /// The two inverse loopbacks must never coexist. + Incompatible { + shape: Shape, + occupied: Shape, + state: &'static str, + }, + /// The capture sink cannot be removed while a loopback may still reference it. + Referenced { + shape: Shape, + dependency: Shape, + state: &'static str, + }, +} + +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() + ), + LedgerError::Closed { shape } => write!( + f, + "the module ledger is closing; refusing new work on the {} slot", + shape.label() + ), + LedgerError::Incompatible { + shape, + occupied, + state, + } => write!( + f, + "cannot load the {} while the incompatible {} slot is {state}", + shape.label(), + occupied.label() + ), + LedgerError::Referenced { + shape, + dependency, + state, + } => write!( + f, + "cannot unload the {} while the dependent {} slot is {state}", + shape.label(), + dependency.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>, + next_permit: AtomicU64, + closed: AtomicBool, + operations: OperationCoordinator, +} + +/// Registers module mutations before they can be detached from their caller. +/// +/// `spawn_blocking` work continues when the awaiting future is cancelled. The +/// count is therefore registered synchronously, while the slot lock is held, and +/// is released only by the affine permit's `Drop`. Teardown closes the ledger and +/// waits on this condition variable before it asks the server what exists. +struct OperationCoordinator { + active: Mutex, + idle: Condvar, + server: Mutex<()>, +} + +impl OperationCoordinator { + fn new() -> Self { + Self { + active: Mutex::new(0), + idle: Condvar::new(), + server: Mutex::new(()), + } + } + + fn register(&self) { + let mut active = self.active.lock().unwrap_or_else(|e| e.into_inner()); + *active = active + .checked_add(1) + .expect("module operation count overflow"); + } + + fn finish(&self) { + let mut active = self.active.lock().unwrap_or_else(|e| e.into_inner()); + *active = active + .checked_sub(1) + .expect("module operation count underflow"); + if *active == 0 { + self.idle.notify_all(); + } + } + + fn wait_idle(&self) { + let mut active = self.active.lock().unwrap_or_else(|e| e.into_inner()); + while *active != 0 { + active = self.idle.wait(active).unwrap_or_else(|e| e.into_inner()); + } + } + + fn run(&self, operation: impl FnOnce() -> T) -> T { + let _serial = self.server.lock().unwrap_or_else(|e| e.into_inner()); + operation() + } +} + +impl ModuleLedger { + pub fn new() -> Arc { + Arc::new(Self { + slots: Mutex::new(BTreeMap::new()), + next_permit: AtomicU64::new(1), + closed: AtomicBool::new(false), + operations: OperationCoordinator::new(), + }) + } + + /// Stop event-driven mutations and wait until every already-registered load + /// or unload has settled its affine permit. + /// + /// The slots-lock hand-off closes the only race that matters: an operation + /// that saw `closed == false` has registered itself before this method can + /// pass the hand-off and begin waiting. + pub(super) fn close(&self) { + self.closed.store(true, Ordering::SeqCst); + drop(self.slots.lock().unwrap_or_else(|e| e.into_inner())); + } + + pub(super) fn wait_for_operations(&self) { + self.operations.wait_idle(); + } + + pub(super) fn close_and_wait(&self) { + self.close(); + self.wait_for_operations(); + } + + /// Serialize one server mutation or observation with every other such action. + pub(super) fn with_server_operation(&self, operation: impl FnOnce() -> T) -> T { + self.operations.run(operation) + } + + /// 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, + shape: Shape, + pid: u32, + token: OwnerToken, + ) -> Result { + let mut slots = self.slots.lock().unwrap(); + if self.closed.load(Ordering::SeqCst) { + return Err(LedgerError::Closed { shape }); + } + 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(), + }); + } + } + if let Some(occupied) = inverse_loopback(shape) { + let state = slots.get(&occupied).cloned().unwrap_or(SlotState::Vacant); + if !matches!(state, SlotState::Vacant) { + return Err(LedgerError::Incompatible { + shape, + occupied, + state: state.label(), + }); + } + } + let permit = self.next_permit.fetch_add(1, Ordering::Relaxed); + self.operations.register(); + 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. + /// + /// `Ok(None)` means confirmed vacancy. Busy, poisoned, closed, and still- + /// referenced states are errors so callers cannot mistake uncertainty for + /// absence and load an inverse loopback or destroy the capture sink. + pub fn begin_unload( + self: &Arc, + shape: Shape, + ) -> Result, LedgerError> { + self.begin_unload_inner(shape, false) + } + + /// Teardown-only form of [`ModuleLedger::begin_unload`]. New event work is + /// closed by then, but cleanup must still be able to remove tracked modules. + pub(super) fn begin_cleanup_unload( + self: &Arc, + shape: Shape, + ) -> Result, LedgerError> { + self.begin_unload_inner(shape, true) + } + + fn begin_unload_inner( + self: &Arc, + shape: Shape, + cleanup: bool, + ) -> Result, LedgerError> { + let mut slots = self.slots.lock().unwrap(); + if !cleanup && self.closed.load(Ordering::SeqCst) { + return Err(LedgerError::Closed { shape }); + } + let current = slots.get(&shape).cloned().unwrap_or(SlotState::Vacant); + let fp = match current { + SlotState::Vacant => return Ok(None), + SlotState::Loaded { fp } => fp, + SlotState::Poisoned { reason } => { + return Err(LedgerError::Poisoned { shape, reason }); + } + other => { + return Err(LedgerError::Busy { + shape, + state: other.label(), + }); + } + }; + if shape == Shape::LegacyCaptureSink { + for dependency in [Shape::LoopbackIntoCapture, Shape::LoopbackOutOfCapture] { + let state = slots.get(&dependency).cloned().unwrap_or(SlotState::Vacant); + if !matches!(state, SlotState::Vacant) { + return Err(LedgerError::Referenced { + shape, + dependency, + state: state.label(), + }); + } + } + } + let permit = self.next_permit.fetch_add(1, Ordering::Relaxed); + self.operations.register(); + slots.insert( + shape, + SlotState::Unloading { + fp: fp.clone(), + permit, + }, + ); + Ok(Some(UnloadPermit { + ledger: Arc::clone(self), + shape, + fp, + permit, + settled: false, + })) + } + + /// 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 { + 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 an operation is in + /// flight, its outcome is ambiguous, or a conflict poisoned the slot. + pub fn is_settled(&self) -> bool { + self.slots + .lock() + .unwrap() + .values() + .all(|state| matches!(state, SlotState::Vacant | SlotState::Loaded { .. })) + } + + /// Has teardown removed every module rather than merely accounted for it? + pub fn is_clean(&self) -> bool { + self.slots + .lock() + .unwrap() + .values() + .all(|state| matches!(state, SlotState::Vacant)) + } + + /// A stable snapshot used to stop cleanup when another pass made no progress. + pub(super) fn snapshot(&self) -> BTreeMap { + self.slots.lock().unwrap().clone() + } + + /// 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); + } +} + +fn inverse_loopback(shape: Shape) -> Option { + match shape { + Shape::LoopbackIntoCapture => Some(Shape::LoopbackOutOfCapture), + Shape::LoopbackOutOfCapture => Some(Shape::LoopbackIntoCapture), + Shape::LegacyCaptureSink => None, + } +} + +/// 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, + shape: Shape, + pid: u32, + token: OwnerToken, + permit: u64, + settled: bool, +} + +impl LoadPermit { + /// Run the server-facing part of this load in the same serialized domain as + /// unload verification and reconciliation. + pub fn with_server_operation(&self, operation: impl FnOnce() -> T) -> T { + self.ledger.with_server_operation(operation) + } + + /// 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 { + 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 { + 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(), + }), + ); + } + } + self.ledger.operations.finish(); + } +} + +/// Permission to perform exactly one unload, which must be settled by value. +/// +/// Like [`LoadPermit`], this is affine. Cancellation drops it unsettled, retaining +/// the full fingerprint as a reconciliation question instead of leaving the slot +/// stuck in an invisible `Unloading` state. +#[must_use = "an unsettled permit marks the unload ambiguous when dropped"] +pub struct UnloadPermit { + ledger: Arc, + shape: Shape, + fp: Fingerprint, + permit: u64, + settled: bool, +} + +impl UnloadPermit { + pub fn fingerprint(&self) -> &Fingerprint { + &self.fp + } + + pub fn with_server_operation(&self, operation: impl FnOnce() -> T) -> T { + self.ledger.with_server_operation(operation) + } + + /// Record how the server operation ended. An uncertain answer retains the + /// fingerprint and makes the next action reconcile it before doing anything. + pub fn finish(mut self, outcome: UnloadOutcome) { + let mut slots = self.ledger.slots.lock().unwrap(); + let Some(SlotState::Unloading { fp, permit }) = slots.get(&self.shape).cloned() else { + self.settled = true; + return; + }; + if permit != self.permit { + self.settled = true; + 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(self.shape, next); + drop(slots); + self.settled = true; + } +} + +impl Drop for UnloadPermit { + fn drop(&mut self) { + if !self.settled { + let mut slots = self.ledger.slots.lock().unwrap(); + if let Some(SlotState::Unloading { permit, .. }) = slots.get(&self.shape) + && *permit == self.permit + { + tracing::warn!( + shape = self.shape.label(), + module = self.fp.id, + "audio ledger: an unload was cancelled before its outcome was known; \ + the slot needs reconciling" + ); + slots.insert( + self.shape, + SlotState::Ambiguous(Reconcile::Unload { + fp: self.fp.clone(), + }), + ); + } + } + self.ledger.operations.finish(); + } +} + +/// 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 = 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) -> Result<()> { + let ledger = Arc::clone(ledger); + tokio::task::spawn_blocking(move || reconcile_pending_blocking(&ledger)) + .await + .context("the Pulse reconciliation task failed to run")? +} + +/// Synchronous reconciliation for [`Routing::drop`](crate::host::audio::Routing). +/// The caller closes and quiesces the ledger first; serialization here also makes +/// ordinary async reconciliation wait behind any server operation already running. +pub(super) fn reconcile_pending_blocking(ledger: &Arc) -> Result<()> { + ledger.with_server_operation(|| { + 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 mut session = PulseSession::connect()?; + let observations = session + .list_modules() + .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"); + let permit = ledger + .begin_unload(Shape::LoopbackIntoCapture) + .expect("the ledger accepts the unload") + .expect("the loaded slot issues a permit"); + assert_eq!(permit.fingerprint(), &fp); + permit.finish(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) + .expect("the ledger accepts the unload") + .expect("the loaded slot issues a permit") + .finish(UnloadOutcome::Confirmed); + assert_eq!(ledger.state(Shape::LoopbackIntoCapture), SlotState::Vacant); + assert!(ledger.is_settled()); + assert!(ledger.is_clean()); + } + + #[test] + fn dropping_an_unload_permit_marks_the_slot_ambiguous() { + 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"); + let permit = ledger + .begin_unload(Shape::LoopbackIntoCapture) + .expect("the ledger accepts the unload") + .expect("the loaded slot issues a permit"); + assert!(matches!( + ledger.state(Shape::LoopbackIntoCapture), + SlotState::Unloading { .. } + )); + assert!(!ledger.is_settled(), "in-flight unloads are not settled"); + drop(permit); + assert_eq!( + ledger.state(Shape::LoopbackIntoCapture), + SlotState::Ambiguous(Reconcile::Unload { fp }) + ); + } + + #[test] + fn inverse_loopbacks_cannot_coexist_or_cross_an_unknown_boundary() { + let ledger = ModuleLedger::new(); + ledger + .begin_load(Shape::LoopbackIntoCapture, 42, token(7)) + .expect("the first loopback may load") + .commit(3) + .expect("3 is a real index"); + assert_eq!( + ledger + .begin_load(Shape::LoopbackOutOfCapture, 42, token(8)) + .err(), + Some(LedgerError::Incompatible { + shape: Shape::LoopbackOutOfCapture, + occupied: Shape::LoopbackIntoCapture, + state: "loaded" + }) + ); + + let unload = ledger + .begin_unload(Shape::LoopbackIntoCapture) + .expect("the ledger accepts the unload") + .expect("the loaded slot issues a permit"); + drop(unload); + assert_eq!( + ledger + .begin_load(Shape::LoopbackOutOfCapture, 42, token(9)) + .err(), + Some(LedgerError::Incompatible { + shape: Shape::LoopbackOutOfCapture, + occupied: Shape::LoopbackIntoCapture, + state: "ambiguous" + }), + "an unconfirmed absence must block the inverse loopback" + ); + } + + #[test] + fn capture_sink_unload_waits_for_every_dependent_slot_to_be_vacant() { + let ledger = ModuleLedger::new(); + ledger + .begin_load(Shape::LegacyCaptureSink, 42, token(1)) + .expect("the sink may load") + .commit(1) + .expect("1 is a real index"); + ledger + .begin_load(Shape::LoopbackIntoCapture, 42, token(2)) + .expect("the mirror may load") + .commit(2) + .expect("2 is a real index"); + assert_eq!( + ledger.begin_unload(Shape::LegacyCaptureSink).err(), + Some(LedgerError::Referenced { + shape: Shape::LegacyCaptureSink, + dependency: Shape::LoopbackIntoCapture, + state: "loaded" + }) + ); + ledger + .begin_unload(Shape::LoopbackIntoCapture) + .expect("the ledger accepts the unload") + .expect("the mirror issues an unload permit") + .finish(UnloadOutcome::Confirmed); + assert!( + ledger + .begin_unload(Shape::LegacyCaptureSink) + .expect("the sink is no longer referenced") + .is_some() + ); + } + + #[test] + fn closing_waits_for_a_previously_registered_operation() { + use std::sync::mpsc; + use std::time::Duration; + + let ledger = ModuleLedger::new(); + let permit = ledger + .begin_load(Shape::LegacyCaptureSink, 42, token(7)) + .expect("the operation registers before it can be detached"); + let (entered_tx, entered_rx) = mpsc::channel(); + let (done_tx, done_rx) = mpsc::channel(); + let waiter_ledger = Arc::clone(&ledger); + let waiter = std::thread::spawn(move || { + entered_tx.send(()).unwrap(); + waiter_ledger.close_and_wait(); + done_tx.send(()).unwrap(); + }); + entered_rx.recv().unwrap(); + assert!( + matches!( + done_rx.recv_timeout(Duration::from_millis(30)), + Err(mpsc::RecvTimeoutError::Timeout) + ), + "the close barrier must not pass a live affine permit" + ); + permit.abandon(); + done_rx + .recv_timeout(Duration::from_secs(1)) + .expect("settling the operation releases teardown"); + waiter.join().unwrap(); + assert!(ledger.is_clean()); + } + + #[test] + fn a_closed_ledger_refuses_event_work_but_allows_cleanup() { + let ledger = ModuleLedger::new(); + ledger + .begin_load(Shape::LegacyCaptureSink, 42, token(7)) + .expect("the sink may load before close") + .commit(3) + .expect("3 is a real index"); + ledger.close_and_wait(); + assert_eq!( + ledger + .begin_load(Shape::LoopbackIntoCapture, 42, token(8)) + .err(), + Some(LedgerError::Closed { + shape: Shape::LoopbackIntoCapture + }) + ); + assert_eq!( + ledger.begin_unload(Shape::LegacyCaptureSink).err(), + Some(LedgerError::Closed { + shape: Shape::LegacyCaptureSink + }) + ); + assert!( + ledger + .begin_cleanup_unload(Shape::LegacyCaptureSink) + .expect("teardown retains its cleanup authority") + .is_some() + ); + } + + #[test] + fn begin_unload_distinguishes_vacant_from_busy() { + let ledger = ModuleLedger::new(); + assert!( + ledger + .begin_unload(Shape::LegacyCaptureSink) + .expect("vacancy is not an error") + .is_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).err(), + Some(LedgerError::Busy { + shape: Shape::LegacyCaptureSink, + state: "loading" + }), + "an in-flight load must not look like a confirmed vacancy" + ); + } + + #[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_a_loopback_before_the_capture_sink() { + // 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::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 = ledger.loaded().into_iter().map(|fp| fp.shape).collect(); + assert_eq!( + order, + vec![Shape::LoopbackOutOfCapture, Shape::LegacyCaptureSink] + ); + assert_eq!( + order.last(), + Some(&Shape::LegacyCaptureSink), + "the sink must be unloaded after everything that references it" + ); + } +} diff --git a/src/host/mod.rs b/src/host/mod.rs index f02b0b6..944c43f 100644 --- a/src/host/mod.rs +++ b/src/host/mod.rs @@ -2,6 +2,7 @@ pub mod aec; pub mod audio; pub mod audit; mod capture; +pub mod ledger; mod observer; mod pipeline; mod quality; diff --git a/src/host/pipeline.rs b/src/host/pipeline.rs index 9f53825..fda7694 100644 --- a/src/host/pipeline.rs +++ b/src/host/pipeline.rs @@ -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;