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

repair: find the orphans a connection-owned sink leaves behind

Browse Source 
Phase 0c's first landed piece. `--repair` does not get smaller when the capture
sink becomes connection-owned — it goes blind, which is the opposite of the
assumption I started from and the finding that reordered this phase.

Today repair learns a dead host's pid ONLY from `module-null-sink
sink_name=pixelpass_capture_<pid>`, and matches loopbacks only if their pid is
already in that set. After 0c the sink is a native node that removes itself with
its owning connection, so a hard-killed host leaves two Pulse loopbacks and *no
null-sink module to learn the pid from*. The set stays empty, nothing matches,
and the orphans are invisible forever. Candidate pids are now derived
independently from all three module shapes.

Split into a pure planner (`repair/plan.rs`) and an I/O shell, because every
interesting property here is a decision — which pid is dead, which module is
whose, in what order to unload — and none of them need PipeWire to exercise.

The rule that is new, and that the old code could not express: **a plan is not a
licence.** Pulse module indices are reused verbatim, so an id planned against one
module can name a different live module by the time the unload runs; a pid
recheck alone does not catch that. Every action now carries a full fingerprint
(id, module name, normalized args, derived pid, shape) which is re-verified
against a FRESH snapshot immediately before each unload, with liveness rechecked
last, closest to the destruction. Anything that does not match exactly is
skipped and said out loud — never unloaded on the strength of a stale plan.

Ordering is carried by `Shape`'s declaration order rather than by two separate
passes, so loopbacks unload before the sink they reference by construction.
Liveness is asked once per pid, not once per module: a flapping answer must not
be able to half-repair a host, which is the one outcome worse than doing nothing.

Field-tested against a real post-0c orphan, not just mocked: a connection-owned
sink created via `pw-cli create-node adapter` (module-null-sink count: 0, so the
old discovery provably could not see it), both loopback shapes loaded against it,
then SIGKILL of the owning connection. The sink vanished on its own, both
loopbacks survived, `--repair` removed exactly those two, and the module table
was otherwise byte-identical before and after — the collateral-damage half of the
two-host safety property.

Mutation-verified, five mutants, each killed by its own gate: null-sink-only
discovery (the 0c blindness itself), id-ordered unloads instead of shape-ordered,
a fingerprint that compares only the index (exactly one), no liveness filter, and
un-normalized args (exactly one).

Still owed: the live two-host gate (two real hosts, kill one, prove the other's
graph and modules are untouched) cannot run until the native sink exists, so it
is deferred to 0c's combined exit gate. This lands with test + single-host field
proof only, which Codex agreed is an acceptable phase dependency rather than an
objection to landing repair first.

242 tests (+11), clippy clean, fmt clean apart from the known pre-existing
`taint/tests.rs:2683` — pixelpass is never `cargo fmt`ed.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
mollusk
2026-07-26 20:09:40 -04:00
co-authored by Claude Opus 5
parent c78eb2dd39
commit 919d5bdef2
3 changed files with 658 additions and 236 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/repair.rs
-236
View File
@@ -1,236 +0,0 @@
//! `--repair`: clean up null-sinks and loopbacks left behind by a crashed
//! pixelpass host. Identifies orphans by the `pixelpass_capture_<pid>`
//! name pattern + dead-PID check, then unloads paired loopbacks first
//! (mirrors `Routing::shutdown`'s order so PipeWire doesn't leave zombie
//! links). Live PIDs — including this process and any other running
//! pixelpass — are left alone.
use anyhow::{Context, Result, bail};
use std::collections::HashSet;
use std::path::Path;
use std::process::Command;
const SINK_NAME_PREFIX: &str = "pixelpass_capture_";
pub async fn run() -> Result<()> {
let modules = list_modules().context("failed to list pactl modules")?;
let mut dead_sinks: Vec<OrphanSink> = Vec::new();
let mut dead_pids: HashSet<u32> = HashSet::new();
let mut live_skipped: u32 = 0;
for m in &modules {
if m.name != "module-null-sink" {
continue;
}
let Some(sink_name) = extract_kv(&m.args, "sink_name") else {
continue;
};
let Some(pid_str) = sink_name.strip_prefix(SINK_NAME_PREFIX) else {
continue;
};
let Ok(pid) = pid_str.parse::<u32>() else {
continue;
};
if is_pid_alive(pid) {
live_skipped += 1;
continue;
}
dead_pids.insert(pid);
dead_sinks.push(OrphanSink {
id: m.id,
sink_name: sink_name.to_string(),
pid,
});
}
let mut dead_loopbacks: Vec<u32> = Vec::new();
for m in &modules {
if m.name != "module-loopback" {
continue;
}
// A pixelpass loopback references a capture sink either as its
// destination (`sink=pixelpass_capture_<pid>` — the default→null
// mirror) or as its source (`source=pixelpass_capture_<pid>.monitor`
// — the local monitor that lets the sharer hear the app). Match both.
let Some(pid) = loopback_capture_pid(&m.args) else {
continue;
};
if dead_pids.contains(&pid) {
dead_loopbacks.push(m.id);
}
}
if dead_sinks.is_empty() && dead_loopbacks.is_empty() {
if live_skipped > 0 {
println!(
"[pixelpass] --repair: nothing to clean up ({live_skipped} live pixelpass host(s) left alone)."
);
} else {
println!("[pixelpass] --repair: nothing to clean up.");
}
return Ok(());
}
let mut unloaded = 0u32;
let mut failed = 0u32;
for id in &dead_loopbacks {
match unload_module(*id) {
Ok(()) => {
println!("[pixelpass] --repair: unloaded loopback module #{id}");
unloaded += 1;
}
Err(e) => {
eprintln!("[pixelpass] --repair: failed to unload loopback #{id}: {e:#}");
failed += 1;
}
}
}
for orphan in &dead_sinks {
match unload_module(orphan.id) {
Ok(()) => {
println!(
"[pixelpass] --repair: unloaded {} (orphaned from pid {})",
orphan.sink_name, orphan.pid
);
unloaded += 1;
}
Err(e) => {
eprintln!(
"[pixelpass] --repair: failed to unload {} (#{}): {e:#}",
orphan.sink_name, orphan.id
);
failed += 1;
}
}
}
if live_skipped > 0 {
println!("[pixelpass] --repair: left {live_skipped} live pixelpass host(s) alone.");
}
if failed > 0 {
bail!("--repair: {failed} module(s) failed to unload (see errors above)");
}
println!("[pixelpass] --repair: cleaned up {unloaded} module(s).");
Ok(())
}
struct Module {
id: u32,
name: String,
args: String,
}
struct OrphanSink {
id: u32,
sink_name: String,
pid: u32,
}
fn list_modules() -> Result<Vec<Module>> {
let output = Command::new("pactl")
.args(["list", "short", "modules"])
.output()
.context("failed to run `pactl list short modules`")?;
if !output.status.success() {
bail!(
"pactl list short modules failed: {}",
String::from_utf8_lossy(&output.stderr).trim()
);
}
let text = String::from_utf8(output.stdout).context("pactl returned non-UTF-8")?;
let mut modules = Vec::new();
// `pactl list short modules` is tab-separated, but some modules have
// multi-line `{ ... }` argument blocks that wrap onto continuation
// lines starting with whitespace. The wrap lines never parse as a
// u32 ID, so the simple per-line + parse-id filter is robust.
for line in text.lines() {
let mut parts = line.splitn(4, '\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("").to_string();
modules.push(Module {
id,
name: name.to_string(),
args,
});
}
Ok(modules)
}
/// The `pixelpass_capture_<pid>` PID a loopback references, whether the capture
/// sink is its destination (`sink=pixelpass_capture_<pid>`) or its source
/// (`source=pixelpass_capture_<pid>.monitor`). `None` for unrelated loopbacks.
fn loopback_capture_pid(args: &str) -> Option<u32> {
let from_sink = extract_kv(args, "sink").and_then(|v| v.strip_prefix(SINK_NAME_PREFIX));
let from_source = extract_kv(args, "source")
.and_then(|v| v.strip_prefix(SINK_NAME_PREFIX))
.and_then(|rest| rest.strip_suffix(".monitor"));
from_sink
.or(from_source)
.and_then(|pid| pid.parse::<u32>().ok())
}
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);
}
}
None
}
fn is_pid_alive(pid: u32) -> bool {
Path::new(&format!("/proc/{pid}")).exists()
}
fn unload_module(id: u32) -> Result<()> {
let output = Command::new("pactl")
.arg("unload-module")
.arg(id.to_string())
.output()
.context("failed to run pactl unload-module")?;
if !output.status.success() {
bail!(
"pactl unload-module #{id}: {}",
String::from_utf8_lossy(&output.stderr).trim()
);
}
Ok(())
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn loopback_pid_matches_default_null_mirror_by_sink() {
// The default→null loopback: capture sink is the destination.
let args = "source=@DEFAULT_SINK@.monitor sink=pixelpass_capture_4242 latency_msec=20";
assert_eq!(loopback_capture_pid(args), Some(4242));
}
#[test]
fn loopback_pid_matches_local_monitor_by_source() {
// The local monitor: capture sink's monitor is the source, and the
// destination is the real default sink (not a pixelpass name).
let args = "source=pixelpass_capture_4242.monitor sink=@DEFAULT_SINK@ latency_msec=20";
assert_eq!(loopback_capture_pid(args), Some(4242));
}
#[test]
fn loopback_pid_ignores_unrelated_loopback() {
assert_eq!(
loopback_capture_pid("source=alsa_output.pci.monitor sink=some_other_sink"),
None
);
}
}
src/repair/mod.rs
+206
View File
@@ -0,0 +1,206 @@
//! `--repair`: clean up the Pulse modules left behind by a crashed pixelpass
//! 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.
//!
//! 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.
pub mod plan;
use anyhow::{Context, Result, bail};
use std::path::Path;
use std::process::Command;
use plan::{Fingerprint, ModuleObservation};
pub async fn run() -> Result<()> {
let modules = list_modules().context("failed to list pactl modules")?;
let planned = plan::plan(&modules, is_pid_alive);
if planned.is_empty() {
if planned.live_pids.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()
);
}
return Ok(());
}
let mut unloaded = 0u32;
let mut skipped = 0u32;
let mut failed = 0u32;
for fp in &planned.unload {
// 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 Some(obs) = current.iter().find(|m| m.id == fp.id) else {
println!(
"[pixelpass] --repair: module #{} is already gone; skipping",
fp.id
);
skipped += 1;
continue;
};
if !fp.still_matches(obs) {
// The index now names something else, or the same module's
// arguments changed. Either way we no longer know what we would be
// destroying, so we do not destroy it.
eprintln!(
"[pixelpass] --repair: module #{} no longer matches what was planned \
(index reused?); refusing to unload it",
fp.id
);
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
);
skipped += 1;
continue;
}
match unload_module(fp.id) {
Ok(()) => {
println!("[pixelpass] --repair: {}", describe(fp));
unloaded += 1;
}
Err(e) => {
eprintln!("[pixelpass] --repair: failed to unload #{}: {e:#}", fp.id);
failed += 1;
}
}
}
if !planned.live_pids.is_empty() {
println!(
"[pixelpass] --repair: left {} live pixelpass host(s) alone.",
planned.live_pids.len()
);
}
if skipped > 0 {
println!("[pixelpass] --repair: skipped {skipped} module(s) that changed under us.");
}
if failed > 0 {
bail!("--repair: {failed} module(s) failed to unload (see errors above)");
}
println!("[pixelpass] --repair: cleaned up {unloaded} module(s).");
Ok(())
}
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)
}
fn list_modules() -> Result<Vec<ModuleObservation>> {
let output = Command::new("pactl")
.args(["list", "short", "modules"])
.output()
.context("failed to run `pactl list short modules`")?;
if !output.status.success() {
bail!(
"pactl list short modules failed: {}",
String::from_utf8_lossy(&output.stderr).trim()
);
}
let text = String::from_utf8(output.stdout).context("pactl returned non-UTF-8")?;
Ok(parse_modules(&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();
for line in text.lines() {
let mut parts = line.splitn(4, '\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));
}
modules
}
fn is_pid_alive(pid: u32) -> bool {
Path::new(&format!("/proc/{pid}")).exists()
}
fn unload_module(id: u32) -> Result<()> {
let output = Command::new("pactl")
.arg("unload-module")
.arg(id.to_string())
.output()
.context("failed to run pactl unload-module")?;
if !output.status.success() {
bail!(
"pactl unload-module #{id}: {}",
String::from_utf8_lossy(&output.stderr).trim()
);
}
Ok(())
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn parses_the_tab_separated_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");
}
/// Modules whose arguments render as a `{ … }` block wrap onto continuation
/// lines; swallowing one as a module would fabricate an entry.
#[test]
fn ignores_continuation_lines_of_multi_line_argument_blocks() {
let text = "1\tlibpipewire-module-rt\t{\n\
\x20 nice.level = -11\n\
\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);
}
/// 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, "");
}
}
src/repair/plan.rs
+452
View File
@@ -0,0 +1,452 @@
//! The pure half of `--repair`: turn one observed module snapshot into an
//! ordered list of unloads, with no I/O and no destruction.
//!
//! # Why this is a separate, pure module
//!
//! 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.
//!
//! # The discovery rule (phase 0c)
//!
//! Before 0c the capture sink was a `module-null-sink`, so repair could learn a
//! host's pid from that module and then match its loopbacks. After 0c the sink is
//! a **connection-owned native node**: it disappears on its own when the host
//! dies, and it is not a Pulse module at all. A dead host therefore leaves
//! loopbacks with **no null-sink module to learn the pid from**, and the old
//! discovery could not see them — repair did not get smaller, it went blind.
//!
//! So candidate pids are derived **independently from all three shapes**
//! ([`Shape`]), and a shape's absence is never taken as permission to skip
//! another shape's cleanup.
//!
//! # The safety rules, in order of how much damage they prevent
//!
//! 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.
//! 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.
//! 3. **A plan is not a licence.** Pulse module indices are reused verbatim, so an
//! id planned against one module can name a *different* live module by the time
//! the unload runs. Every action therefore carries a full [`Fingerprint`] which
//! 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.
use std::collections::{BTreeMap, BTreeSet};
/// Every pixelpass capture sink is named `pixelpass_capture_<pid>`; the pid in
/// 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.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ModuleObservation {
pub id: u32,
pub name: String,
pub args: String,
}
impl ModuleObservation {
pub fn new(id: u32, name: &str, args: &str) -> Self {
Self {
id,
name: name.to_string(),
args: args.to_string(),
}
}
}
/// 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.
LoopbackIntoCapture,
/// `module-loopback source=pixelpass_capture_<pid>.monitor sink=@DEFAULT_SINK@`
/// — 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
/// predate the connection-owned sink, and for mixed sessions.
///
/// Declared **last** on purpose: `Ord` gives the unload ordering, and the
/// sink must go after the loopbacks that reference it.
LegacyCaptureSink,
}
/// Everything that must *still* be true of a module at the moment it is
/// unloaded — not merely when the plan was made.
#[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.
pub args: String,
pub pid: u32,
pub shape: Shape,
}
impl Fingerprint {
/// Is `obs` still the very same 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.
pub fn still_matches(&self, obs: &ModuleObservation) -> bool {
classify(obs).as_ref() == Some(self)
}
}
/// The result of planning: what to unload, and what was deliberately left alone.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct Plan {
/// In execution order: every loopback before any legacy sink.
pub unload: Vec<Fingerprint>,
/// Pids that are still alive and were therefore skipped entirely. Includes
/// this process and any other running pixelpass.
pub live_pids: BTreeSet<u32>,
/// Pids we concluded are gone.
pub dead_pids: BTreeSet<u32>,
}
impl Plan {
pub fn is_empty(&self) -> bool {
self.unload.is_empty()
}
}
/// Collapse a module's arguments to a stable comparable form.
///
/// `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);
}
}
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()
}
/// 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,
})
}
/// 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 {
// 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.
let mut seen: BTreeMap<u32, Fingerprint> = BTreeMap::new();
for obs in observations {
if let Some(fp) = classify(obs) {
seen.entry(fp.id).or_insert(fp);
}
}
// Liveness is asked once per distinct pid, not once per module: a host with
// 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();
for pid in seen.values().map(|fp| fp.pid).collect::<BTreeSet<_>>() {
if is_alive(pid) {
live_pids.insert(pid);
} else {
dead_pids.insert(pid);
}
}
let mut unload: Vec<Fingerprint> = seen
.into_values()
.filter(|fp| dead_pids.contains(&fp.pid))
.collect();
// `Shape`'s declaration order is the unload order: loopbacks before the sink
// they reference. Id breaks ties so the plan is deterministic.
unload.sort_by_key(|fp| (fp.shape, fp.id));
Plan {
unload,
live_pids,
dead_pids,
}
}
#[cfg(test)]
mod tests {
use super::*;
fn null_sink(id: u32, pid: u32) -> ModuleObservation {
ModuleObservation::new(
id,
"module-null-sink",
&format!("sink_name=pixelpass_capture_{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"),
)
}
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"),
)
}
fn nothing_is_alive(_: u32) -> bool {
false
}
fn ids(plan: &Plan) -> Vec<u32> {
plan.unload.iter().map(|fp| fp.id).collect()
}
/// 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
/// pids only from `module-null-sink`, so it found nothing here.
#[test]
fn orphan_loopbacks_are_found_without_any_null_sink() {
let modules = [mirror(10, 4242), local_monitor(11, 4242)];
let plan = plan(&modules, nothing_is_alive);
assert_eq!(ids(&plan), vec![10, 11]);
assert_eq!(plan.dead_pids, BTreeSet::from([4242]));
}
/// Either loopback shape alone must be enough to identify the owner — the
/// local monitor names the capture sink only as its *source*.
#[test]
fn each_loopback_shape_identifies_the_owner_on_its_own() {
assert_eq!(
ids(&plan(&[local_monitor(11, 7)], nothing_is_alive)),
vec![11]
);
assert_eq!(ids(&plan(&[mirror(10, 7)], nothing_is_alive)), vec![10]);
}
/// The legacy shape still works, and the sink unloads *after* both
/// loopbacks that reference it.
#[test]
fn legacy_sink_unloads_after_the_loopbacks_that_reference_it() {
// Deliberately snapshot-ordered sink-first, so passing requires the
// plan to reorder rather than to preserve input order.
let modules = [
null_sink(5, 4242),
local_monitor(11, 4242),
mirror(10, 4242),
];
let plan = plan(&modules, nothing_is_alive);
assert_eq!(ids(&plan), vec![10, 11, 5]);
assert_eq!(
plan.unload.last().unwrap().shape,
Shape::LegacyCaptureSink,
"the sink must be last or PipeWire is asked to destroy a sink with a live loopback"
);
}
/// A live pid is never touched, even when a dead one is being cleaned up in
/// the same run.
#[test]
fn a_live_host_is_left_alone_while_a_dead_one_is_cleaned() {
let modules = [
null_sink(5, 100),
mirror(10, 100),
null_sink(6, 200),
mirror(12, 200),
];
let plan = plan(&modules, |pid| pid == 200);
assert_eq!(ids(&plan), vec![10, 5]);
assert_eq!(plan.live_pids, BTreeSet::from([200]));
assert_eq!(plan.dead_pids, BTreeSet::from([100]));
}
/// 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]
fn two_live_hosts_produce_no_actions_at_all() {
let modules = [
null_sink(5, 100),
mirror(10, 100),
local_monitor(11, 100),
null_sink(6, 200),
mirror(12, 200),
];
let plan = plan(&modules, |_| true);
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());
}
/// Modules belonging to anything else are invisible to repair, including
/// near-misses that mention a sink we do not own.
#[test]
fn unrelated_and_malformed_modules_are_never_planned() {
let modules = [
ModuleObservation::new(1, "module-null-sink", "sink_name=some_other_sink"),
ModuleObservation::new(
2,
"module-loopback",
"source=alsa_output.pci.monitor sink=x",
),
ModuleObservation::new(3, "module-echo-cancel", "sink_name=pixelpass_capture_9"),
// Our prefix, but no parseable pid — we do not guess an owner.
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"),
];
let plan = plan(&modules, nothing_is_alive);
assert!(plan.is_empty(), "{plan:?}");
assert!(plan.dead_pids.is_empty(), "no owner may be invented");
}
/// A repeated observation of one module must not become two unloads of an
/// index that may have been reused between them.
#[test]
fn a_duplicated_observation_yields_one_action() {
let modules = [mirror(10, 4242), mirror(10, 4242)];
assert_eq!(ids(&plan(&modules, nothing_is_alive)), vec![10]);
}
/// Liveness is asked once per pid. Without this, a `is_alive` 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.
#[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
});
assert_eq!(calls.get(), 1, "one pid, one liveness question");
assert_eq!(ids(&plan), vec![10, 11, 5]);
}
/// Re-verification: the same module still matches, and a reused index
/// carrying a different module does not.
#[test]
fn a_fingerprint_only_matches_the_module_it_was_taken_from() {
let obs = mirror(10, 4242);
let fp = classify(&obs).expect("ours");
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"
)));
// Same module, different index.
assert!(!fp.still_matches(&mirror(11, 4242)));
// Same identity, arguments rewritten.
assert!(!fp.still_matches(&ModuleObservation::new(
10,
"module-loopback",
"source=@DEFAULT_SINK@.monitor sink=pixelpass_capture_4242 latency_msec=200"
)));
}
/// 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.
#[test]
fn re_rendered_whitespace_still_matches() {
let fp = classify(&mirror(10, 4242)).expect("ours");
assert!(fp.still_matches(&ModuleObservation::new(
10,
"module-loopback",
" source=@DEFAULT_SINK@.monitor\n sink=pixelpass_capture_4242\tlatency_msec=20 "
)));
}
/// The plan must be a pure function of the snapshot: same input, same
/// order, every time.
#[test]
fn planning_is_deterministic_regardless_of_snapshot_order() {
let a = [null_sink(5, 42), mirror(10, 42), local_monitor(11, 42)];
let b = [local_monitor(11, 42), null_sink(5, 42), mirror(10, 42)];
assert_eq!(plan(&a, nothing_is_alive), plan(&b, nothing_is_alive));
}
}