diff --git a/src/host/audio.rs b/src/host/audio.rs index 033e91f..3b4c1e0 100644 --- a/src/host/audio.rs +++ b/src/host/audio.rs @@ -33,33 +33,27 @@ 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; use std::thread::JoinHandle; -use std::time::Duration; +use std::time::{Duration, Instant}; use crate::cli::HostOpts; 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. +/// 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 +/// deliberately generous because exceeding it is no longer destructive: the /// ledger records the attempt, and reconciliation finds whatever the server -/// actually did. +/// 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); - -/// 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; +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; @@ -86,6 +80,17 @@ impl Routing { 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 @@ -119,13 +124,6 @@ impl Routing { "audio routing: null-sink ready (loopback skipped in strict app mode)" ); - let mut routing = Self { - ledger: Arc::clone(&ledger), - 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 ledger_for_task = Arc::clone(&ledger); @@ -138,7 +136,8 @@ impl Routing { tracing::info!( "audio routing: first stream routed → unloading default-sink loopback" ); - unload_module(&ledger_for_task, Shape::LoopbackIntoCapture).await; + 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 @@ -146,7 +145,15 @@ impl Routing { // sourced from the null-sink monitor — the chosen app // only, never the desktop/call — so it can't echo into // the capture. - ensure_loaded(&ledger_for_task, Shape::LoopbackOutOfCapture, pid).await; + 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 { state: AppAudioState::Routed, @@ -161,7 +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. - unload_module(&ledger_for_task, Shape::LoopbackOutOfCapture).await; + 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 @@ -172,6 +180,13 @@ impl Routing { ); continue; } + 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 @@ -216,6 +231,10 @@ impl Routing { /// 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. @@ -240,22 +259,23 @@ impl Routing { } } - 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; - } + // 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_settled() { + if !self.ledger.is_clean() { tracing::warn!( - "audio routing: some audio modules could not be accounted for; \ + settled = self.ledger.is_settled(), + "audio routing: some audio modules could not be removed safely; \ `pixelpass --repair` will clean up anything left behind" ); } @@ -270,22 +290,19 @@ impl Drop for Routing { /// /// 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(); } - 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); - } - if !self.ledger.is_settled() { + self.ledger.close_and_wait(); + cleanup_modules_blocking(&self.ledger); + if !self.ledger.is_clean() { tracing::warn!( - "audio routing: torn down without settling the module ledger; \ + 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" ); } @@ -408,65 +425,61 @@ async fn load_module(ledger: &Arc, shape: Shape, pid: u32) -> Resu .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)); - let mut cmd = tokio::process::Command::new("pactl"); - cmd.arg("load-module") - .arg(shape.module_name()) - .args(shape.render_args(pid, Some(&owner))) - .kill_on_drop(true); + // 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"); + } + }; - // 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(_) => { + 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 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); + 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}"); } - bail!("pactl load-module failed: {stderr}"); - } - 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. - 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) + 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" + ); + Ok(fp) + }) + .await + .context("the pactl load worker failed")? } /// Load `shape` unless the slot already holds it. @@ -496,49 +509,273 @@ async fn ensure_loaded(ledger: &Arc, shape: Shape, pid: u32) { /// 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); +async fn unload_module(ledger: &Arc, shape: Shape) -> bool { + unload_module_inner(ledger, shape, false).await } -/// 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() - { - Ok(output) if output.status.success() => { - tracing::info!(module = id, "audio routing: unloaded pactl module"); - UnloadOutcome::Confirmed +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!( + 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; } - 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}")), } } @@ -835,6 +1072,7 @@ 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 @@ -868,6 +1106,41 @@ mod tests { .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. @@ -925,33 +1198,42 @@ mod tests { 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; + // 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; - // 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. + // 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.pending().len() + ledger.loaded().len(), 1, - "the cancelled load must have left exactly one question behind" + "the cancelled load must retain exactly one tracked outcome" ); - ledger::reconcile_pending(&ledger) - .await - .expect("the ledger reconciles against the server"); - for fp in ledger.loaded() { - unload_module(&ledger, fp.shape).await; - } + cleanup_modules(&ledger).await; assert!( - ledger.is_settled(), - "every slot must end in a state we can explain" + ledger.is_clean(), + "every tracked module must be removed, not merely explained" ); assert_eq!( module_snapshot(), diff --git a/src/host/ledger.rs b/src/host/ledger.rs index 6007eec..c3b0980 100644 --- a/src/host/ledger.rs +++ b/src/host/ledger.rs @@ -33,12 +33,13 @@ //! # Why the permit is affine //! //! [`LoadPermit`] is not `Clone`, is consumed by value to settle, and its [`Drop`] -//! marks the slot [`Reconcile::Load`] when it was never settled. That is what makes -//! the guarantee structural rather than a discipline: a cancelled task drops its -//! locals, so an aborted load *cannot* silently forget a module the server may -//! already have created. Two permitted loads for one slot cannot both commit, -//! because [`ModuleLedger::begin_load`] issues a permit only for a `Vacant` slot -//! and every other state — including the ambiguous one — refuses. +//! 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 //! @@ -50,8 +51,8 @@ //! proceed until the question is answered is the whole point. use std::collections::BTreeMap; -use std::sync::atomic::{AtomicU64, Ordering}; -use std::sync::{Arc, Mutex}; +use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; +use std::sync::{Arc, Condvar, Mutex}; use anyhow::{Context as _, Result}; @@ -99,7 +100,7 @@ pub enum SlotState { 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 }, + 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. @@ -129,6 +130,20 @@ pub enum LedgerError { 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 { @@ -145,6 +160,31 @@ impl std::fmt::Display for LedgerError { "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() + ), } } } @@ -180,6 +220,59 @@ pub enum Resolution { 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 { @@ -187,9 +280,36 @@ impl ModuleLedger { 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 @@ -219,6 +339,9 @@ impl ModuleLedger { 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 => {} @@ -232,7 +355,18 @@ impl ModuleLedger { }); } } + 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 { @@ -253,36 +387,76 @@ impl ModuleLedger { /// Move a `Loaded` slot to `Unloading` and hand back what to unload. /// - /// `None` for any other state: there is nothing to unload, or the slot is not - /// in a condition to be acted on. - pub fn begin_unload(&self, shape: Shape) -> Option { - let mut slots = self.slots.lock().unwrap(); - let SlotState::Loaded { fp } = slots.get(&shape).cloned()? else { - return None; - }; - slots.insert(shape, SlotState::Unloading { fp: fp.clone() }); - Some(fp) + /// `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) } - /// Record how an unload turned out. An uncertain one becomes ambiguous rather - /// than being assumed done — the module is not forgotten either way. - pub fn finish_unload(&self, shape: Shape, outcome: UnloadOutcome) { + /// 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(); - let Some(SlotState::Unloading { fp }) = slots.get(&shape).cloned() else { - return; - }; - let next = match outcome { - UnloadOutcome::Confirmed => SlotState::Vacant, - UnloadOutcome::Uncertain(why) => { - tracing::warn!( - shape = fp.shape.label(), - module = fp.id, - "audio ledger: unload outcome uncertain ({why}); slot needs reconciling" - ); - SlotState::Ambiguous(Reconcile::Unload { fp }) + 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(), + }); } }; - slots.insert(shape, next); + 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 — @@ -313,14 +487,28 @@ impl ModuleLedger { .collect() } - /// Is every slot in a state we can explain? False while anything is ambiguous - /// or poisoned. + /// 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::Ambiguous(_) | SlotState::Poisoned { .. })) + .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. @@ -359,6 +547,14 @@ impl ModuleLedger { } } +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 @@ -375,6 +571,12 @@ pub struct LoadPermit { } 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 @@ -427,29 +629,106 @@ impl LoadPermit { impl Drop for LoadPermit { fn drop(&mut self) { - if self.settled { + 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 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(), - }), - ); + 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(); } } @@ -509,26 +788,35 @@ pub fn resolve(reconcile: &Reconcile, observations: &[ModuleObservation]) -> Res /// 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 pending = ledger.pending(); - if pending.is_empty() { - return Ok(()); - } - tracing::info!( - n = pending.len(), - "audio ledger: reconciling unresolved module slots against the server" - ); - let observations = tokio::task::spawn_blocking(|| -> Result> { - let mut session = PulseSession::connect()?; - session.list_modules() - }) - .await - .context("the Pulse listing task failed to run")? - .context("could not list Pulse modules to reconcile the audio ledger")?; + let ledger = Arc::clone(ledger); + tokio::task::spawn_blocking(move || reconcile_pending_blocking(&ledger)) + .await + .context("the Pulse reconciliation task failed to run")? +} - for (shape, reconcile) in pending { - ledger.apply(shape, resolve(&reconcile, &observations)); - } - Ok(()) +/// 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)] @@ -684,14 +972,12 @@ mod tests { .expect("a vacant slot issues a permit") .commit(3) .expect("3 is a real index"); - assert_eq!( - ledger.begin_unload(Shape::LoopbackIntoCapture), - Some(fp.clone()) - ); - ledger.finish_unload( - Shape::LoopbackIntoCapture, - UnloadOutcome::Uncertain("pactl was killed".to_string()), - ); + 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 }), @@ -707,23 +993,194 @@ mod tests { .expect("a vacant slot issues a permit") .commit(3) .expect("3 is a real index"); - ledger.begin_unload(Shape::LoopbackIntoCapture); - ledger.finish_unload(Shape::LoopbackIntoCapture, UnloadOutcome::Confirmed); + 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 begin_unload_only_acts_on_a_loaded_slot() { + fn dropping_an_unload_permit_marks_the_slot_ambiguous() { let ledger = ModuleLedger::new(); - assert_eq!(ledger.begin_unload(Shape::LegacyCaptureSink), None); + 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), - None, - "a load in flight has no id to unload yet" + 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" ); } @@ -864,13 +1321,12 @@ mod tests { } #[test] - fn loaded_lists_modules_in_shape_declaration_order() { + 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::LoopbackIntoCapture, 2), (Shape::LoopbackOutOfCapture, 3), ] { ledger @@ -880,7 +1336,10 @@ mod tests { .expect("a real index"); } let order: Vec = ledger.loaded().into_iter().map(|fp| fp.shape).collect(); - assert_eq!(order, crate::repair::plan::ALL_SHAPES.to_vec()); + assert_eq!( + order, + vec![Shape::LoopbackOutOfCapture, Shape::LegacyCaptureSink] + ); assert_eq!( order.last(), Some(&Shape::LegacyCaptureSink),