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

repair: exact-form matching, tri-state liveness, and a reference-gated sink

Browse Source 
Codex's review of the repair planner returned changes-requested: no P1s, four
reachable P2s plus a P3, all in the observation and execution layers rather than
in the discovery fix itself. All applied.

**Only the canonical forms are ours** (P2, plan.rs). `classify` recognised any
loopback with one pixelpass-looking endpoint, so a third party's
`module-loopback source=some_mic sink=pixelpass_capture_4242` was ours to unload
once that pid died — and a `sink=` token nested inside a quoted
`sink_input_properties` value could be mistaken for a top-level argument. The
whole recorded argument string must now equal what pixelpass itself would have
written.

The matcher's templates are **generated from the loader's own renderers**, not
written out beside them: hard-coding `latency_msec=20` in a matcher means a
loader change silently blinds repair to every module the new build loads, which
is the fail-closed-and-silent failure this project has been bitten by three
times. `host/audio.rs` now loads through those same renderers, so the two cannot
drift. Blindness is also reported rather than assumed impossible —
`unrecognised_pixelpass_modules` finds modules that name our sinks but match no
canonical form, and `--repair` says so loudly.

Measured on the live server before relying on it (pactl 17.0): arguments come
back byte-for-byte as passed, joined with single spaces, with `@DEFAULT_SINK@`
NOT resolved. Both facts are load-bearing for exact matching and both have a
test.

**Ordering is not a licence either** (P2, mod.rs). The plan put loopbacks before
the sink, but an unload can fail or be skipped and a loopback can appear after
planning, so the executor could still destroy a sink that something was attached
to. The sink unload is now gated on `sink_still_referenced` against the fresh
snapshot — any other module naming that sink blocks it, ours or not, because the
question is what would break, not who owns it.

**Undecidable is not dead** (P2, mod.rs). `Path::exists()` maps permission
errors, a missing `/proc` and a foreign pid namespace all to `false`, which here
read as "dead, go ahead and unload". Liveness is now `Alive | Dead | Unknown`
via `try_exists()` behind a `/proc/self/stat` preflight, and `Unknown` is
treated exactly like alive and reported separately.

**The short listing cannot carry a fingerprint** (P2, mod.rs). Its arguments are
tab-delimited text that a module argument may itself contain, and a continuation
line beginning with a digit could fabricate a row. Observations now come from two
listings: the short one for the module index, and `pactl -f json list modules`
for the exact argument. Codex proposed JSON alone; on pactl 17 its records carry
`"index": null`, so it cannot be used on its own — verified, hence the
correlation. The pairing is positional and *checked* (same count, same name at
every position, else refuse), which is also what makes a fabricated row harmless
instead of exploitable: it has no JSON counterpart, so the sequences misalign.

Normalisation is gone (P3). Within one invocation every snapshot comes from the
same server, so re-rendering does not happen, and normalising only made
genuinely different arguments compare equal. The residual ABA window — planned
module vanishes, byte-identical one takes its index — cannot be closed through an
unload API whose only argument is an index; that is now said plainly in the
fingerprint's own doc comment rather than implied away.

Five vacuity gaps Codex found in the tests, closed: a raw-pactl-output-to-plan
test (the planner suite survived a parser that dropped every argument), the
liveness-once test now uses two pids with per-pid counters, non-canonical and
nested-quoted arguments have their own cases, and the reference gate has one.

Field-verified on the live graph, both new rules: the A/B orphan test still
removes exactly the two orphans with the module table otherwise byte-identical,
and a fixture of a dead pid's legacy sink plus a non-canonical loopback naming it
leaves both alone and reports why.

