mollusk/pixelpass
1
1
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(audio): add the module ledger as a pure state machine

Browse Source 
Module ids live in three `Option<u32>`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 <noreply@anthropic.com>
This commit is contained in:
mollusk
2026-08-07 14:28:54 -04:00
co-authored by Claude Opus 5
parent 911f0521f8
commit cc9694c13f
2 changed files with 896 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/host/ledger.rs
+895
View File
@@ -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<u32>`s, two of them shared with the
//! event task behind an `Arc<Mutex<…>>`. 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<u32>` 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=<name>.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<BTreeMap<Shape, SlotState>>,
next_permit: AtomicU64,
}
impl ModuleLedger {
pub fn new() -> Arc<Self> {
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<Self>,
shape: Shape,
pid: u32,
token: OwnerToken,
) -> Result<LoadPermit, LedgerError> {
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<Fingerprint> {
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<Fingerprint> {
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<ModuleLedger>,
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<Fingerprint, LedgerError> {
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<Fingerprint> = 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<ModuleLedger>) -> 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<Vec<ModuleObservation>> {
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<Shape> = 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"
);
}
}
src/host/mod.rs
+1
View File
@@ -2,6 +2,7 @@ pub mod aec;
pub mod audio;
pub mod audit;
mod capture;
pub mod ledger;
mod observer;
mod pipeline;
mod quality;