From fa792b992764b7108115391bdc7d43bee0c69948 Mon Sep 17 00:00:00 2001 From: mollusk Date: Fri, 7 Aug 2026 14:38:46 -0400 Subject: [PATCH] fix(audio): close the teardown orphan race by wiring in the ledger MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Teardown used to `event_task.abort()` and then read three `Option`s. The task's work was a synchronous `pactl` call with no await point, so the abort could not land until the load had already returned: teardown saw `None`, unloaded the sink, and the task then stored the new module's id into a mutex nobody would read again. An orphan loopback, pointing at a sink that no longer existed. Three changes close it: - Loads and unloads are `tokio::process::Command` with `kill_on_drop` and a bound, so cancellation is expressible at all. They are deliberately not `select!`ed against a cancel signal — dropping a completed load's index on the floor is the defect, not the fix. Cancellation happens by dropping the future, and the permit's `Drop` turns that into a question. - `Routing::shutdown` is async and *awaits* the event task through `&mut JoinHandle`, falling back to abort-then-await. Dropping the handle would detach the task, which is how a load could still land after teardown believed it had finished. It then runs two reconcile-then-unload rounds: one round can raise exactly one new question, and a second settles it. - `Drop` stays as the narrower synchronous backstop for the paths that never reach `shutdown`. It cannot await or reconcile, so when the ledger is left unexplained it says so and names `--repair`. A load whose outcome cannot be observed is now distinguished from one the server refused: a clean non-zero `pactl` exit abandons the permit (nothing was created), while a signal death, a timeout, an unreadable index or `PA_INVALID_INDEX` all leave it unsettled for reconciliation. Two live gates, both A/B against the real module table: teardown leaves it byte-identical with both modules carrying owner tokens, and a load cancelled mid-flight is reconciled rather than orphaned. The second asserts the slot is pending *before* reconciling, so it cannot pass by aborting before the load ever began. Both mutate global state, so they need `--test-threads=1` — running them in parallel makes each see the other's modules, which is how the first run failed. 273 tests, clippy clean under `-D warnings`, `--doctor` all checks pass. Co-Authored-By: Claude Opus 5 --- src/host/audio.rs | 509 +++++++++++++++++++++++++++++++------------ src/host/ledger.rs | 19 +- src/host/pipeline.rs | 2 +- 3 files changed, 383 insertions(+), 147 deletions(-) diff --git a/src/host/audio.rs b/src/host/audio.rs index a2d3030..033e91f 100644 --- a/src/host/audio.rs +++ b/src/host/audio.rs @@ -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, - /// 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 +85,17 @@ 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(); // 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`'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") + 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::() - .with_context(|| format!("pactl returned unexpected module ID: {id_str:?}")) + let Ok(index) = id_str.parse::() 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, 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) { + 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() { diff --git a/src/host/ledger.rs b/src/host/ledger.rs index f3a9743..6007eec 100644 --- a/src/host/ledger.rs +++ b/src/host/ledger.rs @@ -191,6 +191,13 @@ impl ModuleLedger { } /// 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() @@ -200,10 +207,6 @@ impl ModuleLedger { .unwrap_or(SlotState::Vacant) } - fn set(&self, shape: Shape, state: SlotState) { - self.slots.lock().unwrap().insert(shape, state); - } - /// Take permission to load `shape`, moving the slot to /// [`SlotState::Loading`]. /// @@ -372,14 +375,6 @@ pub struct LoadPermit { } impl LoadPermit { - pub fn shape(&self) -> Shape { - self.shape - } - - pub fn token(&self) -> &OwnerToken { - &self.token - } - /// Record that the server created the module at `index`. /// /// Fails on [`PA_INVALID_INDEX`], leaving the permit unsettled so that 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;