251 tests (+9), clippy clean, fmt clean apart from the pre-existing
taint/tests.rs:2683.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
mollusk
2026-07-26 21:31:30 -04:00
co-authored by Claude Opus 5
parent 919d5bdef2
commit 9145b2a726
3 changed files with 881 additions and 203 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/host/audio.rs
+17 -22
View File
@@ -39,6 +39,7 @@ use std::sync::{Arc, Mutex};
use std::thread::JoinHandle;
use crate::cli::HostOpts;
use crate::repair::plan as repair_plan;
/// Owns the pactl-loaded modules plus, when filtering is active, the
/// libpipewire stream-router thread. Drop unloads modules as a backstop;
@@ -64,9 +65,9 @@ impl Routing {
/// also spawn the libpipewire thread that reroutes matching streams.
pub async fn start(opts: &HostOpts) -> Result<Self> {
let pid = std::process::id();
let sink_name = format!("pixelpass_capture_{pid}");
let sink_name = repair_plan::sink_name_for(pid);
let sink_module = load_module(&["module-null-sink", &format!("sink_name={sink_name}")])
let sink_module = load_module("module-null-sink", &repair_plan::null_sink_args(pid))
.context("failed to load module-null-sink")?;
// In strict per-app mode we never mirror the default sink: the viewer
@@ -82,13 +83,8 @@ impl Routing {
None
} else {
Some(
load_module(&[
"module-loopback",
"source=@DEFAULT_SINK@.monitor",
&format!("sink={sink_name}"),
"latency_msec=20",
])
.context("failed to load module-loopback (null-sink cleaned up on Drop)")?,
load_module("module-loopback", &repair_plan::mirror_args(pid))
.context("failed to load module-loopback (null-sink cleaned up on Drop)")?,
)
};
@@ -115,7 +111,6 @@ impl Routing {
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 sink_name_for_task = sink_name.clone();
let strict = opts.strict_audio;
let event_task = tokio::spawn(async move {
use crate::common::output::{self, AppAudioState};
@@ -137,12 +132,10 @@ impl Routing {
// 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(&[
match load_module(
"module-loopback",
&format!("source={sink_name_for_task}.monitor"),
"sink=@DEFAULT_SINK@",
"latency_msec=20",
]) {
&repair_plan::local_monitor_args(pid),
) {
Ok(id) => {
tracing::info!(
module = id,
@@ -195,12 +188,7 @@ impl Routing {
tracing::info!(
"audio routing: last routed stream gone → restoring default-sink loopback"
);
match load_module(&[
"module-loopback",
"source=@DEFAULT_SINK@.monitor",
&format!("sink={sink_name_for_task}"),
"latency_msec=20",
]) {
match load_module("module-loopback", &repair_plan::mirror_args(pid)) {
Ok(id) => {
*loopback_for_task.lock().unwrap() = Some(id);
}
@@ -349,9 +337,16 @@ struct SinkInputProperties {
// pactl module helpers
// ──────────────────────────────────────────────────────────────────────
fn load_module(args: &[&str]) -> Result<u32> {
/// Load one Pulse module and return its index.
///
/// `args` comes from the renderers in [`crate::repair::plan`] rather than being
/// written out here, so that `--repair`'s exact-form matcher and this loader are
/// one source of truth. A latency or argument change that only moved one of them
/// would leave repair silently unable to recognise the modules this build loads.
fn load_module(module: &str, args: &[String]) -> Result<u32> {
let output = Command::new("pactl")
.arg("load-module")
.arg(module)
.args(args)
.output()
.context("failed to run pactl load-module")?;
src/repair/mod.rs
+391 -59
View File
@@ -2,14 +2,36 @@
//! host.
//!
//! All of the judgement lives in [`plan`], which is pure. What remains here is
//! I/O plus one rule that cannot be expressed in a plan: **re-verify immediately
//! before destroying anything**. Pulse module indices are reused verbatim, and a
//! host can die (or come back) between the snapshot and the unload, so the plan
//! is treated as evidence that expires — never as a licence.
//! I/O plus the two rules that cannot be expressed in a plan:
//!
//! Repair never touches native PipeWire nodes. Since phase 0c the capture sink
//! is connection-owned and removes itself when its host dies, so there is
//! nothing there for repair to do and no safe way for it to help.
//! - **Re-verify immediately before destroying anything.** Pulse module indices
//! are reused verbatim, and a host can die (or come back) between the snapshot
//! and the unload, so the plan is treated as evidence that expires — never as a
//! licence.
//! - **Gate the sink on what is still attached to it**, not on the plan's ordering
//! having succeeded. An unload can fail or be skipped, and a loopback can appear
//! after the plan was made.
//!
//! Repair never touches native PipeWire nodes. Since phase 0c the capture sink is
//! connection-owned and removes itself when its host dies, so there is nothing
//! there for repair to do and no safe way for it to help.
//!
//! # Where the observations come from
//!
//! Two `pactl` invocations, because neither alone is sufficient:
//!
//! - `pactl list short modules` carries the **module index**, which is the only
//! thing `unload-module` accepts.
//! - `pactl -f json list modules` carries the **exact argument string**. Measured
//! on pactl 17.0: its records have no index at all (`"index": null`), so it
//! cannot be used on its own. Its arguments are exact, though, where the short
//! listing's are tab-delimited text that a module argument may itself contain.
//!
//! They are correlated **positionally** and the pairing is *checked*: same count,
//! same module name at every position, or the run refuses. That check is also what
//! makes a fabricated row harmless — a line crafted to look like a module in the
//! short listing has no JSON counterpart, so the sequences misalign and repair
//! stops instead of unloading an index it inferred from text.
pub mod plan;
@@ -17,19 +39,65 @@ use anyhow::{Context, Result, bail};
use std::path::Path;
use std::process::Command;
use plan::{Fingerprint, ModuleObservation};
use plan::{Fingerprint, Liveness, ModuleObservation, Shape};
/// How many times to re-take the pair of listings when they disagree. A module
/// loaded or unloaded by anyone between the two `pactl` calls shifts the pairing;
/// that is a transient, so retry a couple of times before giving up.
const SNAPSHOT_ATTEMPTS: u32 = 3;
pub async fn run() -> Result<()> {
let modules = list_modules().context("failed to list pactl modules")?;
let planned = plan::plan(&modules, is_pid_alive);
let liveness = LivenessProbe::new();
if let Some(reason) = liveness.degraded_reason() {
eprintln!(
"[pixelpass] --repair: cannot determine process liveness ({reason}); \
refusing to unload anything."
);
}
let modules = snapshot().context("failed to observe pactl modules")?;
// Say so loudly when something names our sinks but matches no shape we know:
// that is either a third party using our names, or a newer pixelpass whose
// modules this build cannot recognise. The second is how repair would go
// silently blind, so it never gets inferred from a clean exit.
let unrecognised = plan::unrecognised_pixelpass_modules(&modules);
if !unrecognised.is_empty() {
eprintln!(
"[pixelpass] --repair: {} module(s) name a pixelpass capture sink but do not match \
any shape this build knows; they are being LEFT ALONE:",
unrecognised.len()
);
for obs in &unrecognised {
eprintln!(
"[pixelpass] --repair: #{} {} {}",
obs.id, obs.name, obs.args
);
}
}
let planned = plan::plan(&modules, |pid| liveness.of(pid));
if planned.is_empty() {
if planned.live_pids.is_empty() {
let mut held = Vec::new();
if !planned.live_pids.is_empty() {
held.push(format!(
"{} live pixelpass host(s)",
planned.live_pids.len()
));
}
if !planned.unknown_pids.is_empty() {
held.push(format!(
"{} pid(s) of undeterminable liveness",
planned.unknown_pids.len()
));
}
if held.is_empty() {
println!("[pixelpass] --repair: nothing to clean up.");
} else {
println!(
"[pixelpass] --repair: nothing to clean up ({} live pixelpass host(s) left alone).",
planned.live_pids.len()
"[pixelpass] --repair: nothing to clean up ({} left alone).",
held.join(", ")
);
}
return Ok(());
@@ -43,7 +111,7 @@ pub async fn run() -> Result<()> {
// Fresh snapshot per action. Deliberately not hoisted out of the loop:
// each unload changes the module table, and the point is to decide
// against the table as it is *now*, not as it was when the plan was made.
let current = list_modules().context("failed to re-list pactl modules")?;
let current = snapshot().context("failed to re-observe pactl modules")?;
let Some(obs) = current.iter().find(|m| m.id == fp.id) else {
println!(
"[pixelpass] --repair: module #{} is already gone; skipping",
@@ -64,17 +132,43 @@ pub async fn run() -> Result<()> {
skipped += 1;
continue;
}
// Liveness last, and closest to the unload: a host that came back — or
// a stranger that inherited the pid — outranks any amount of evidence
// that this module looked orphaned.
if is_pid_alive(fp.pid) {
println!(
"[pixelpass] --repair: pid {} is alive again; leaving module #{} alone",
fp.pid, fp.id
// The sink goes last in the plan, but "last" is not the same as "nothing
// is attached any more": a loopback unload may have failed or been
// skipped, or a new one may have arrived since. Ask the fresh snapshot.
if fp.shape == Shape::LegacyCaptureSink
&& let Some(holder) = plan::sink_still_referenced(&current, fp.pid, fp.id)
{
eprintln!(
"[pixelpass] --repair: module #{} (capture sink for pid {}) is still referenced \
by module #{}; leaving the sink loaded",
fp.id, fp.pid, holder
);
skipped += 1;
continue;
}
// Liveness last, and closest to the unload: a host that came back — or
// a stranger that inherited the pid — outranks any amount of evidence
// that this module looked orphaned. Undecidable counts as back.
match liveness.of(fp.pid) {
Liveness::Dead => {}
Liveness::Alive => {
println!(
"[pixelpass] --repair: pid {} is alive again; leaving module #{} alone",
fp.pid, fp.id
);
skipped += 1;
continue;
}
Liveness::Unknown => {
eprintln!(
"[pixelpass] --repair: pid {}'s liveness became undeterminable; \
leaving module #{} alone",
fp.pid, fp.id
);
skipped += 1;
continue;
}
}
match unload_module(fp.id) {
Ok(()) => {
@@ -94,8 +188,17 @@ pub async fn run() -> Result<()> {
planned.live_pids.len()
);
}
if !planned.unknown_pids.is_empty() {
println!(
"[pixelpass] --repair: left {} pid(s) alone whose liveness could not be determined.",
planned.unknown_pids.len()
);
}
if skipped > 0 {
println!("[pixelpass] --repair: skipped {skipped} module(s) that changed under us.");
// Deliberately not "changed under us": a skip can also mean the module is
// still referenced, or its owner's liveness stopped being decidable. The
// per-module reason was printed above.
println!("[pixelpass] --repair: skipped {skipped} module(s) (reasons above).");
}
if failed > 0 {
bail!("--repair: {failed} module(s) failed to unload (see errors above)");
@@ -105,20 +208,116 @@ pub async fn run() -> Result<()> {
}
fn describe(fp: &Fingerprint) -> String {
use plan::Shape::*;
let what = match fp.shape {
LoopbackIntoCapture => "default-sink mirror",
LoopbackOutOfCapture => "local monitor",
LegacyCaptureSink => "legacy capture sink",
};
format!("unloaded {what} #{} (orphaned from pid {})", fp.id, fp.pid)
format!(
"unloaded {} #{} (orphaned from pid {})",
fp.shape.label(),
fp.id,
fp.pid
)
}
fn list_modules() -> Result<Vec<ModuleObservation>> {
// ──────────────────────────────────────────────────────────────────────
// Liveness
// ──────────────────────────────────────────────────────────────────────
/// Answers "is this pid still around", and knows when it cannot answer.
///
/// `Path::exists()` maps every error — permission, missing `/proc`, a filtered
/// mount — to `false`, which in this module would read as "dead, go ahead and
/// unload". So the probe uses `try_exists()` and preflights `/proc` itself: if
/// procfs is not answering for *this* process, no pid can be called dead.
struct LivenessProbe {
/// `None` when procfs looks usable; `Some(reason)` when every answer must be
/// [`Liveness::Unknown`].
degraded: Option<String>,
}
impl LivenessProbe {
fn new() -> Self {
// If our own entry is not visible, procfs is not a source of truth here
// (no /proc mounted, a sandbox filter, a foreign pid namespace).
let degraded = match Path::new("/proc/self/stat").try_exists() {
Ok(true) => None,
Ok(false) => Some("/proc/self/stat is not visible".to_string()),
Err(e) => Some(format!("/proc/self/stat could not be read: {e}")),
};
Self { degraded }
}
fn degraded_reason(&self) -> Option<&str> {
self.degraded.as_deref()
}
fn of(&self, pid: u32) -> Liveness {
if self.degraded.is_some() {
return Liveness::Unknown;
}
match Path::new(&format!("/proc/{pid}")).try_exists() {
Ok(true) => Liveness::Alive,
Ok(false) => Liveness::Dead,
// A pid we are not allowed to ask about is not a pid we may destroy
// state for.
Err(_) => Liveness::Unknown,
}
}
}
// ──────────────────────────────────────────────────────────────────────
// Observation
// ──────────────────────────────────────────────────────────────────────
/// One correlated observation of the whole module table.
fn snapshot() -> Result<Vec<ModuleObservation>> {
let mut last_error = None;
for attempt in 1..=SNAPSHOT_ATTEMPTS {
let short = list_short().context("failed to run `pactl list short modules`")?;
let json = list_json().context("failed to run `pactl -f json list modules`")?;
match correlate(short, json) {
Ok(observations) => return Ok(observations),
Err(e) => {
if attempt < SNAPSHOT_ATTEMPTS {
tracing::debug!("repair: module listings disagreed, retrying: {e:#}");
}
last_error = Some(e);
}
}
}
Err(last_error.expect("at least one attempt ran")).context(
"the two pactl module listings never agreed (is something loading modules right now?)",
)
}
/// Pair `(id, name)` rows with `(name, argument)` records by position, checking
/// the pairing rather than trusting it.
fn correlate(
short: Vec<(u32, String)>,
json: Vec<(String, String)>,
) -> Result<Vec<ModuleObservation>> {
if short.len() != json.len() {
bail!(
"pactl reported {} modules in the short listing and {} in JSON",
short.len(),
json.len()
);
}
let mut observations = Vec::with_capacity(short.len());
for ((id, short_name), (json_name, argument)) in short.into_iter().zip(json) {
if short_name != json_name {
bail!(
"module listings disagree at #{id}: short says {short_name:?}, JSON says \
{json_name:?}"
);
}
observations.push(ModuleObservation::new(id, &short_name, &argument));
}
Ok(observations)
}
fn list_short() -> Result<Vec<(u32, String)>> {
let output = Command::new("pactl")
.args(["list", "short", "modules"])
.output()
.context("failed to run `pactl list short modules`")?;
.context("failed to run pactl")?;
if !output.status.success() {
bail!(
"pactl list short modules failed: {}",
@@ -126,30 +325,60 @@ fn list_modules() -> Result<Vec<ModuleObservation>> {
);
}
let text = String::from_utf8(output.stdout).context("pactl returned non-UTF-8")?;
Ok(parse_modules(&text))
Ok(parse_short(&text))
}
/// `pactl list short modules` is tab-separated, but some modules render their
/// arguments as a multi-line `{ … }` block whose continuation lines start with
/// whitespace. Those wrap lines never parse as a u32 id, so filtering on that
/// is enough to keep them out.
fn parse_modules(text: &str) -> Vec<ModuleObservation> {
let mut modules = Vec::new();
/// `pactl list short modules` is tab-separated. Only the id and name are taken
/// from it — the argument comes from the JSON listing, because a module argument
/// can itself contain tabs and newlines, which this format cannot escape.
///
/// Some modules render their arguments as a multi-line `{ … }` block whose
/// continuation lines start with whitespace; those never parse as a u32 id, so
/// filtering on that keeps them out.
fn parse_short(text: &str) -> Vec<(u32, String)> {
let mut rows = Vec::new();
for line in text.lines() {
let mut parts = line.splitn(4, '\t');
let mut parts = line.split('\t');
let Some(id_str) = parts.next() else { continue };
let Ok(id) = id_str.parse::<u32>() else {
continue;
};
let Some(name) = parts.next() else { continue };
let args = parts.next().unwrap_or("");
modules.push(ModuleObservation::new(id, name, args));
rows.push((id, name.to_string()));
}
modules
rows
}
fn is_pid_alive(pid: u32) -> bool {
Path::new(&format!("/proc/{pid}")).exists()
fn list_json() -> Result<Vec<(String, String)>> {
let output = Command::new("pactl")
.args(["-f", "json", "list", "modules"])
.output()
.context("failed to run pactl")?;
if !output.status.success() {
bail!(
"pactl -f json list modules failed: {}",
String::from_utf8_lossy(&output.stderr).trim()
);
}
parse_json(&output.stdout)
}
/// The JSON listing carries the exact argument string but, on pactl 17, no module
/// index (`"index": null`) — hence the correlation with the short listing.
fn parse_json(stdout: &[u8]) -> Result<Vec<(String, String)>> {
#[derive(serde::Deserialize)]
struct JsonModule {
name: String,
/// Absent or null for a module loaded without arguments.
#[serde(default)]
argument: Option<String>,
}
let modules: Vec<JsonModule> =
serde_json::from_slice(stdout).context("pactl returned unparseable JSON")?;
Ok(modules
.into_iter()
.map(|m| (m.name, m.argument.unwrap_or_default()))
.collect())
}
fn unload_module(id: u32) -> Result<()> {
@@ -172,17 +401,22 @@ mod tests {
use super::*;
#[test]
fn parses_the_tab_separated_short_listing() {
fn parses_id_and_name_from_the_short_listing() {
let text = "5\tmodule-null-sink\tsink_name=pixelpass_capture_42\n\
10\tmodule-loopback\tsource=@DEFAULT_SINK@.monitor sink=pixelpass_capture_42\n";
let modules = parse_modules(text);
assert_eq!(modules.len(), 2);
assert_eq!(modules[0].id, 5);
assert_eq!(modules[1].name, "module-loopback");
let rows = parse_short(text);
assert_eq!(
rows,
vec![
(5, "module-null-sink".to_string()),
(10, "module-loopback".to_string())
]
);
}
/// Modules whose arguments render as a `{ … }` block wrap onto continuation
/// lines; swallowing one as a module would fabricate an entry.
/// lines; swallowing one as a module would fabricate an entry — and a
/// fabricated entry is an index we might later unload.
#[test]
fn ignores_continuation_lines_of_multi_line_argument_blocks() {
let text = "1\tlibpipewire-module-rt\t{\n\
@@ -190,17 +424,115 @@ mod tests {
\x20 rt.prio = 88\n\
}\n\
5\tmodule-null-sink\tsink_name=pixelpass_capture_42\n";
let modules = parse_modules(text);
assert_eq!(modules.len(), 2, "{modules:?}");
assert_eq!(modules[1].id, 5);
let rows = parse_short(text);
assert_eq!(rows.len(), 2, "{rows:?}");
assert_eq!(rows[1].0, 5);
}
/// A module with no arguments at all still parses — `pactl` simply stops
/// after the name.
#[test]
fn a_module_without_arguments_parses_with_empty_args() {
let modules = parse_modules("7\tmodule-always-sink\n");
assert_eq!(modules.len(), 1);
assert_eq!(modules[0].args, "");
fn parses_the_json_listing_including_an_argumentless_module() {
let json = br#"[
{"index":null,"name":"module-null-sink","argument":"sink_name=pixelpass_capture_42",
"properties":{},"usage_counter":null},
{"index":null,"name":"module-metadata","argument":null,"properties":{}}
]"#;
let records = parse_json(json).expect("valid JSON");
assert_eq!(
records,
vec![
(
"module-null-sink".to_string(),
"sink_name=pixelpass_capture_42".to_string()
),
("module-metadata".to_string(), String::new())
]
);
}
/// The correlation is the whole reason two listings are safe to combine: the
/// id comes from one and the argument from the other, so a disagreement means
/// we do not know which argument belongs to which index.
#[test]
fn correlation_pairs_ids_with_exact_arguments() {
let short = vec![
(5, "module-null-sink".to_string()),
(10, "module-loopback".to_string()),
];
let json = vec![
(
"module-null-sink".to_string(),
"sink_name=pixelpass_capture_42".to_string(),
),
(
"module-loopback".to_string(),
"source=@DEFAULT_SINK@.monitor sink=pixelpass_capture_42 latency_msec=20"
.to_string(),
),
];
let observations = correlate(short, json).expect("aligned");
assert_eq!(observations[0].id, 5);
assert!(observations[1].args.contains("latency_msec=20"));
}
/// A row that exists in one listing and not the other must stop the run. This
/// is also what makes a fabricated short-listing row harmless rather than
/// exploitable: it has no JSON counterpart, so the sequences misalign.
#[test]
fn correlation_refuses_a_mismatched_pairing() {
let short = vec![
(5, "module-null-sink".to_string()),
(10, "module-loopback".to_string()),
];
let json = vec![(
"module-null-sink".to_string(),
"sink_name=pixelpass_capture_42".to_string(),
)];
assert!(
correlate(short.clone(), json).is_err(),
"count mismatch must fail"
);
let reordered = vec![
("module-loopback".to_string(), "a=b".to_string()),
("module-null-sink".to_string(), "c=d".to_string()),
];
assert!(
correlate(short, reordered).is_err(),
"a name mismatch at any position must fail"
);
}
/// End-to-end from raw pactl text: the parser feeding the planner. Without
/// this, a parser that dropped every argument would leave the entire planner
/// suite green while real `--repair` recognised nothing at all.
#[test]
fn raw_pactl_output_becomes_a_plan() {
let short_text = "1\tlibpipewire-module-rt\t{\n\
\x20 nice.level = -11\n\
}\n\
10\tmodule-loopback\tsource=@DEFAULT_SINK@.monitor \
sink=pixelpass_capture_4242 latency_msec=20\n\
11\tmodule-loopback\tsource=pixelpass_capture_4242.monitor \
sink=@DEFAULT_SINK@ latency_msec=20\n";
let json_bytes = br#"[
{"index":null,"name":"libpipewire-module-rt","argument":"{ nice.level = -11 }"},
{"index":null,"name":"module-loopback",
"argument":"source=@DEFAULT_SINK@.monitor sink=pixelpass_capture_4242 latency_msec=20"},
{"index":null,"name":"module-loopback",
"argument":"source=pixelpass_capture_4242.monitor sink=@DEFAULT_SINK@ latency_msec=20"}
]"#;
let observations = correlate(
parse_short(short_text),
parse_json(json_bytes).expect("valid JSON"),
)
.expect("aligned");
let planned = plan::plan(&observations, |_| Liveness::Dead);
assert_eq!(
planned.unload.iter().map(|fp| fp.id).collect::<Vec<_>>(),
vec![10, 11],
"both orphan loopbacks must be planned from raw output: {planned:?}"
);
assert_eq!(planned.dead_pids.len(), 1);
}
}
src/repair/plan.rs
+473 -122
View File
@@ -6,9 +6,9 @@
//! Repair destroys server-side state belonging to processes it does not own, so
//! every interesting property is a *decision* property — which pid is dead, which
//! module belongs to it, in what order to unload — and none of them need PipeWire
//! to be exercised. Splitting the decision out means the two-host safety rules
//! below are unit-testable exactly, and the I/O shell in the parent module has
//! nothing left in it worth arguing about.
//! to be exercised. Splitting the decision out means the safety rules below are
//! unit-testable exactly, and the I/O shell in the parent module has nothing left
//! in it worth arguing about.
//!
//! # The discovery rule (phase 0c)
//!
@@ -28,7 +28,9 @@
//! 1. **A live pid is never touched**, even if it is not pixelpass. Pid reuse is
//! real, so "this pid is alive" always wins over "this module looks orphaned".
//! Leaving a stale module behind is recoverable; unloading a live host's audio
//! is not.
//! is not. **An *undecidable* pid counts as live** ([`Liveness::Unknown`]):
//! "I cannot see whether that process exists" must never become "it is dead,
//! go ahead".
//! 2. **Native nodes are never destroyed.** Repair only ever unloads Pulse
//! modules it can fingerprint. It has no business touching a live graph object,
//! and after 0c the sink cleans itself up anyway.
@@ -38,9 +40,32 @@
//! the caller must re-verify against a fresh snapshot immediately before each
//! unload ([`Fingerprint::still_matches`]). Anything that does not match
//! exactly is skipped, never unloaded.
//! 4. **Loopbacks unload before the sink they reference**, mirroring
//! `Routing::cleanup`, so PipeWire is never asked to destroy a sink that still
//! has an active loopback attached to it.
//! 4. **Ordering is not a licence either.** Planning loopbacks before the sink
//! they reference is necessary but not sufficient: an unload can *fail* or be
//! skipped, and a loopback can be created after the plan was made. So the sink
//! unload is additionally gated at execution time on
//! [`sink_still_referenced`] against the fresh snapshot — never on the plan's
//! own ordering having been followed.
//! 5. **Only the canonical forms are ours.** A module is recognised only if its
//! recorded argument string matches, exactly, what pixelpass itself would have
//! written ([`Shape::template`]). Recognising "a loopback with one
//! pixelpass-looking endpoint" would let repair unload a third party's module
//! that merely names one of our sinks.
//!
//! # Why the templates are generated, not written out
//!
//! The matcher's prefixes and suffixes are derived at runtime from the *same*
//! renderers the loader uses ([`null_sink_args`], [`mirror_args`],
//! [`local_monitor_args`]). Hard-coding `latency_msec=20` in a matcher would mean
//! that changing the loader silently blinds repair to every module the new version
//! loads — the fail-closed-and-silent failure this project has now been bitten by
//! three times. With one source of truth, a loader change moves the matcher with
//! it or fails to compile.
//!
//! Blindness is also reported rather than assumed impossible:
//! [`unrecognised_pixelpass_modules`] finds modules that name a
//! `pixelpass_capture_*` sink but do **not** match any canonical form, so the I/O
//! shell can say so loudly instead of quietly cleaning up nothing.
use std::collections::{BTreeMap, BTreeSet};
@@ -48,7 +73,57 @@ use std::collections::{BTreeMap, BTreeSet};
/// that name is the only owner identity these modules carry.
pub const SINK_NAME_PREFIX: &str = "pixelpass_capture_";
/// One module exactly as `pactl list short modules` reported it.
/// Loopback latency. Pulse's default of 200 ms is perceptible; 20 ms keeps the
/// mirrored audio tight. Shared with the matcher, so it cannot drift.
pub const LOOPBACK_LATENCY_MSEC: u32 = 20;
/// The capture sink name for a host pid.
pub fn sink_name_for(pid: u32) -> String {
format!("{SINK_NAME_PREFIX}{pid}")
}
/// `pactl load-module` arguments for the capture sink (pre-0c hosts only).
pub fn null_sink_args(pid: u32) -> Vec<String> {
vec![format!("sink_name={}", sink_name_for(pid))]
}
/// `pactl load-module` arguments for the default-sink mirror: the viewer hears
/// system audio.
pub fn mirror_args(pid: u32) -> Vec<String> {
vec![
"source=@DEFAULT_SINK@.monitor".to_string(),
format!("sink={}", sink_name_for(pid)),
format!("latency_msec={LOOPBACK_LATENCY_MSEC}"),
]
}
/// `pactl load-module` arguments for the local monitor: the sharer hears the app
/// they are sharing.
pub fn local_monitor_args(pid: u32) -> Vec<String> {
vec![
format!("source={}.monitor", sink_name_for(pid)),
"sink=@DEFAULT_SINK@".to_string(),
format!("latency_msec={LOOPBACK_LATENCY_MSEC}"),
]
}
/// How the server records an argument vector we passed as separate argv entries.
///
/// Measured on pactl 17.0 against a live pipewire-pulse: the arguments come back
/// byte-for-byte as passed, joined with single spaces, in the order given, with
/// `@DEFAULT_SINK@` **not** resolved to the concrete device name. Both facts are
/// load-bearing for exact-form matching, so both have their own test.
pub fn recorded_argument(args: &[String]) -> String {
args.join(" ")
}
/// One module exactly as the server reported it.
///
/// `args` is the **exact** recorded argument string, not a normalised one. Within
/// a single repair invocation every snapshot comes from the same server, so
/// re-rendering is not a thing that happens, and normalising would only make two
/// genuinely different arguments compare equal (whitespace inside a quoted
/// property value is not layout).
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ModuleObservation {
pub id: u32,
@@ -66,16 +141,33 @@ impl ModuleObservation {
}
}
/// Whether the owner of a module is still around.
///
/// Three states, not two: `/proc` can fail to answer (a different pid namespace,
/// a permission error, `hidepid`), and `Path::exists` collapses every one of
/// those into "no". That collapse points the wrong way — it turns "cannot tell"
/// into "dead, go ahead and unload".
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Liveness {
Alive,
Dead,
/// Undecidable. Treated exactly like `Alive` for the purposes of destroying
/// anything, and reported separately so the user knows repair held back.
Unknown,
}
/// The three module shapes pixelpass is capable of loading. Each one carries the
/// owner pid in a different place, which is exactly why discovery must consider
/// all three independently.
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
pub enum Shape {
/// `module-loopback source=@DEFAULT_SINK@.monitor sink=pixelpass_capture_<pid>`
/// — the default-sink mirror, so the viewer hears system audio.
/// `module-loopback source=@DEFAULT_SINK@.monitor sink=pixelpass_capture_<pid>
/// latency_msec=20` — the default-sink mirror, so the viewer hears system
/// audio.
LoopbackIntoCapture,
/// `module-loopback source=pixelpass_capture_<pid>.monitor sink=@DEFAULT_SINK@`
/// — the local monitor, so the sharer hears the app they are sharing.
/// `module-loopback source=pixelpass_capture_<pid>.monitor sink=@DEFAULT_SINK@
/// latency_msec=20` — the local monitor, so the sharer hears the app they are
/// sharing.
LoopbackOutOfCapture,
/// `module-null-sink sink_name=pixelpass_capture_<pid>` — the legacy capture
/// sink. Post-0c hosts do not load this at all; it exists for hosts that
@@ -86,26 +178,120 @@ pub enum Shape {
LegacyCaptureSink,
}
/// Every shape, in unload order.
pub const ALL_SHAPES: [Shape; 3] = [
Shape::LoopbackIntoCapture,
Shape::LoopbackOutOfCapture,
Shape::LegacyCaptureSink,
];
/// The pid used to render a shape's argument string when deriving its template.
/// Any value works as long as its decimal form appears exactly once in the
/// rendered arguments, which [`Template::derive`] asserts.
const TEMPLATE_SENTINEL_PID: u32 = u32::MAX;
/// An exact-match matcher for one shape, derived from that shape's own renderer.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Template {
pub module_name: &'static str,
prefix: String,
suffix: String,
}
impl Template {
/// Split a shape's rendered arguments around the pid, giving a total matcher.
fn derive(module_name: &'static str, rendered: String) -> Self {
let sentinel = TEMPLATE_SENTINEL_PID.to_string();
let (prefix, suffix) = rendered
.split_once(&sentinel)
.expect("a rendered shape must contain its pid exactly once");
debug_assert!(
!suffix.contains(&sentinel),
"the sentinel pid must appear exactly once in {rendered:?}"
);
Self {
module_name,
prefix: prefix.to_string(),
suffix: suffix.to_string(),
}
}
/// The pid this argument string names, if it is *exactly* this shape.
///
/// Total: the argument must equal `prefix ++ <pid> ++ suffix` with nothing
/// left over. A canonical decimal is required — no sign, no leading zeroes,
/// no whitespace — because `u32::from_str` accepts a leading `+`, and because
/// we only ever render a pid one way, so `pixelpass_capture_007` is not a name
/// we wrote.
pub fn pid_of(&self, args: &str) -> Option<u32> {
let digits = args
.strip_prefix(self.prefix.as_str())?
.strip_suffix(self.suffix.as_str())?;
if digits.is_empty() || !digits.bytes().all(|b| b.is_ascii_digit()) {
return None;
}
if digits.len() > 1 && digits.starts_with('0') {
return None;
}
digits.parse::<u32>().ok()
}
}
impl Shape {
/// The exact-match matcher for this shape, generated from the loader's own
/// renderer so the two cannot drift apart.
pub fn template(self) -> Template {
let pid = TEMPLATE_SENTINEL_PID;
match self {
Shape::LoopbackIntoCapture => {
Template::derive("module-loopback", recorded_argument(&mirror_args(pid)))
}
Shape::LoopbackOutOfCapture => Template::derive(
"module-loopback",
recorded_argument(&local_monitor_args(pid)),
),
Shape::LegacyCaptureSink => {
Template::derive("module-null-sink", recorded_argument(&null_sink_args(pid)))
}
}
}
/// Human label for reporting.
pub fn label(self) -> &'static str {
match self {
Shape::LoopbackIntoCapture => "default-sink mirror",
Shape::LoopbackOutOfCapture => "local monitor",
Shape::LegacyCaptureSink => "legacy capture sink",
}
}
}
/// Everything that must *still* be true of a module at the moment it is
/// unloaded — not merely when the plan was made.
///
/// It is an identity of the *observable* module, not of a generation: if the
/// planned module vanishes and a byte-identical one takes its index, this
/// compares equal. That residual ABA window cannot be closed through an unload
/// API whose only argument is an index; what closes it in practice is the
/// liveness recheck, which happens after this and nearer the unload.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Fingerprint {
pub id: u32,
pub module_name: String,
/// Whitespace-normalized, so a re-render of the same arguments compares
/// equal while any real change compares different.
/// The exact recorded argument string.
pub args: String,
pub pid: u32,
pub shape: Shape,
}
impl Fingerprint {
/// Is `obs` still the very same module this fingerprint was taken from?
/// Does `obs` still present the same observable module this fingerprint was
/// taken from?
///
/// Deliberately total: id, module name, normalized args, derived pid and
/// shape must all agree. A module index that has been reused will fail on
/// the name or the args; a module whose arguments were rewritten fails on
/// the args. Either way the caller must skip it rather than guess.
/// Deliberately total: id, module name, exact args, derived pid and shape must
/// all agree. A module index that has been reused fails on the name or the
/// args; a module whose arguments were rewritten fails on the args. Either way
/// the caller must skip it rather than guess.
pub fn still_matches(&self, obs: &ModuleObservation) -> bool {
classify(obs).as_ref() == Some(self)
}
@@ -121,6 +307,10 @@ pub struct Plan {
pub live_pids: BTreeSet<u32>,
/// Pids we concluded are gone.
pub dead_pids: BTreeSet<u32>,
/// Pids whose liveness could not be determined. Skipped like live ones, but
/// reported separately: this is the case where repair is not safe rather than
/// not needed.
pub unknown_pids: BTreeSet<u32>,
}
impl Plan {
@@ -129,79 +319,73 @@ impl Plan {
}
}
/// Collapse a module's arguments to a stable comparable form.
/// Recognise one of pixelpass's three module shapes, or `None` for a module that
/// is not ours.
///
/// `pactl list short modules` renders some modules' arguments as a multi-line
/// `{ … }` block, so the raw text carries layout that is not part of the
/// module's identity. Token order is preserved — only spacing is normalized —
/// so a genuine argument change still compares unequal.
pub fn normalize_args(args: &str) -> String {
args.split_whitespace().collect::<Vec<_>>().join(" ")
}
/// Extract `key=value` from a pactl argument string.
fn extract_kv<'a>(args: &'a str, key: &str) -> Option<&'a str> {
for token in args.split_whitespace() {
if let Some(rest) = token.strip_prefix(key)
&& let Some(value) = rest.strip_prefix('=')
{
return Some(value);
/// Exact-form only. A loopback that merely mentions one of our sink names — say a
/// third-party controller's `module-loopback source=some_mic
/// sink=pixelpass_capture_4242` — is **not** ours and must never be unloaded, and
/// a `sink=` token nested inside a quoted `sink_input_properties` value cannot be
/// mistaken for a top-level argument because the whole string must match.
pub fn classify(obs: &ModuleObservation) -> Option<Fingerprint> {
for shape in ALL_SHAPES {
let template = shape.template();
if obs.name != template.module_name {
continue;
}
if let Some(pid) = template.pid_of(&obs.args) {
return Some(Fingerprint {
id: obs.id,
module_name: obs.name.clone(),
args: obs.args.clone(),
pid,
shape,
});
}
}
None
}
/// The pid encoded in a `pixelpass_capture_<pid>` sink name.
fn pid_from_sink_name(value: &str) -> Option<u32> {
value.strip_prefix(SINK_NAME_PREFIX)?.parse::<u32>().ok()
/// Modules that name a `pixelpass_capture_*` sink but match no canonical form.
///
/// These are never touched. They exist to be *reported*: either a third party is
/// naming our sinks, or a newer pixelpass loads a shape this build does not
/// recognise. The second case is how repair would go silently blind, so it gets
/// said out loud instead of inferred from a clean exit.
pub fn unrecognised_pixelpass_modules(
observations: &[ModuleObservation],
) -> Vec<&ModuleObservation> {
observations
.iter()
.filter(|obs| obs.args.contains(SINK_NAME_PREFIX) && classify(obs).is_none())
.collect()
}
/// The pid encoded in a `pixelpass_capture_<pid>.monitor` source name.
fn pid_from_monitor_name(value: &str) -> Option<u32> {
let rest = value.strip_prefix(SINK_NAME_PREFIX)?;
rest.strip_suffix(".monitor")?.parse::<u32>().ok()
}
/// Recognise one of pixelpass's three module shapes, or `None` for a module
/// that is not ours. Never guesses: a module that merely *looks* related but
/// carries no parseable `pixelpass_capture_<pid>` identity is not ours.
pub fn classify(obs: &ModuleObservation) -> Option<Fingerprint> {
let args = normalize_args(&obs.args);
let (pid, shape) = match obs.name.as_str() {
"module-null-sink" => (
pid_from_sink_name(extract_kv(&args, "sink_name")?)?,
Shape::LegacyCaptureSink,
),
"module-loopback" => {
// Destination first: the default→capture mirror names our sink as
// `sink=`, the local monitor names it as `source=…monitor`. A
// loopback that somehow did both would be the mirror by this rule,
// which is also the shape whose unload ordering matters more.
let into = extract_kv(&args, "sink").and_then(pid_from_sink_name);
let out_of = extract_kv(&args, "source").and_then(pid_from_monitor_name);
match (into, out_of) {
(Some(pid), _) => (pid, Shape::LoopbackIntoCapture),
(None, Some(pid)) => (pid, Shape::LoopbackOutOfCapture),
(None, None) => return None,
}
}
_ => return None,
};
Some(Fingerprint {
id: obs.id,
module_name: obs.name.clone(),
args,
pid,
shape,
})
/// Is anything else in this snapshot still attached to `pid`'s capture sink?
///
/// Returns the id of the first module that references it. Deliberately textual and
/// broad — any mention of the sink name by any *other* module counts, canonical or
/// not — because the question here is "would destroying this sink break something
/// that is attached to it", not "is that attachment ours". Answering it wrongly in
/// the permissive direction is the one thing rule 4 exists to prevent.
pub fn sink_still_referenced(
observations: &[ModuleObservation],
pid: u32,
sink_module_id: u32,
) -> Option<u32> {
let sink_name = sink_name_for(pid);
observations
.iter()
.find(|obs| obs.id != sink_module_id && obs.args.contains(&sink_name))
.map(|obs| obs.id)
}
/// Turn one snapshot into an ordered unload plan.
///
/// `is_alive` is injected rather than read from `/proc` so the decision is
/// testable, and so the caller can re-check liveness again at execution time —
/// this plan is evidence, not permission.
pub fn plan(observations: &[ModuleObservation], is_alive: impl Fn(u32) -> bool) -> Plan {
/// `liveness` is injected rather than read from `/proc` so the decision is
/// testable, and so the caller can re-ask at execution time — this plan is
/// evidence, not permission.
pub fn plan(observations: &[ModuleObservation], liveness: impl Fn(u32) -> Liveness) -> Plan {
// Deduplicate by module id: a snapshot should not repeat one, but a repeated
// entry must not become a repeated unload of an index that has since been
// reused by someone else.
@@ -216,12 +400,13 @@ pub fn plan(observations: &[ModuleObservation], is_alive: impl Fn(u32) -> bool)
// three modules must not be able to change its own verdict mid-plan.
let mut live_pids = BTreeSet::new();
let mut dead_pids = BTreeSet::new();
let mut unknown_pids = BTreeSet::new();
for pid in seen.values().map(|fp| fp.pid).collect::<BTreeSet<_>>() {
if is_alive(pid) {
live_pids.insert(pid);
} else {
dead_pids.insert(pid);
}
match liveness(pid) {
Liveness::Alive => live_pids.insert(pid),
Liveness::Dead => dead_pids.insert(pid),
Liveness::Unknown => unknown_pids.insert(pid),
};
}
let mut unload: Vec<Fingerprint> = seen
@@ -236,6 +421,7 @@ pub fn plan(observations: &[ModuleObservation], is_alive: impl Fn(u32) -> bool)
unload,
live_pids,
dead_pids,
unknown_pids,
}
}
@@ -247,34 +433,72 @@ mod tests {
ModuleObservation::new(
id,
"module-null-sink",
&format!("sink_name=pixelpass_capture_{pid}"),
&recorded_argument(&null_sink_args(pid)),
)
}
fn mirror(id: u32, pid: u32) -> ModuleObservation {
ModuleObservation::new(
id,
"module-loopback",
&format!("source=@DEFAULT_SINK@.monitor sink=pixelpass_capture_{pid} latency_msec=20"),
)
ModuleObservation::new(id, "module-loopback", &recorded_argument(&mirror_args(pid)))
}
fn local_monitor(id: u32, pid: u32) -> ModuleObservation {
ModuleObservation::new(
id,
"module-loopback",
&format!("source=pixelpass_capture_{pid}.monitor sink=@DEFAULT_SINK@ latency_msec=20"),
&recorded_argument(&local_monitor_args(pid)),
)
}
fn nothing_is_alive(_: u32) -> bool {
false
fn nothing_is_alive(_: u32) -> Liveness {
Liveness::Dead
}
fn ids(plan: &Plan) -> Vec<u32> {
plan.unload.iter().map(|fp| fp.id).collect()
}
/// The renderers are the contract with the live server. These strings were
/// measured on pactl 17.0 / pipewire-pulse: arguments come back joined with
/// single spaces, in order, with `@DEFAULT_SINK@` unresolved. If this test is
/// ever changed, the matcher's exactness claim has to be re-measured.
#[test]
fn the_canonical_argument_strings_are_what_the_server_records() {
assert_eq!(
recorded_argument(&mirror_args(4242)),
"source=@DEFAULT_SINK@.monitor sink=pixelpass_capture_4242 latency_msec=20"
);
assert_eq!(
recorded_argument(&local_monitor_args(4242)),
"source=pixelpass_capture_4242.monitor sink=@DEFAULT_SINK@ latency_msec=20"
);
assert_eq!(
recorded_argument(&null_sink_args(4242)),
"sink_name=pixelpass_capture_4242"
);
}
/// Every shape must round-trip through its own generated template. This is
/// what makes the loader and the matcher one source of truth: change a
/// renderer and this fails unless the template follows.
#[test]
fn every_shape_round_trips_through_its_generated_template() {
for shape in ALL_SHAPES {
let template = shape.template();
for pid in [1_u32, 7, 4242, 999_999, u32::MAX - 1] {
let args = match shape {
Shape::LoopbackIntoCapture => recorded_argument(&mirror_args(pid)),
Shape::LoopbackOutOfCapture => recorded_argument(&local_monitor_args(pid)),
Shape::LegacyCaptureSink => recorded_argument(&null_sink_args(pid)),
};
assert_eq!(
template.pid_of(&args),
Some(pid),
"{shape:?} failed to round-trip pid {pid}"
);
}
}
}
/// The defect 0c introduces and this rewrite exists for: a dead host whose
/// capture sink was connection-owned leaves loopbacks behind with **no**
/// null-sink module to learn its pid from. The old discovery derived dead
@@ -328,12 +552,44 @@ mod tests {
null_sink(6, 200),
mirror(12, 200),
];
let plan = plan(&modules, |pid| pid == 200);
let plan = plan(&modules, |pid| {
if pid == 200 {
Liveness::Alive
} else {
Liveness::Dead
}
});
assert_eq!(ids(&plan), vec![10, 5]);
assert_eq!(plan.live_pids, BTreeSet::from([200]));
assert_eq!(plan.dead_pids, BTreeSet::from([100]));
}
/// Undecidable liveness must behave exactly like alive. `/proc` answering
/// "no" because of a pid namespace, a permission error or `hidepid` is the
/// one way a wrong verdict destroys a *running* host's audio.
#[test]
fn an_undecidable_pid_is_never_touched() {
let modules = [null_sink(5, 100), mirror(10, 100), local_monitor(11, 200)];
let plan = plan(&modules, |pid| {
if pid == 100 {
Liveness::Unknown
} else {
Liveness::Dead
}
});
assert_eq!(
ids(&plan),
vec![11],
"only the decidably-dead pid is planned"
);
assert_eq!(plan.unknown_pids, BTreeSet::from([100]));
assert!(plan.dead_pids.contains(&200));
assert!(
!plan.dead_pids.contains(&100),
"unknown must not be recorded as dead"
);
}
/// Two live hosts: repair must be a complete no-op. This is the unit-level
/// half of the two-host gate — the live half still has to be run for real.
#[test]
@@ -345,7 +601,7 @@ mod tests {
null_sink(6, 200),
mirror(12, 200),
];
let plan = plan(&modules, |_| true);
let plan = plan(&modules, |_| Liveness::Alive);
assert!(plan.is_empty(), "no live host may be touched: {plan:?}");
assert_eq!(plan.live_pids, BTreeSet::from([100, 200]));
assert!(plan.dead_pids.is_empty());
@@ -367,13 +623,72 @@ mod tests {
ModuleObservation::new(4, "module-null-sink", "sink_name=pixelpass_capture_"),
ModuleObservation::new(5, "module-null-sink", "sink_name=pixelpass_capture_abc"),
ModuleObservation::new(6, "module-null-sink", "sink_name=pixelpass_capture_-1"),
ModuleObservation::new(7, "module-loopback", "latency_msec=20"),
// `u32::from_str` accepts a leading `+`; a name we wrote never has one.
ModuleObservation::new(7, "module-null-sink", "sink_name=pixelpass_capture_+1"),
// Leading zeroes are not a name we render.
ModuleObservation::new(8, "module-null-sink", "sink_name=pixelpass_capture_007"),
ModuleObservation::new(9, "module-loopback", "latency_msec=20"),
];
let plan = plan(&modules, nothing_is_alive);
assert!(plan.is_empty(), "{plan:?}");
assert!(plan.dead_pids.is_empty(), "no owner may be invented");
}
/// A loopback that merely *names* one of our sinks is not ours. This is the
/// case where over-eager recognition would unload a live third party's
/// module: only one endpoint is a pixelpass name, so no canonical form
/// matches.
#[test]
fn a_loopback_with_only_one_pixelpass_endpoint_is_not_ours() {
let modules = [
// A third-party controller routing some microphone into our sink.
ModuleObservation::new(
20,
"module-loopback",
"source=some_mic sink=pixelpass_capture_4242 latency_msec=20",
),
// Our sink's monitor into somewhere that is not the default sink.
ModuleObservation::new(
21,
"module-loopback",
"source=pixelpass_capture_4242.monitor sink=other_sink latency_msec=20",
),
// The canonical shape with a different latency — a version of
// pixelpass this build does not know how to recognise.
ModuleObservation::new(
22,
"module-loopback",
"source=@DEFAULT_SINK@.monitor sink=pixelpass_capture_4242 latency_msec=50",
),
// Extra arguments appended: not the string we write.
ModuleObservation::new(
23,
"module-loopback",
"source=@DEFAULT_SINK@.monitor sink=pixelpass_capture_4242 latency_msec=20 \
remix=false",
),
];
let plan = plan(&modules, nothing_is_alive);
assert!(plan.is_empty(), "none of these are ours: {plan:?}");
// …but every one of them is *reported*, so blindness is never silent.
assert_eq!(unrecognised_pixelpass_modules(&modules).len(), 4);
}
/// A pixelpass-looking token nested inside a quoted property value must not
/// promote a foreign module to ours. `module-loopback` really does accept
/// `sink_input_properties`, so this argument is legal.
#[test]
fn a_nested_quoted_property_is_not_a_top_level_argument() {
let obs = ModuleObservation::new(
30,
"module-loopback",
"source=some_mic sink=real_sink \
sink_input_properties=\"media.name=sink=pixelpass_capture_4242\" latency_msec=20",
);
assert_eq!(classify(&obs), None);
assert!(plan(&[obs], nothing_is_alive).is_empty());
}
/// A repeated observation of one module must not become two unloads of an
/// index that may have been reused between them.
#[test]
@@ -382,20 +697,31 @@ mod tests {
assert_eq!(ids(&plan(&modules, nothing_is_alive)), vec![10]);
}
/// Liveness is asked once per pid. Without this, a `is_alive` that flips
/// Liveness is asked once per pid. Without this, a `liveness` that flips
/// mid-plan could unload some of a host's modules and keep others — the
/// worst possible outcome, since a half-repaired host is neither working
/// nor cleanable.
/// nor cleanable. Two pids, counted separately: one pid cannot prove
/// "once *per* pid".
#[test]
fn liveness_is_decided_once_per_pid_not_once_per_module() {
let calls = std::cell::Cell::new(0);
let modules = [null_sink(5, 42), mirror(10, 42), local_monitor(11, 42)];
let plan = plan(&modules, |_| {
calls.set(calls.get() + 1);
false
use std::cell::RefCell;
let calls: RefCell<BTreeMap<u32, u32>> = RefCell::new(BTreeMap::new());
let modules = [
null_sink(5, 42),
mirror(10, 42),
local_monitor(11, 42),
mirror(12, 99),
local_monitor(13, 99),
];
let plan = plan(&modules, |pid| {
*calls.borrow_mut().entry(pid).or_insert(0) += 1;
Liveness::Dead
});
assert_eq!(calls.get(), 1, "one pid, one liveness question");
assert_eq!(ids(&plan), vec![10, 11, 5]);
let calls = calls.into_inner();
assert_eq!(calls.get(&42), Some(&1), "one question for pid 42");
assert_eq!(calls.get(&99), Some(&1), "one question for pid 99");
assert_eq!(calls.len(), 2, "no pid asked that we have no module for");
assert_eq!(ids(&plan), vec![10, 12, 11, 13, 5]);
}
/// Re-verification: the same module still matches, and a reused index
@@ -408,16 +734,8 @@ mod tests {
assert!(fp.still_matches(&obs));
// Same index, someone else's module — the reuse case that makes a plan
// unsafe to execute blind.
assert!(!fp.still_matches(&ModuleObservation::new(
10,
"module-loopback",
"source=@DEFAULT_SINK@.monitor sink=pixelpass_capture_9999 latency_msec=20"
)));
assert!(!fp.still_matches(&ModuleObservation::new(
10,
"module-null-sink",
"sink_name=pixelpass_capture_4242"
)));
assert!(!fp.still_matches(&mirror(10, 9999)));
assert!(!fp.still_matches(&null_sink(10, 4242)));
// Same module, different index.
assert!(!fp.still_matches(&mirror(11, 4242)));
// Same identity, arguments rewritten.
@@ -426,21 +744,54 @@ mod tests {
"module-loopback",
"source=@DEFAULT_SINK@.monitor sink=pixelpass_capture_4242 latency_msec=200"
)));
// Same shape and pid, but observed under the *other* loopback shape.
assert!(!fp.still_matches(&local_monitor(10, 4242)));
}
/// Layout is not identity: pactl re-rendering the same arguments with
/// different spacing must still match, or repair would refuse to clean up
/// anything at all.
/// Whitespace is identity, not layout. The exact recorded argument is
/// compared, so a re-spaced string is a *different* argument — inside a
/// quoted property value that difference can be semantic.
#[test]
fn re_rendered_whitespace_still_matches() {
fn respaced_arguments_do_not_match() {
let fp = classify(&mirror(10, 4242)).expect("ours");
assert!(fp.still_matches(&ModuleObservation::new(
assert!(!fp.still_matches(&ModuleObservation::new(
10,
"module-loopback",
" source=@DEFAULT_SINK@.monitor\n sink=pixelpass_capture_4242\tlatency_msec=20 "
"source=@DEFAULT_SINK@.monitor sink=pixelpass_capture_4242 latency_msec=20"
)));
}
/// Rule 4: the sink's unload is gated on nothing else being attached, not on
/// the plan's ordering having been carried out. A loopback that failed to
/// unload — or one created after planning — must block it.
#[test]
fn a_remaining_loopback_blocks_the_sink_unload() {
// The mirror is still there (its unload failed, or it arrived late).
let snapshot = [null_sink(5, 4242), mirror(10, 4242)];
assert_eq!(sink_still_referenced(&snapshot, 4242, 5), Some(10));
// A foreign module attached to our sink blocks it too: the question is
// what would break, not who owns it.
let foreign = [
null_sink(5, 4242),
ModuleObservation::new(
77,
"module-loopback",
"source=some_mic sink=pixelpass_capture_4242 latency_msec=20",
),
];
assert_eq!(sink_still_referenced(&foreign, 4242, 5), Some(77));
// Nothing else attached: clear to unload. The sink's own argument names
// itself, which must not count.
let alone = [null_sink(5, 4242)];
assert_eq!(sink_still_referenced(&alone, 4242, 5), None);
// Another host's loopback is not a reference to *this* sink.
let other_host = [null_sink(5, 4242), mirror(10, 9999)];
assert_eq!(sink_still_referenced(&other_host, 4242, 5), None);
}
/// The plan must be a pure function of the snapshot: same input, same
/// order, every time.
#[test]