From cc9694c13fd32132860f0355a8a2057fdf6818b1 Mon Sep 17 00:00:00 2001 From: mollusk Date: Fri, 7 Aug 2026 14:28:54 -0400 Subject: [PATCH] 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;