From cc9694c13fd32132860f0355a8a2057fdf6818b1 Mon Sep 17 00:00:00 2001 From: mollusk Date: Fri, 7 Aug 2026 14:28:54 -0400 Subject: [PATCH 1/3] feat(audio): add the module ledger as a pure state machine MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Module ids live in three `Option`s today, two of them shared with the event task. That representation cannot express "a load is in flight", and the gap is reachable: 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 sees `None`, unloads the sink, and the still-running task stores the new module's id into a mutex nobody will ever read again. A slot is therefore a state machine whose transitions admit "we do not know": Vacant / Loading / Loaded / Unloading / Ambiguous / Poisoned. The load permit is affine — not `Clone`, consumed by value to settle — and its `Drop` marks the slot ambiguous when it was never settled, so a cancelled task cannot silently forget a module the server may already have created. An ambiguous slot refuses the next load, because two sinks may share a `node.name` and `pulsesrc` attaches to the older one: loading over unresolved debris would silently steal the next session's capture. Reconciliation is by owner token, whose nonce is minted per load and so names one attempt: exactly one match adopts, zero means the load never happened, and two or more fails closed rather than guessing. The Pulse session it lists through is deliberately short-lived, because `repair::introspect` documents that the binding leaks a timed-out request's callback until disconnect — bounded for a session that ends immediately, unacceptable for one held open for the life of a share. That is the one deviation from the round-19 design, and it is why loads stay on `pactl`. Pure: no I/O in the state machine, so all 17 gates run without a Pulse server. All 8 mutants killed, each by its own named test. The stranger-at-the-same- index gate needed strengthening first — its original fixture was a non-canonical module, which `classify` discards regardless of how the match was made, so an id-only comparator would have survived it. Wiring into `host/audio.rs` follows; the dead-code warnings go with it. Co-Authored-By: Claude Opus 5 --- src/host/ledger.rs | 895 +++++++++++++++++++++++++++++++++++++++++++++ src/host/mod.rs | 1 + 2 files changed, 896 insertions(+) create mode 100644 src/host/ledger.rs diff --git a/src/host/ledger.rs b/src/host/ledger.rs new file mode 100644 index 0000000..f3a9743 --- /dev/null +++ b/src/host/ledger.rs @@ -0,0 +1,895 @@ +//! 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. That is what makes +//! the guarantee structural rather than a discipline: a cancelled task drops its +//! locals, so an aborted load *cannot* silently forget a module the server may +//! already have created. Two permitted loads for one slot cannot both commit, +//! because [`ModuleLedger::begin_load`] issues a permit only for a `Vacant` slot +//! and every other state — including the ambiguous one — refuses. +//! +//! # Why an ambiguous slot blocks the next load +//! +//! An unresolved load may or may not have created a module carrying our sink name. +//! Starting another one on top of it risks two sinks sharing a `node.name`, which +//! is measurably not an error the server reports: it accepts both, and +//! `pulsesrc device=.monitor` attaches to the *older* one. A second capture +//! session would then be silently stolen by the debris of the first. Refusing to +//! proceed until the question is answered is the whole point. + +use std::collections::BTreeMap; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::sync::{Arc, Mutex}; + +use anyhow::{Context as _, Result}; + +use crate::repair::introspect::PulseSession; +use crate::repair::plan::{ + Fingerprint, ModuleObservation, OwnerToken, Shape, classify, recorded_argument, +}; + +/// PulseAudio's "no such index" — `PA_INVALID_INDEX`, `(uint32_t) -1`. +/// +/// Every Pulse load callback reports failure by handing back this value rather +/// than an index. We load through `pactl`, which reports failure by exiting +/// non-zero instead, so seeing it on stdout would mean the tool printed a +/// sentinel we must not mistake for a module: unloading it would be a request +/// about something that cannot exist. Treated as a load whose outcome is unknown, +/// never as a successful one. +pub const PA_INVALID_INDEX: u32 = u32::MAX; + +/// What an unresolved slot needs looked up before it can be trusted again. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum Reconcile { + /// A load whose outcome is unknown: `pactl` was killed, timed out, or printed + /// something we could not read as an index. The server may or may not have + /// created the module, so it is looked for **by owner token** — the nonce is + /// minted per load, so it names this attempt and no other. + Load { + shape: Shape, + pid: u32, + token: OwnerToken, + }, + /// An unload whose outcome is unknown. The module is looked for by its full + /// fingerprint: still present means the unload did not happen, absent means it + /// did. An id alone would not do, because Pulse reuses module indices verbatim. + Unload { fp: Fingerprint }, +} + +/// One shape's slot in the module table. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum SlotState { + /// Nothing loaded, and nothing outstanding. + Vacant, + /// A load is in flight under `permit`. The id is meaningless until it commits. + Loading { token: OwnerToken, permit: u64 }, + /// A module we loaded and can name exactly. + Loaded { fp: Fingerprint }, + /// An unload is in flight. The fingerprint is retained deliberately: an unload + /// that times out must not leave the module unrecorded. + Unloading { fp: Fingerprint }, + /// The slot's real state is unknown and must be resolved against the server. + Ambiguous(Reconcile), + /// Resolution found something we refuse to act on. Terminal. + Poisoned { reason: String }, +} + +impl SlotState { + /// A short, stable label for logs and refusal messages. + fn label(&self) -> &'static str { + match self { + SlotState::Vacant => "vacant", + SlotState::Loading { .. } => "loading", + SlotState::Loaded { .. } => "loaded", + SlotState::Unloading { .. } => "unloading", + SlotState::Ambiguous(_) => "ambiguous", + SlotState::Poisoned { .. } => "poisoned", + } + } +} + +/// Why the ledger refused an operation. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum LedgerError { + /// The slot was not free. Carries the state that refused, for the log line. + Busy { shape: Shape, state: &'static str }, + /// The slot is poisoned and will not be used again this session. + Poisoned { shape: Shape, reason: String }, + /// `pactl` reported `PA_INVALID_INDEX` where an index was expected. + InvalidIndex { shape: Shape }, +} + +impl std::fmt::Display for LedgerError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + LedgerError::Busy { shape, state } => { + write!(f, "the {} slot is {state}, not free", shape.label()) + } + LedgerError::Poisoned { shape, reason } => { + write!(f, "the {} slot is poisoned: {reason}", shape.label()) + } + LedgerError::InvalidIndex { shape } => write!( + f, + "pactl reported PA_INVALID_INDEX for the {} module", + shape.label() + ), + } + } +} + +impl std::error::Error for LedgerError {} + +/// The outcome of an attempted unload, as the caller observed it. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum UnloadOutcome { + /// The server acknowledged the unload. + Confirmed, + /// The unload may or may not have happened: the command failed, was killed, + /// or its result could not be read. + Uncertain(String), +} + +/// What a reconciliation found. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum Resolution { + /// Exactly one module answers the question. Adopt it. + Adopt(Fingerprint), + /// No module answers it: the load never happened, or the unload did. + Nothing, + /// More than one module answers it. Refuse to guess — see [`Resolution`]'s use + /// in [`ModuleLedger::apply`], which poisons the slot rather than picking. + Conflict(usize), +} + +/// The ledger proper: one slot per [`Shape`], plus the permit counter. +/// +/// Held behind an `Arc` because the permit needs a way back to it from [`Drop`], +/// which is what makes a cancelled load impossible to lose. +pub struct ModuleLedger { + slots: Mutex>, + next_permit: AtomicU64, +} + +impl ModuleLedger { + pub fn new() -> Arc { + Arc::new(Self { + slots: Mutex::new(BTreeMap::new()), + next_permit: AtomicU64::new(1), + }) + } + + /// The current state of one slot. Absent keys read as [`SlotState::Vacant`]. + pub fn state(&self, shape: Shape) -> SlotState { + self.slots + .lock() + .unwrap() + .get(&shape) + .cloned() + .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`]. + /// + /// 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(); + let current = slots.get(&shape).cloned().unwrap_or(SlotState::Vacant); + match current { + SlotState::Vacant => {} + SlotState::Poisoned { reason } => { + return Err(LedgerError::Poisoned { shape, reason }); + } + other => { + return Err(LedgerError::Busy { + shape, + state: other.label(), + }); + } + } + let permit = self.next_permit.fetch_add(1, Ordering::Relaxed); + slots.insert( + shape, + SlotState::Loading { + token: token.clone(), + permit, + }, + ); + drop(slots); + Ok(LoadPermit { + ledger: Arc::clone(self), + shape, + pid, + token, + permit, + settled: false, + }) + } + + /// Move a `Loaded` slot to `Unloading` and hand back what to unload. + /// + /// `None` for any other state: there is nothing to unload, or the slot is not + /// in a condition to be acted on. + pub fn begin_unload(&self, shape: Shape) -> Option { + let mut slots = self.slots.lock().unwrap(); + let SlotState::Loaded { fp } = slots.get(&shape).cloned()? else { + return None; + }; + slots.insert(shape, SlotState::Unloading { fp: fp.clone() }); + Some(fp) + } + + /// Record how an unload turned out. An uncertain one becomes ambiguous rather + /// than being assumed done — the module is not forgotten either way. + pub fn finish_unload(&self, shape: Shape, outcome: UnloadOutcome) { + let mut slots = self.slots.lock().unwrap(); + let Some(SlotState::Unloading { fp }) = slots.get(&shape).cloned() else { + return; + }; + let next = match outcome { + UnloadOutcome::Confirmed => SlotState::Vacant, + UnloadOutcome::Uncertain(why) => { + tracing::warn!( + shape = fp.shape.label(), + module = fp.id, + "audio ledger: unload outcome uncertain ({why}); slot needs reconciling" + ); + SlotState::Ambiguous(Reconcile::Unload { fp }) + } + }; + slots.insert(shape, next); + } + + /// Every slot currently holding a module, in [`Shape`] declaration order — + /// which is unload order: the loopbacks that reference the capture sink come + /// before the sink itself. + pub fn loaded(&self) -> Vec { + self.slots + .lock() + .unwrap() + .values() + .filter_map(|state| match state { + SlotState::Loaded { fp } => Some(fp.clone()), + _ => None, + }) + .collect() + } + + /// Every question outstanding against the server, in shape order. + pub fn pending(&self) -> Vec<(Shape, Reconcile)> { + self.slots + .lock() + .unwrap() + .iter() + .filter_map(|(shape, state)| match state { + SlotState::Ambiguous(r) => Some((*shape, r.clone())), + _ => None, + }) + .collect() + } + + /// Is every slot in a state we can explain? False while anything is ambiguous + /// or poisoned. + pub fn is_settled(&self) -> bool { + self.slots + .lock() + .unwrap() + .values() + .all(|state| !matches!(state, SlotState::Ambiguous(_) | SlotState::Poisoned { .. })) + } + + /// Apply a reconciliation result to an ambiguous slot. + /// + /// A slot that is no longer ambiguous is left alone: the answer is stale, and + /// overwriting a live state with it would be worse than ignoring it. + pub fn apply(&self, shape: Shape, resolution: Resolution) { + let mut slots = self.slots.lock().unwrap(); + if !matches!(slots.get(&shape), Some(SlotState::Ambiguous(_))) { + return; + } + let next = match resolution { + Resolution::Adopt(fp) => { + tracing::info!( + shape = shape.label(), + module = fp.id, + "audio ledger: reconciled — adopting the module the server actually has" + ); + SlotState::Loaded { fp } + } + Resolution::Nothing => { + tracing::info!( + shape = shape.label(), + "audio ledger: reconciled — the server has no such module" + ); + SlotState::Vacant + } + Resolution::Conflict(n) => { + let reason = + format!("{n} modules answer to this slot's token; refusing to choose one"); + tracing::error!(shape = shape.label(), "audio ledger: {reason}"); + SlotState::Poisoned { reason } + } + }; + slots.insert(shape, next); + } +} + +/// Permission to perform exactly one load, which must be settled by value. +/// +/// Dropping it unsettled is not an error — it is the *reporting* path for a +/// cancelled or panicking load, and it marks the slot ambiguous so the module the +/// server may have created is looked for rather than forgotten. +#[must_use = "an unsettled permit marks the slot ambiguous when dropped"] +pub struct LoadPermit { + ledger: Arc, + shape: Shape, + pid: u32, + token: OwnerToken, + permit: u64, + settled: bool, +} + +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 + /// 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 { + return; + } + let mut slots = self.ledger.slots.lock().unwrap(); + // Only claim the slot if it is still *our* load. Anything else already + // moved past this permit. + if let Some(SlotState::Loading { permit, .. }) = slots.get(&self.shape) + && *permit == self.permit + { + tracing::warn!( + shape = self.shape.label(), + "audio ledger: a load was cancelled before its outcome was known; \ + the slot needs reconciling" + ); + slots.insert( + self.shape, + SlotState::Ambiguous(Reconcile::Load { + shape: self.shape, + pid: self.pid, + token: self.token.clone(), + }), + ); + } + } +} + +/// The fingerprint a successful load of `shape` for `pid` under `token` must have. +/// +/// Built from the shape's own renderer, exactly as [`classify`] rebuilds it from +/// an observation, so an adopted module and a committed one are the same value. +fn expected_fingerprint(shape: Shape, pid: u32, token: &OwnerToken, id: u32) -> Fingerprint { + Fingerprint { + id, + module_name: shape.module_name().to_string(), + args: recorded_argument(&shape.render_args(pid, Some(token))), + pid, + shape, + owner: Some(token.clone()), + } +} + +/// Answer one outstanding question against a snapshot of the module table. +/// +/// Pure: the snapshot is the only input, so every branch is testable without a +/// Pulse server. +pub fn resolve(reconcile: &Reconcile, observations: &[ModuleObservation]) -> Resolution { + let matches: Vec = 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 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")?; + + for (shape, reconcile) in pending { + ledger.apply(shape, resolve(&reconcile, &observations)); + } + Ok(()) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn token(nonce: u64) -> OwnerToken { + OwnerToken { + machine: "abc123".to_string(), + boot: "def456".to_string(), + pid_ns: 4_026_531_836, + nonce, + } + } + + /// A module observation as the server would report it for one of our loads. + fn observed(id: u32, shape: Shape, pid: u32, token: &OwnerToken) -> ModuleObservation { + ModuleObservation::new( + id, + shape.module_name(), + &recorded_argument(&shape.render_args(pid, Some(token))), + ) + } + + #[test] + fn a_committed_load_becomes_loaded() { + let ledger = ModuleLedger::new(); + let permit = ledger + .begin_load(Shape::LegacyCaptureSink, 42, token(7)) + .expect("a vacant slot issues a permit"); + assert!(matches!( + ledger.state(Shape::LegacyCaptureSink), + SlotState::Loading { .. } + )); + let fp = permit.commit(9).expect("9 is a real index"); + assert_eq!(fp.id, 9); + assert_eq!( + ledger.state(Shape::LegacyCaptureSink), + SlotState::Loaded { fp } + ); + } + + #[test] + fn dropping_a_permit_unsettled_marks_the_slot_ambiguous() { + // The orphan race in one test: a load that is cancelled between spawning + // pactl and reading its id must leave the slot asking a question, never + // looking empty. + let ledger = ModuleLedger::new(); + let permit = ledger + .begin_load(Shape::LoopbackOutOfCapture, 42, token(7)) + .expect("a vacant slot issues a permit"); + drop(permit); + assert_eq!( + ledger.state(Shape::LoopbackOutOfCapture), + SlotState::Ambiguous(Reconcile::Load { + shape: Shape::LoopbackOutOfCapture, + pid: 42, + token: token(7), + }), + "a cancelled load must be remembered as a question, not as a vacancy" + ); + assert!(!ledger.is_settled()); + } + + #[test] + fn abandoning_a_cleanly_failed_load_returns_the_slot_to_vacant() { + let ledger = ModuleLedger::new(); + ledger + .begin_load(Shape::LoopbackIntoCapture, 42, token(7)) + .expect("a vacant slot issues a permit") + .abandon(); + assert_eq!( + ledger.state(Shape::LoopbackIntoCapture), + SlotState::Vacant, + "a load the server refused created nothing to reconcile" + ); + assert!(ledger.is_settled()); + } + + #[test] + fn a_second_permit_is_refused_while_a_load_is_in_flight() { + let ledger = ModuleLedger::new(); + let _first = ledger + .begin_load(Shape::LegacyCaptureSink, 42, token(7)) + .expect("a vacant slot issues a permit"); + let second = ledger.begin_load(Shape::LegacyCaptureSink, 42, token(8)); + assert_eq!( + second.err(), + Some(LedgerError::Busy { + shape: Shape::LegacyCaptureSink, + state: "loading" + }), + "two loads for one slot must not both be permitted" + ); + } + + #[test] + fn an_ambiguous_slot_refuses_the_next_load_until_it_is_reconciled() { + // Why this matters: two sinks may share a node.name, and pulsesrc attaches + // to the older one — so loading over unresolved debris silently steals the + // next session's capture. + let ledger = ModuleLedger::new(); + drop( + ledger + .begin_load(Shape::LegacyCaptureSink, 42, token(7)) + .expect("a vacant slot issues a permit"), + ); + assert_eq!( + ledger + .begin_load(Shape::LegacyCaptureSink, 42, token(8)) + .err(), + Some(LedgerError::Busy { + shape: Shape::LegacyCaptureSink, + state: "ambiguous" + }) + ); + + // Reconciling to "nothing was created" frees it again. + ledger.apply(Shape::LegacyCaptureSink, Resolution::Nothing); + assert!( + ledger + .begin_load(Shape::LegacyCaptureSink, 42, token(9)) + .is_ok() + ); + } + + #[test] + fn an_invalid_index_is_not_adopted_and_leaves_a_question_behind() { + let ledger = ModuleLedger::new(); + let permit = ledger + .begin_load(Shape::LegacyCaptureSink, 42, token(7)) + .expect("a vacant slot issues a permit"); + assert_eq!( + permit.commit(PA_INVALID_INDEX).err(), + Some(LedgerError::InvalidIndex { + shape: Shape::LegacyCaptureSink + }) + ); + assert!( + matches!( + ledger.state(Shape::LegacyCaptureSink), + SlotState::Ambiguous(_) + ), + "a sentinel where an index belongs is exactly the unknown outcome" + ); + } + + #[test] + fn an_uncertain_unload_is_not_forgotten() { + let ledger = ModuleLedger::new(); + let fp = ledger + .begin_load(Shape::LoopbackIntoCapture, 42, token(7)) + .expect("a vacant slot issues a permit") + .commit(3) + .expect("3 is a real index"); + assert_eq!( + ledger.begin_unload(Shape::LoopbackIntoCapture), + Some(fp.clone()) + ); + ledger.finish_unload( + Shape::LoopbackIntoCapture, + UnloadOutcome::Uncertain("pactl was killed".to_string()), + ); + assert_eq!( + ledger.state(Shape::LoopbackIntoCapture), + SlotState::Ambiguous(Reconcile::Unload { fp }), + "an unload whose outcome is unknown must keep naming the module" + ); + } + + #[test] + fn a_confirmed_unload_empties_the_slot() { + let ledger = ModuleLedger::new(); + ledger + .begin_load(Shape::LoopbackIntoCapture, 42, token(7)) + .expect("a vacant slot issues a permit") + .commit(3) + .expect("3 is a real index"); + ledger.begin_unload(Shape::LoopbackIntoCapture); + ledger.finish_unload(Shape::LoopbackIntoCapture, UnloadOutcome::Confirmed); + assert_eq!(ledger.state(Shape::LoopbackIntoCapture), SlotState::Vacant); + assert!(ledger.is_settled()); + } + + #[test] + fn begin_unload_only_acts_on_a_loaded_slot() { + let ledger = ModuleLedger::new(); + assert_eq!(ledger.begin_unload(Shape::LegacyCaptureSink), None); + let _permit = ledger + .begin_load(Shape::LegacyCaptureSink, 42, token(7)) + .expect("a vacant slot issues a permit"); + assert_eq!( + ledger.begin_unload(Shape::LegacyCaptureSink), + None, + "a load in flight has no id to unload yet" + ); + } + + #[test] + fn resolve_adopts_the_one_module_carrying_our_token() { + let mine = token(7); + let reconcile = Reconcile::Load { + shape: Shape::LegacyCaptureSink, + pid: 42, + token: mine.clone(), + }; + let observations = vec![ + // A previous load by the same pid and shape: same everything but the + // per-load nonce, which is the whole reason the nonce exists. + observed(4, Shape::LegacyCaptureSink, 42, &token(6)), + observed(5, Shape::LegacyCaptureSink, 42, &mine), + // Somebody else's module entirely. + ModuleObservation::new(6, "module-null-sink", "sink_name=other"), + ]; + assert_eq!( + resolve(&reconcile, &observations), + Resolution::Adopt(expected_fingerprint(Shape::LegacyCaptureSink, 42, &mine, 5)) + ); + } + + #[test] + fn resolve_reports_nothing_when_the_server_never_created_it() { + let reconcile = Reconcile::Load { + shape: Shape::LegacyCaptureSink, + pid: 42, + token: token(7), + }; + let observations = vec![observed(4, Shape::LegacyCaptureSink, 42, &token(6))]; + assert_eq!(resolve(&reconcile, &observations), Resolution::Nothing); + } + + #[test] + fn resolve_fails_closed_when_more_than_one_module_answers() { + let mine = token(7); + let reconcile = Reconcile::Load { + shape: Shape::LegacyCaptureSink, + pid: 42, + token: mine.clone(), + }; + // Two modules carrying the same token should be impossible. If it ever + // happens, guessing which to keep is how a live sink gets unloaded. + let observations = vec![ + observed(4, Shape::LegacyCaptureSink, 42, &mine), + observed(5, Shape::LegacyCaptureSink, 42, &mine), + ]; + assert_eq!(resolve(&reconcile, &observations), Resolution::Conflict(2)); + } + + #[test] + fn a_conflict_poisons_the_slot_and_it_stays_poisoned() { + let ledger = ModuleLedger::new(); + drop( + ledger + .begin_load(Shape::LegacyCaptureSink, 42, token(7)) + .expect("a vacant slot issues a permit"), + ); + ledger.apply(Shape::LegacyCaptureSink, Resolution::Conflict(2)); + let SlotState::Poisoned { reason } = ledger.state(Shape::LegacyCaptureSink) else { + panic!("a conflict must poison the slot"); + }; + assert!(!ledger.is_settled()); + assert_eq!( + ledger + .begin_load(Shape::LegacyCaptureSink, 42, token(8)) + .err(), + Some(LedgerError::Poisoned { + shape: Shape::LegacyCaptureSink, + reason + }), + "poisoning is terminal for the session" + ); + } + + #[test] + fn resolve_finds_a_module_an_uncertain_unload_left_behind() { + let mine = token(7); + let fp = expected_fingerprint(Shape::LoopbackIntoCapture, 42, &mine, 5); + let observations = vec![observed(5, Shape::LoopbackIntoCapture, 42, &mine)]; + assert_eq!( + resolve(&Reconcile::Unload { fp: fp.clone() }, &observations), + Resolution::Adopt(fp.clone()), + "still present means the unload did not happen" + ); + assert_eq!( + resolve(&Reconcile::Unload { fp }, &[]), + Resolution::Nothing, + "absent means it did" + ); + } + + #[test] + fn an_unload_reconcile_ignores_a_stranger_at_the_same_index() { + // Pulse reuses module indices verbatim, so "something is at index 5" must + // not be read as "our module is still at index 5". + // + // The stranger has to be a module that *classifies* — another pixelpass + // host's canonical loopback — or this gate is vacuous: a comparator using + // the id alone would still return `Nothing` for junk, because junk is + // discarded by `classify` regardless of how the match was made. + let fp = expected_fingerprint(Shape::LoopbackIntoCapture, 42, &token(7), 5); + let another_hosts = observed(5, Shape::LoopbackIntoCapture, 99, &token(3)); + assert_eq!( + resolve( + &Reconcile::Unload { fp: fp.clone() }, + std::slice::from_ref(&another_hosts) + ), + Resolution::Nothing, + "another host's module at our old index is not our module" + ); + // And plain junk at that index is ignored too. + let junk = ModuleObservation::new(5, "module-loopback", "source=some_mic sink=theirs"); + assert_eq!( + resolve(&Reconcile::Unload { fp }, std::slice::from_ref(&junk)), + Resolution::Nothing + ); + } + + #[test] + fn a_stale_answer_does_not_overwrite_a_live_slot() { + let ledger = ModuleLedger::new(); + // Nothing is ambiguous, so an answer arriving late is meaningless. + let fp = ledger + .begin_load(Shape::LegacyCaptureSink, 42, token(7)) + .expect("a vacant slot issues a permit") + .commit(3) + .expect("3 is a real index"); + ledger.apply(Shape::LegacyCaptureSink, Resolution::Nothing); + assert_eq!( + ledger.state(Shape::LegacyCaptureSink), + SlotState::Loaded { fp }, + "a live state must win over a stale reconciliation" + ); + } + + #[test] + fn loaded_lists_modules_in_shape_declaration_order() { + // Declaration order is unload order: the loopbacks that reference the + // capture sink must come before the sink itself. + let ledger = ModuleLedger::new(); + for (shape, id) in [ + (Shape::LegacyCaptureSink, 1), + (Shape::LoopbackIntoCapture, 2), + (Shape::LoopbackOutOfCapture, 3), + ] { + ledger + .begin_load(shape, 42, token(u64::from(id))) + .expect("a vacant slot issues a permit") + .commit(id) + .expect("a real index"); + } + let order: Vec = ledger.loaded().into_iter().map(|fp| fp.shape).collect(); + assert_eq!(order, crate::repair::plan::ALL_SHAPES.to_vec()); + assert_eq!( + order.last(), + Some(&Shape::LegacyCaptureSink), + "the sink must be unloaded after everything that references it" + ); + } +} 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; From fa792b992764b7108115391bdc7d43bee0c69948 Mon Sep 17 00:00:00 2001 From: mollusk Date: Fri, 7 Aug 2026 14:38:46 -0400 Subject: [PATCH 2/3] 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; From 6f3e26a78c0d0dd000bef6f7e5a1ebd0d1af29f2 Mon Sep 17 00:00:00 2001 From: mollusk Date: Mon, 10 Aug 2026 03:43:31 -0400 Subject: [PATCH 3/3] fix(audio): make module teardown cancellation-safe --- src/host/audio.rs | 594 ++++++++++++++++++++++++++++++----------- src/host/ledger.rs | 647 ++++++++++++++++++++++++++++++++++++++------- 2 files changed, 991 insertions(+), 250 deletions(-) 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),