diff --git a/src/host/audio.rs b/src/host/audio.rs index 615538e..9238a60 100644 --- a/src/host/audio.rs +++ b/src/host/audio.rs @@ -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 { 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 { +/// 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 { let output = Command::new("pactl") .arg("load-module") + .arg(module) .args(args) .output() .context("failed to run pactl load-module")?; diff --git a/src/repair/mod.rs b/src/repair/mod.rs index ea920ba..cfa61b9 100644 --- a/src/repair/mod.rs +++ b/src/repair/mod.rs @@ -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(¤t, 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> { +// ────────────────────────────────────────────────────────────────────── +// 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, +} + +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> { + 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> { + 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> { 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> { ); } 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 { - 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::() 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> { + 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> { + #[derive(serde::Deserialize)] + struct JsonModule { + name: String, + /// Absent or null for a module loaded without arguments. + #[serde(default)] + argument: Option, + } + let modules: Vec = + 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![10, 11], + "both orphan loopbacks must be planned from raw output: {planned:?}" + ); + assert_eq!(planned.dead_pids.len(), 1); } } diff --git a/src/repair/plan.rs b/src/repair/plan.rs index da440f2..bf299d5 100644 --- a/src/repair/plan.rs +++ b/src/repair/plan.rs @@ -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 { + 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 { + 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 { + 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_` - /// — the default-sink mirror, so the viewer hears system audio. + /// `module-loopback source=@DEFAULT_SINK@.monitor sink=pixelpass_capture_ + /// latency_msec=20` — the default-sink mirror, so the viewer hears system + /// audio. LoopbackIntoCapture, - /// `module-loopback source=pixelpass_capture_.monitor sink=@DEFAULT_SINK@` - /// — the local monitor, so the sharer hears the app they are sharing. + /// `module-loopback source=pixelpass_capture_.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_` — 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 ++ ++ 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 { + 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::().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, /// Pids we concluded are gone. pub dead_pids: BTreeSet, + /// 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, } 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::>().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 { + 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_` sink name. -fn pid_from_sink_name(value: &str) -> Option { - value.strip_prefix(SINK_NAME_PREFIX)?.parse::().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_.monitor` source name. -fn pid_from_monitor_name(value: &str) -> Option { - let rest = value.strip_prefix(SINK_NAME_PREFIX)?; - rest.strip_suffix(".monitor")?.parse::().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_` identity is not ours. -pub fn classify(obs: &ModuleObservation) -> Option { - 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 { + 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::>() { - 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 = 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 { 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> = 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]