diff --git a/Cargo.lock b/Cargo.lock index 9653c4c..9b29631 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2958,6 +2958,33 @@ version = "0.2.16" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b6d2cec3eae94f9f509c767b45932f1ada8350c4bdb85af2fcab4a3c14807981" +[[package]] +name = "libpulse-binding" +version = "2.30.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "909eb3049e16e373680fe65afe6e2a722ace06b671250cc4849557bc57d6a397" +dependencies = [ + "bitflags 2.13.0", + "libc", + "libpulse-sys", + "num-derive", + "num-traits", + "winapi", +] + +[[package]] +name = "libpulse-sys" +version = "1.23.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d74371848b22e989f829cc1621d2ebd74960711557d8b45cfe740f60d0a05e61" +dependencies = [ + "libc", + "num-derive", + "num-traits", + "pkg-config", + "winapi", +] + [[package]] name = "libredox" version = "0.1.18" @@ -3539,6 +3566,17 @@ version = "0.2.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "521739c6d2bac4aa25192232afe6841231376b2b26d4d9fae5ecf8ca5772e441" +[[package]] +name = "num-derive" +version = "0.4.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ed3955f1a9c7c0c15e092f9c887db08b1fc683305fdf6eb6684f22555355e202" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] + [[package]] name = "num-traits" version = "0.2.19" @@ -4163,6 +4201,7 @@ dependencies = [ "iroh", "iroh-tickets", "ksni", + "libpulse-binding", "nix 0.30.1", "notify-rust", "pipewire", diff --git a/Cargo.toml b/Cargo.toml index 7f94156..7632e8d 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -46,6 +46,16 @@ serde_json = "1" directories = "5" ashpd = { version = "0.9", default-features = false, features = ["tokio"] } pipewire = "0.9" +# `--repair` reads and unloads Pulse modules through libpulse introspection rather +# than by parsing `pactl` output. `pa_module_info` carries index, name and the exact +# argument in one record, and `pa_context_is_local()` answers whether the server we +# actually reached is local — neither of which the text listings can do (an argument +# may contain tabs and newlines that the short format cannot escape, the JSON +# listing carries no module index at all, and `PULSE_SERVER` is a fallback list, so +# it never proved locality). Vetted at 2.30.1: MIT/Apache-2.0, no build script +# beyond a pkg-config probe, no network or subprocess use, and all three historical +# RustSec advisories (2018-0020, 2018-0021, 2019-0038) fixed by 2.6.0. +libpulse-binding = "2.30" x11rb = { version = "0.13", default-features = false, features = ["allow-unsafe-code"] } uuid = { version = "1", features = ["v4"] } iroh-tickets = "1.0.0" diff --git a/src/cli.rs b/src/cli.rs index aa7e4e8..680b9bc 100644 --- a/src/cli.rs +++ b/src/cli.rs @@ -105,6 +105,18 @@ pub struct Cli { #[arg(long)] pub repair: bool, + /// With `--repair`: also clean up modules that carry no ownership token, + /// judging them by process id alone. + /// + /// Modules loaded by pixelpass versions before ownership tokens existed cannot + /// be attributed to a machine, boot or pid namespace, so `--repair` refuses them + /// by default: a process id means different processes in different namespaces, + /// and acting on the wrong one unloads a *running* host's audio. Use this only + /// on the machine that ran the crashed host, and only when the reported + /// candidates look right. + #[arg(long, requires = "repair")] + pub repair_legacy_untagged: bool, + /// Print an environment diagnostic report (display server, capture/encode /// dependencies, VA-API H.264 support, viewer player, relay reachability), /// then exit. Use this to check a machine can host or view before a real diff --git a/src/host/audio.rs b/src/host/audio.rs index 615538e..a2d3030 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::{self as repair_plan, Shape}; /// Owns the pactl-loaded modules plus, when filtering is active, the /// libpipewire stream-router thread. Drop unloads modules as a backstop; @@ -64,9 +65,14 @@ 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}")]) + // Every module this host loads carries an ownership token, minted per + // load, so `--repair` can tell whose pid the name refers to instead of + // assuming the number means the same thing everywhere. Without it a repair + // run in another pid namespace can unload a live host's audio; see + // `repair::plan::OwnerToken`. + let sink_module = load_module(Shape::LegacyCaptureSink, pid) .context("failed to load module-null-sink")?; // In strict per-app mode we never mirror the default sink: the viewer @@ -82,13 +88,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(Shape::LoopbackIntoCapture, pid) + .context("failed to load module-loopback (null-sink cleaned up on Drop)")?, ) }; @@ -115,7 +116,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 +137,7 @@ 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(&[ - "module-loopback", - &format!("source={sink_name_for_task}.monitor"), - "sink=@DEFAULT_SINK@", - "latency_msec=20", - ]) { + match load_module(Shape::LoopbackOutOfCapture, pid) { Ok(id) => { tracing::info!( module = id, @@ -195,12 +190,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(Shape::LoopbackIntoCapture, pid) { Ok(id) => { *loopback_for_task.lock().unwrap() = Some(id); } @@ -349,10 +339,47 @@ struct SinkInputProperties { // pactl module helpers // ────────────────────────────────────────────────────────────────────── -fn load_module(args: &[&str]) -> Result { +/// Mint an ownership token for one module load. +/// +/// **Per load, not per session.** The nonce is what makes two loads by the same pid +/// render different arguments, which is what lets a fingerprint tell a module from +/// its replacement at the same index. A token minted once and reused for every +/// reload would be a host-session nonce and would not do that, so the counter is +/// bumped on every call and mixed with the clock. +fn owner_token(pid: u32) -> Result { + use std::sync::atomic::{AtomicU64, Ordering}; + static LOADS: AtomicU64 = AtomicU64::new(0); + + let local = crate::repair::local_identity()?; + // A nonce only has to be unlikely to repeat, not unguessable. The counter makes + // two loads within the same clock tick distinct; the clock keeps two runs of the + // same process distinct. + let counter = LOADS.fetch_add(1, Ordering::Relaxed); + let nanos = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map(|d| d.as_nanos() as u64) + .unwrap_or(0); + Ok(repair_plan::OwnerToken { + machine: local.machine, + boot: local.boot, + pid_ns: local.pid_ns, + nonce: nanos ^ (counter << 48) ^ (u64::from(pid) << 32), + }) +} + +/// Load the Pulse module for one [`Shape`] and return its index. +/// +/// Both the module name and its arguments come from the shape itself +/// ([`crate::repair::plan::Shape`]) 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 moved only one of them would leave repair +/// silently unable to recognise the modules this build loads. +fn load_module(shape: Shape, pid: u32) -> Result { + let owner = owner_token(pid).context("could not build an audio ownership token")?; let output = Command::new("pactl") .arg("load-module") - .args(args) + .arg(shape.module_name()) + .args(shape.render_args(pid, Some(&owner))) .output() .context("failed to run pactl load-module")?; if !output.status.success() { diff --git a/src/host/taint/tests.rs b/src/host/taint/tests.rs index b70da24..afc854e 100644 --- a/src/host/taint/tests.rs +++ b/src/host/taint/tests.rs @@ -2683,5 +2683,12 @@ fn an_ambiguous_client_id_remembers_every_claimant_for_stickiness() { client_members >= 2, "both ambiguous-id clients should be remembered: {sticky:#?}" ); - let _ = (ClientSnapshot { serial: Serial(0), id: GlobalId(0), sec_pid: None }, firefox); + let _ = ( + ClientSnapshot { + serial: Serial(0), + id: GlobalId(0), + sec_pid: None, + }, + firefox, + ); } diff --git a/src/main.rs b/src/main.rs index b01fddf..1b9ac65 100644 --- a/src/main.rs +++ b/src/main.rs @@ -49,7 +49,7 @@ async fn main() -> Result<()> { pipewire::init(); if cli.repair { - return repair::run().await; + return repair::run(cli.repair_legacy_untagged).await; } // Read-only diagnostic: observe the graph, report what the audio-exclusion diff --git a/src/repair.rs b/src/repair.rs deleted file mode 100644 index fc92e04..0000000 --- a/src/repair.rs +++ /dev/null @@ -1,236 +0,0 @@ -//! `--repair`: clean up null-sinks and loopbacks left behind by a crashed -//! pixelpass host. Identifies orphans by the `pixelpass_capture_` -//! 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 = Vec::new(); - let mut dead_pids: HashSet = 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::() 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 = 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_` — the default→null - // mirror) or as its source (`source=pixelpass_capture_.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> { - 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::() 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 a loopback references, whether the capture -/// sink is its destination (`sink=pixelpass_capture_`) or its source -/// (`source=pixelpass_capture_.monitor`). `None` for unrelated loopbacks. -fn loopback_capture_pid(args: &str) -> Option { - 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::().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 - ); - } -} diff --git a/src/repair/introspect.rs b/src/repair/introspect.rs new file mode 100644 index 0000000..16330c4 --- /dev/null +++ b/src/repair/introspect.rs @@ -0,0 +1,297 @@ +//! Structured Pulse introspection: the observation and destruction layer for +//! `--repair`. +//! +//! # Why this replaced parsing `pactl` +//! +//! Three separate defects, all of them consequences of reading a human-oriented +//! text format rather than the protocol: +//! +//! 1. **Record boundaries were unprovable.** `pactl list short modules` prints a +//! module's argument raw into a tab-and-newline-delimited format with no +//! escaping. A *genuine* module whose argument contains a newline — say +//! `…latency_msec=20\nremix=false`, and `remix` is a real loopback option — +//! renders a first line that reads byte-exactly like one of our canonical +//! forms, with the remainder dropped as an unparseable continuation. No index +//! is forged, so no duplicate-index check can see it: repair would classify and +//! unload a module it had never actually seen in full. A tab in the same +//! position instead hides a sink reference, which is worse, because the gate +//! that protects a still-referenced sink then cannot see the reference. +//! 2. **Index and argument could be mis-paired.** The one listing that carries the +//! exact argument (`-f json`) carries **no index** at all on pactl 17, and the +//! one that carries the index cannot carry the argument faithfully. Combining +//! them by position is unsound whenever module names repeat: another client +//! loading one module and unloading another between the two calls leaves the +//! counts and names aligned while every argument has shifted by one. +//! 3. **Locality was a guess.** `PULSE_SERVER` is a *fallback list*, so +//! `unix:/missing tcp:remote:4713` passes any "starts with unix:" test and then +//! connects to another machine — where our local pids mean nothing and a live +//! remote host's modules look dead. +//! +//! `pa_module_info` carries index, name and argument together in one structured +//! record, so (1) and (2) cannot arise. `pa_context_is_local()` answers (3) about +//! the connection that actually got established rather than about a string we +//! hoped described it. And because unloading goes back through the *same* +//! connection, there is no window in which listing and destruction could disagree +//! about which server they are talking to. +//! +//! # What is deliberately not here +//! +//! No decisions. This module observes and destroys; every judgement about what may +//! be destroyed lives in [`super::plan`], which is pure and needs no Pulse server +//! to test. The one policy this layer owns is *refusing to talk to the wrong +//! server at all*. + +use anyhow::{Context as _, Result, bail}; +use libpulse_binding::callbacks::ListResult; +use libpulse_binding::context::{Context, FlagSet as ContextFlagSet, State as ContextState}; +use libpulse_binding::mainloop::standard::{IterateResult, Mainloop}; +use libpulse_binding::operation::{Operation, State as OperationState}; +use libpulse_binding::proplist::{Proplist, properties}; +use std::cell::RefCell; +use std::rc::Rc; +use std::time::{Duration, Instant}; + +use super::plan::ModuleObservation; + +/// How long to wait for the connection to reach `Ready`. A one-shot CLI must not +/// hang on an unresponsive server; failing closed here costs the user a re-run. +const CONNECT_BUDGET: Duration = Duration::from_secs(3); + +/// How long any single introspection request may take. +/// +/// ⚠️ On timeout the `Operation` wrapper is dropped while still running. In +/// libpulse-binding 2.30.1 that only unrefs the C operation — the boxed callback +/// and the `Rc`s it captured leak until the context cancels the operation at +/// disconnect. That is bounded and harmless *here*, because `--repair` is a +/// one-shot process that exits immediately afterwards, and it cannot become a +/// use-after-free (the closure owns its clones, and the context clears callbacks +/// before the mainloop is touched). **It would not be acceptable in the long-lived +/// host**, so this module must not be reused for host-side loading until that +/// binding bug is fixed or worked around; `op.cancel()` does not help. +const REQUEST_BUDGET: Duration = Duration::from_secs(3); + +/// How long to sleep between mainloop iterations while waiting. Non-blocking +/// iteration plus a short sleep keeps the deadline enforceable, which +/// `iterate(true)` would not. +const POLL_INTERVAL: Duration = Duration::from_millis(2); + +/// A live, verified-local connection to the Pulse server. +/// +/// Both listing and unloading run through this one connection, so everything +/// repair sees and everything it destroys provably belong to the same server. +/// +/// ⚠️ **Field order is load-bearing, and this was not theoretical.** Rust drops +/// fields in declaration order, and the context's teardown frees IO events that +/// live *in* the mainloop. With `mainloop` declared first, `--repair` did its work +/// correctly and then died on the way out: +/// +/// ```text +/// Assertion '!e->dead' failed at ../pulseaudio/src/pulse/mainloop.c:207, +/// function mainloop_io_free(). Aborting. +/// ``` +/// +/// SIGABRT, a core dump, and exit 134 — so a completely successful repair reported +/// failure to its caller. This is the same invariant phase 0b's +/// `ScreenshareTeardown` exists for, met again one layer down. +/// +/// Rather than leave that resting on where the fields happen to be written, [`Drop`] +/// **explicitly** takes and drops the context first, so the ordering survives a +/// future reorder of this struct. The declaration order below is still correct, and +/// now it is also not load-bearing. +pub struct PulseSession { + /// `Option` only so that `Drop` can `take()` it and destroy it *before* the + /// mainloop. Always `Some` for the whole of the session's usable life. + context: Option, + mainloop: Mainloop, +} + +impl Drop for PulseSession { + fn drop(&mut self) { + // Disconnect, then destroy the context while the mainloop it registered its + // IO events with is still alive. The mainloop then drops after us. + // + // Nothing is drained afterwards on purpose. An earlier version iterated the + // mainloop a few times here to "let teardown settle", which was a ritual + // rather than a barrier: a fixed number of non-blocking polls cannot + // guarantee that any particular event became ready. It is also unnecessary — + // PulseAudio's context unlink cancels outstanding operations and removes the + // context's socket machinery synchronously, so by the time `drop(context)` + // returns there is no obligation left for the mainloop to service. + if let Some(mut context) = self.context.take() { + context.disconnect(); + drop(context); + } + } +} + +impl PulseSession { + /// The live context. Infallible in practice: only `Drop` ever clears it, and + /// nothing can call this afterwards. + fn context(&mut self) -> &mut Context { + self.context + .as_mut() + .expect("the context is only taken during Drop") + } + /// Connect, wait for readiness, and refuse anything but a local server. + pub fn connect() -> Result { + let mut proplist = Proplist::new().context("could not allocate a Pulse proplist")?; + // `set_str` fails only on an invalid key, and these keys are constants. + let _ = proplist.set_str(properties::APPLICATION_NAME, "pixelpass --repair"); + let _ = proplist.set_str(properties::APPLICATION_ID, "xyz.pixelpass.repair"); + + let mut mainloop = Mainloop::new().context("could not create a Pulse mainloop")?; + let mut context = Context::new_with_proplist(&mainloop, "pixelpass --repair", &proplist) + .context("could not create a Pulse context")?; + context + .connect(None, ContextFlagSet::NOFLAGS, None) + .context("could not connect to the Pulse server")?; + + let deadline = Instant::now() + CONNECT_BUDGET; + loop { + iterate_once(&mut mainloop)?; + match context.get_state() { + ContextState::Ready => break, + ContextState::Failed => { + bail!("the Pulse server refused the connection"); + } + ContextState::Terminated => { + bail!("the Pulse connection terminated before it was ready"); + } + _ => { + if Instant::now() >= deadline { + bail!( + "the Pulse server did not become ready within {:?}", + CONNECT_BUDGET + ); + } + std::thread::sleep(POLL_INTERVAL); + } + } + } + + // The connection is established, so this is about the server we actually + // reached — not about what a server string appeared to promise. A remote + // server's module table belongs to another machine's processes, where our + // pids mean nothing, so repair must not touch it. + match context.is_local() { + Some(true) => {} + Some(false) => bail!( + "connected to a REMOTE Pulse server; --repair only ever operates on the local \ + server, because it decides what to unload from local process liveness" + ), + None => bail!( + "could not determine whether the Pulse server is local; refusing to unload \ + anything" + ), + } + + Ok(Self { + context: Some(context), + mainloop, + }) + } + + /// Every loaded module, with its exact argument. + pub fn list_modules(&mut self) -> Result> { + // `Rc>` because the callback is owned by the C library and may + // be invoked many times before the operation completes. + let collected: Rc>> = Rc::new(RefCell::new(Vec::new())); + let failed: Rc> = Rc::new(RefCell::new(false)); + + let sink = Rc::clone(&collected); + let error_flag = Rc::clone(&failed); + let op = self + .context() + .introspect() + .get_module_info_list(move |result| match result { + ListResult::Item(info) => { + // A module with no name is not one we can identify, and an + // argumentless module is simply one loaded without arguments. + let name = info.name.as_deref().unwrap_or_default(); + let args = info.argument.as_deref().unwrap_or_default(); + sink.borrow_mut() + .push(ModuleObservation::new(info.index, name, args)); + } + ListResult::End => {} + ListResult::Error => *error_flag.borrow_mut() = true, + }); + + self.run_to_completion(op, "list modules")?; + if *failed.borrow() { + bail!("the Pulse server returned an error while listing modules"); + } + + let modules = collected.borrow().clone(); + // Impossible per the protocol — an index identifies one module — so this is + // a sanity check on external input, not a safety boundary. It fails closed + // because an ambiguous index is one we could unload wrongly. + for (i, module) in modules.iter().enumerate() { + if modules[..i].iter().any(|earlier| earlier.id == module.id) { + bail!( + "the Pulse server reported module index #{} twice; refusing to unload \ + anything", + module.id + ); + } + } + Ok(modules) + } + + /// Unload one module, over the same connection it was observed on. + pub fn unload_module(&mut self, index: u32) -> Result<()> { + let succeeded: Rc>> = Rc::new(RefCell::new(None)); + let outcome = Rc::clone(&succeeded); + let op = self + .context() + .introspect() + .unload_module(index, move |success| *outcome.borrow_mut() = Some(success)); + + self.run_to_completion(op, "unload module")?; + match *succeeded.borrow() { + Some(true) => Ok(()), + Some(false) => bail!("the Pulse server rejected unloading module #{index}"), + // The operation completed without the callback running, which we cannot + // read as success. + None => bail!("no result was reported for unloading module #{index}"), + } + } + + /// Drive the mainloop until `op` finishes, or the budget expires. + fn run_to_completion(&mut self, op: Operation, what: &str) -> Result<()> { + let deadline = Instant::now() + REQUEST_BUDGET; + loop { + iterate_once(&mut self.mainloop)?; + match op.get_state() { + OperationState::Done => return Ok(()), + OperationState::Cancelled => { + bail!("the Pulse server cancelled the request to {what}"); + } + OperationState::Running => { + // A connection that dies mid-request would otherwise be waited + // out to the full budget. + match self.context().get_state() { + ContextState::Ready => {} + state => { + bail!("the Pulse connection became {state:?} while trying to {what}") + } + } + if Instant::now() >= deadline { + bail!("the Pulse server did not {what} within {REQUEST_BUDGET:?}"); + } + std::thread::sleep(POLL_INTERVAL); + } + } + } + } +} + +/// One non-blocking mainloop iteration, with quit and error surfaced as errors. +fn iterate_once(mainloop: &mut Mainloop) -> Result<()> { + match mainloop.iterate(false) { + IterateResult::Success(_) => Ok(()), + IterateResult::Quit(code) => { + bail!("the Pulse mainloop quit unexpectedly (code {})", code.0) + } + IterateResult::Err(e) => Err(e).context("the Pulse mainloop failed"), + } +} diff --git a/src/repair/mod.rs b/src/repair/mod.rs new file mode 100644 index 0000000..fa8fe77 --- /dev/null +++ b/src/repair/mod.rs @@ -0,0 +1,553 @@ +//! `--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 the two rules 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. +//! - **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 +//! +//! Structured Pulse introspection over one verified-local connection — see +//! [`introspect`], which also documents the three defects that parsing `pactl`'s +//! text output turned out to have. Listing *and* unloading both go through that +//! same connection. + +pub mod introspect; +pub mod plan; + +use anyhow::{Context, Result, bail}; +use std::path::Path; + +use introspect::PulseSession; +use plan::{Fingerprint, Liveness, Shape}; + +pub async fn run(clean_untagged: bool) -> Result<()> { + let liveness = LivenessProbe::new(); + if let Some(reason) = liveness.degraded_reason() { + // Scoped deliberately: modules carrying a token that matches this machine, + // boot and pid namespace are still cleaned, because the token establishes + // what these signals can only guess at. Saying "refusing to unload + // anything" here would be false in exactly the container-recovery case the + // token was added for. + eprintln!( + "[pixelpass] --repair: cannot independently determine process liveness \ + ({reason}); modules WITHOUT an ownership token will be left alone." + ); + } + + let local = local_identity().context("could not establish this process's own identity")?; + let policy = plan::Policy { + local, + untagged: if clean_untagged { + plan::UntaggedPolicy::CleanByPidAlone + } else { + plan::UntaggedPolicy::Refuse + }, + }; + if clean_untagged { + eprintln!( + "[pixelpass] --repair: --repair-legacy-untagged given; untagged modules will be \ + judged by process id ALONE. That is only safe on the machine and in the pid \ + namespace that ran the crashed host." + ); + } + + let mut pulse = PulseSession::connect().context("could not observe the Pulse module table")?; + let modules = pulse + .list_modules() + .context("could not list Pulse 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, &policy, |pid, attribution| { + liveness_for(&liveness, attribution, pid) + }); + + // Ours by shape, but carrying no proof of whose pid they name. Never unloaded by + // default — listed, so an explicit legacy run has something to look at first. + if !planned.untagged.is_empty() { + eprintln!( + "[pixelpass] --repair: {} module(s) are pixelpass's but carry no ownership token, so \ + the process id in their name cannot be attributed to this machine or pid namespace. \ + LEFT ALONE. Re-run with --repair-legacy-untagged to clean them by pid alone:", + planned.untagged.len() + ); + for fp in &planned.untagged { + eprintln!( + "[pixelpass] --repair: #{} {} (claims pid {})", + fp.id, + fp.shape.label(), + fp.pid + ); + } + } + // Tokened, but the token belongs to another machine, boot or namespace. + if !planned.foreign.is_empty() { + eprintln!( + "[pixelpass] --repair: {} module(s) belong to another machine, boot or pid namespace; \ + their process ids mean nothing here. LEFT ALONE:", + planned.foreign.len() + ); + for fp in &planned.foreign { + eprintln!( + "[pixelpass] --repair: #{} {} (claims pid {})", + fp.id, + fp.shape.label(), + fp.pid + ); + } + } + + if planned.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 ({} left alone).", + held.join(", ") + ); + } + return Ok(()); + } + + let mut unloaded = 0u32; + let mut skipped = 0u32; + let mut failed = 0u32; + + for fp in &planned.unload { + // ORDER MATTERS, and it is the opposite of what reads naturally. + // + // Liveness is asked FIRST, and the fresh snapshot is taken AFTER it. The + // tempting order — verify the module, then check liveness, then unload — + // leaves the dangerous window wide open: between `kill` returning ESRCH and + // the unload, this process can be descheduled long enough for the planned + // module to vanish, a new host to inherit both the pid and the module index, + // and its differently-nonced arguments to occupy that index. Nothing would + // re-read those arguments, so the reused index gets unloaded. + // + // Asking liveness first and re-verifying the fingerprint after it means a + // replacement arriving in that window is caught by the argument comparison, + // and only the irreducible snapshot-to-unload interval remains. + match attributed_liveness(&liveness, fp) { + 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; + } + } + + // Fresh snapshot per action, taken after the liveness answer. 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*. + let current = pulse + .list_modules() + .context("could not re-list Pulse 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; + } + // 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; + } + + match pulse.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 !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 { + // 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)"); + } + println!("[pixelpass] --repair: cleaned up {unloaded} module(s)."); + Ok(()) +} + +fn describe(fp: &Fingerprint) -> String { + format!( + "unloaded {} #{} (orphaned from pid {})", + fp.shape.label(), + fp.id, + fp.pid + ) +} + +/// Ask liveness with the module's *attribution* in hand. +/// +/// A module whose token matches this machine, boot and pid namespace has already +/// proven that its pid is a number meaningful here — that is the token's entire +/// job. Running such a module through the probe's degradation checks would defeat +/// it in exactly the situation it exists for: a host crashing inside a container +/// leaves a token that matches perfectly, while a container marker or a multi-entry +/// `NSpid` makes the probe answer `Unknown` for everything, so token-qualified +/// repair would do nothing precisely where it is now safe. +/// +/// The degradation signals therefore guard only the *untagged* path, where a bare +/// pid is all there is and those signals are the only protection left. +fn liveness_for(probe: &LivenessProbe, attribution: plan::Attribution, pid: u32) -> Liveness { + match attribution { + plan::Attribution::Tokened => probe.of_attributed(pid), + plan::Attribution::Untagged => probe.of(pid), + } +} + +/// The same rule, for a fingerprint at execution time. +fn attributed_liveness(probe: &LivenessProbe, fp: &Fingerprint) -> Liveness { + let attribution = match fp.owner { + Some(_) => plan::Attribution::Tokened, + None => plan::Attribution::Untagged, + }; + liveness_for(probe, attribution, fp.pid) +} + +// ────────────────────────────────────────────────────────────────────── +// Identity +// ────────────────────────────────────────────────────────────────────── + +/// This process's machine, boot and pid-namespace identity. +/// +/// Read from the kernel and the system, never guessed: without all three, a token +/// cannot be compared and no module can be attributed. Dashes are stripped so every +/// component is safe inside a single unquoted Pulse property value. +pub fn local_identity() -> Result { + let machine = read_identity_file("/etc/machine-id") + .or_else(|_| read_identity_file("/var/lib/dbus/machine-id")) + .context("could not read a machine id")?; + let boot = read_identity_file("/proc/sys/kernel/random/boot_id") + .context("could not read the boot id")?; + let pid_ns = pid_namespace_id().context("could not read this process's pid namespace")?; + Ok(plan::LocalIdentity { + machine, + boot, + pid_ns, + }) +} + +fn read_identity_file(path: &str) -> Result { + let raw = std::fs::read_to_string(path).with_context(|| format!("could not read {path}"))?; + let cleaned: String = raw + .trim() + .chars() + .filter(|c| c.is_ascii_hexdigit()) + .collect(); + if cleaned.is_empty() { + bail!("{path} held no usable identity"); + } + Ok(cleaned) +} + +/// The inode of `/proc/self/ns/pid` — the kernel's identity for a pid namespace. +/// +/// This is the value that makes a pid meaningful: two processes in different pid +/// namespaces can hold the same number, and only this distinguishes them. +fn pid_namespace_id() -> Result { + use std::os::unix::fs::MetadataExt; + let meta = + std::fs::metadata("/proc/self/ns/pid").context("could not stat /proc/self/ns/pid")?; + Ok(meta.ino()) +} + +// ────────────────────────────────────────────────────────────────────── +// Liveness +// ────────────────────────────────────────────────────────────────────── + +/// Answers "is this pid still around", and knows when it must refuse to answer. +/// +/// **An error is not an absence.** `Path::exists()` folds permission errors and a +/// missing `/proc` into `false`, which here would read as "dead, go ahead and +/// unload". Liveness is asked with `kill(pid, 0)` instead, where `EPERM` *proves* +/// existence. +/// +/// # The limit of what this can prove, stated rather than papered over +/// +/// A pid can be alive and invisible. Inside a pid namespace — a container, a +/// distrobox — `/proc/self` is perfectly visible while every process in the +/// *parent* namespace is not, and `hidepid` has the same self-visible, +/// others-invisible shape. Repair in such a place can reach the host's Pulse +/// socket, see a live host's modules, get `ESRCH` for its pid and unload a running +/// host's audio. +/// +/// The signals below are **negative** ones: they detect *some* cases where pid +/// numbers cannot be trusted, and every one of them fails closed. What they cannot +/// do is prove the converse. `NSpid` reports this process's pid in each namespace +/// that its procfs can see, and its leftmost value is relative to the pid namespace +/// that mounted that procfs — so a nested namespace with its own `/proc` reports a +/// single entry quite legitimately. `NSpid > 1` therefore means "definitely +/// nested", while `NSpid == 1` means only "not detectably nested". +/// +/// Closing that properly needs the module itself to carry an owner token (machine +/// and boot identity plus pid-namespace identity) written at load time, with +/// token-less modules treated as `Unknown`. That changes what pixelpass writes into +/// the graph and how far back `--repair` can clean up, so it is a design decision +/// recorded in the impl plan rather than guessed at here. +struct LivenessProbe { + /// `None` when no signal says pid numbers are untrustworthy; `Some(reason)` + /// when every answer must be [`Liveness::Unknown`]. + degraded: Option, +} + +impl LivenessProbe { + fn new() -> Self { + Self { + degraded: Self::detect_degradation(), + } + } + + fn detect_degradation() -> Option { + // Locality is deliberately NOT checked here. `PULSE_SERVER` is a fallback + // *list*, so `unix:/missing tcp:remote:4713` starts with "unix:" and still + // connects to another machine, and a remote server can be selected by client + // configuration with the variable unset entirely. The authoritative answer + // comes from `pa_context_is_local()` on the connection that actually got + // established — see `introspect::PulseSession::connect`. + match std::fs::read_to_string("/proc/self/status") { + Ok(status) => { + let nspid = status + .lines() + .find_map(|line| line.strip_prefix("NSpid:")) + .map(|rest| rest.split_whitespace().count()); + match nspid { + Some(n) if n > 1 => { + return Some(format!( + "this process is in a nested pid namespace (NSpid has {n} entries), \ + so pids in module names may belong to processes it cannot see" + )); + } + // NB: a single entry is not proof of the initial namespace — see + // the type's doc comment. It only means nothing detected it. + // A kernel too old to report NSpid cannot rule nesting out. + None => { + return Some( + "/proc/self/status does not report NSpid, so pid-namespace identity \ + cannot be established" + .to_string(), + ); + } + Some(_) => {} + } + } + Err(e) => return Some(format!("/proc/self/status could not be read: {e}")), + } + // Belt and braces: container runtimes that leave a marker. + for marker in ["/run/.containerenv", "/.dockerenv"] { + if Path::new(marker).try_exists().unwrap_or(false) { + return Some(format!("{marker} exists, so this is a container")); + } + } + None + } + + fn degraded_reason(&self) -> Option<&str> { + self.degraded.as_deref() + } + + /// Liveness for a pid this process has **no** independent reason to trust — + /// an untagged module. Here the degradation signals are the only protection. + fn of(&self, pid: u32) -> Liveness { + if self.degraded.is_some() { + return Liveness::Unknown; + } + self.of_attributed(pid) + } + + /// Liveness for a pid already proven to belong to this machine, boot and pid + /// namespace by an [`plan::OwnerToken`]. + /// + /// The degradation checks are deliberately skipped: they exist to guess at + /// whether a bare pid is meaningful, and here that is not a guess any more. + fn of_attributed(&self, pid: u32) -> Liveness { + // `kill(0, …)` signals our whole process group and a negative pid signals + // another group, so neither may ever reach `kill`. Neither is a pid we + // could have written into a sink name anyway. + if pid == 0 || pid > i32::MAX as u32 { + return Liveness::Unknown; + } + match nix::sys::signal::kill(nix::unistd::Pid::from_raw(pid as i32), None) { + Ok(()) => Liveness::Alive, + // The process exists; we merely may not signal it. + Err(nix::errno::Errno::EPERM) => Liveness::Alive, + Err(nix::errno::Errno::ESRCH) => Liveness::Dead, + Err(_) => Liveness::Unknown, + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + // The `pactl`-text parser that used to live here is gone, and so are its + // tests: `introspect` gets index, name and argument as structured fields, so + // there is no format left to mis-parse. What replaced those tests is the live + // field gate, since the remaining risk is in talking to the server, which no + // unit test can exercise. The decisions all live in `plan`, which is pure and + // tested there. + + /// The probe must never say `Dead` when it cannot see the whole pid space, and + /// must never ask `kill` about a pid that would signal something other than one + /// process. + #[test] + fn a_degraded_probe_never_reports_dead() { + let degraded = LivenessProbe { + degraded: Some("test".to_string()), + }; + assert_eq!(degraded.of(1), Liveness::Unknown); + assert_eq!(degraded.of(u32::MAX), Liveness::Unknown); + + let probe = LivenessProbe::new(); + // Our own pid is alive by construction — unless this test itself runs + // somewhere the probe must abstain, which is exactly the other branch. + let me = std::process::id(); + match probe.degraded_reason() { + None => assert_eq!(probe.of(me), Liveness::Alive), + Some(_) => assert_eq!(probe.of(me), Liveness::Unknown), + } + // `kill(0, …)` would signal our whole process group, and a pid past + // `i32::MAX` cannot be expressed to `kill` at all. + assert_eq!(probe.of(0), Liveness::Unknown); + assert_eq!(probe.of(u32::MAX), Liveness::Unknown); + } + + /// A token proves the pid is meaningful here, so the probe's namespace + /// guesswork must not veto it. Without this, a host crashing inside a + /// container leaves a perfectly matching token while a container marker makes + /// every answer `Unknown` — and token-qualified repair does nothing in exactly + /// the situation the token was built for. + /// + /// This runs against a *degraded* probe deliberately: on an ordinary desktop + /// the two paths agree, so a test using the real probe's state would pass + /// whether or not the distinction exists. + #[test] + fn a_token_beats_the_degradation_signals_but_a_bare_pid_does_not() { + let degraded = LivenessProbe { + degraded: Some("pretending to be in a container".to_string()), + }; + let me = std::process::id(); + + assert_eq!( + degraded.of_attributed(me), + Liveness::Alive, + "an attributed pid must still be answered when the probe is degraded" + ); + assert_eq!( + degraded.of(me), + Liveness::Unknown, + "a bare pid must not be, since the signals are all it has" + ); + + // And the routing between them, which is what the caller actually uses. + assert_eq!( + liveness_for(°raded, plan::Attribution::Tokened, me), + Liveness::Alive + ); + assert_eq!( + liveness_for(°raded, plan::Attribution::Untagged, me), + Liveness::Unknown + ); + } +} diff --git a/src/repair/plan.rs b/src/repair/plan.rs new file mode 100644 index 0000000..05c855d --- /dev/null +++ b/src/repair/plan.rs @@ -0,0 +1,1470 @@ +//! 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 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 +//! +//! 0. **A pid is not an owner.** The same number is a different process in a +//! different pid namespace, so before liveness can even be *asked*, the module +//! must prove which machine, boot and namespace its pid belongs to — see +//! [`OwnerToken`]. Modules that cannot be attributed are never touched, and +//! their pids are never even looked up: asking is the bug, because the answer +//! would be meaningless. This rule comes first because it gates the others. +//! 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. **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. +//! 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. **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* +//! renderer the loader uses — [`Shape::render_args`], which owns the module name +//! too. 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. +//! +//! The template is only a pre-filter, too. [`classify`] re-renders the pid it +//! extracted and demands byte equality, so the renderer — not a derived pair of +//! strings — is always the authority on what one of our modules looks like. +//! +//! 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}; + +/// Every pixelpass capture sink is named `pixelpass_capture_`; the pid in +/// that name is the only owner identity these modules carry. +pub const SINK_NAME_PREFIX: &str = "pixelpass_capture_"; + +/// 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}") +} + +/// 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** argument string from `pa_module_info`, not a normalised +/// one, and not a reconstruction from a text listing. Normalising would only make +/// two genuinely different arguments compare equal — whitespace inside a quoted +/// property value is not layout — and every snapshot within one invocation comes +/// from the same connection, so there is no re-rendering to absorb. +#[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(), + } + } +} + +/// Whether the owner of a module is still around. +/// +/// Three states, not two, because "I cannot see that process" and "that process +/// does not exist" are different answers and only one of them permits destroying +/// anything. A pid can be alive and invisible: inside a pid namespace every +/// process in the parent namespace is, and `hidepid` hides others while leaving +/// `self` visible. Collapsing those into "absent" would point the wrong way. +#[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_ + /// latency_msec=20` — the default-sink mirror, so the viewer hears system + /// audio. + LoopbackIntoCapture, + /// `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 + /// 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, +} + +/// Every shape, in unload order. +pub const ALL_SHAPES: [Shape; 3] = [ + Shape::LoopbackIntoCapture, + Shape::LoopbackOutOfCapture, + Shape::LegacyCaptureSink, +]; + +// ────────────────────────────────────────────────────────────────────── +// Ownership +// ────────────────────────────────────────────────────────────────────── + +/// The Pulse property every pixelpass module carries its owner token in. +/// +/// Measured on the live server before being relied on: all three shapes accept a +/// property-list argument (`sink_properties` / `sink_input_properties` / +/// `source_output_properties`), the recorded argument comes back byte-identical, and +/// the property really does land on the resulting sink, sink-input and +/// source-output. +pub const OWNER_PROPERTY: &str = "pixelpass.owner"; + +/// Bumped if the token's shape ever changes. A token this build cannot parse is +/// **not** treated as ours, so an older `--repair` meeting a newer token refuses it +/// and reports it rather than guessing. +pub const OWNER_TOKEN_VERSION: u32 = 1; + +/// Proof of *which* machine, boot and pid namespace a module's pid refers to. +/// +/// # Why a pid alone is not ownership +/// +/// The pid in `pixelpass_capture_` is only a number, and a number means +/// different processes in different pid namespaces. Repair running inside a +/// container that can reach the host's Pulse socket sees a live host's modules, +/// asks about that pid in *its own* namespace, is told nothing exists, and unloads +/// a running host's audio. No negative signal closes that: `NSpid == 1` does not +/// prove the initial namespace, because its leftmost value is relative to whichever +/// procfs was mounted. +/// +/// So the module carries the answer with it. If the token's machine, boot and pid +/// namespace all match ours, then its pid is a number we can meaningfully ask +/// about. Otherwise the only safe verdict is [`Liveness::Unknown`]. +/// +/// The `nonce` is per-load, and it narrows the residual ABA window as a side +/// effect: two loads by the same pid no longer render byte-identical arguments, so +/// a module that vanished and a replacement that took its index are distinguishable. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct OwnerToken { + /// `/etc/machine-id`, dash-free. + pub machine: String, + /// `/proc/sys/kernel/random/boot_id`, dash-free — a fresh value per boot. + pub boot: String, + /// Inode of `/proc/self/ns/pid`: which pid namespace the pid belongs to. + pub pid_ns: u64, + /// Distinguishes two loads that are otherwise identical. + pub nonce: u64, +} + +impl OwnerToken { + /// `----`. + /// + /// Every component is dash-free and free of spaces and `=`, so the whole token + /// is a single unquoted Pulse property value and survives the round trip + /// through the module's recorded argument untouched. + pub fn render(&self) -> String { + format!( + "{OWNER_TOKEN_VERSION}-{}-{}-{}-{}", + self.machine, self.boot, self.pid_ns, self.nonce + ) + } + + /// Parse a token, or `None` for anything this build does not fully understand. + pub fn parse(raw: &str) -> Option { + let mut parts = raw.split('-'); + let version = parts.next()?.parse::().ok()?; + if version != OWNER_TOKEN_VERSION { + return None; + } + let machine = parts.next()?; + let boot = parts.next()?; + let pid_ns = parts.next()?.parse::().ok()?; + let nonce = parts.next()?.parse::().ok()?; + if parts.next().is_some() { + return None; + } + // Identities are hex strings; anything else is not a token we wrote. + let identity_ok = |s: &str| !s.is_empty() && s.bytes().all(|b| b.is_ascii_hexdigit()); + if !identity_ok(machine) || !identity_ok(boot) { + return None; + } + Some(Self { + machine: machine.to_string(), + boot: boot.to_string(), + pid_ns, + nonce, + }) + } +} + +/// Who *this* process is, for comparison against a module's [`OwnerToken`]. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct LocalIdentity { + pub machine: String, + pub boot: String, + pub pid_ns: u64, +} + +impl LocalIdentity { + /// Does this token describe a pid we can meaningfully ask about? + /// + /// All three must agree. A different boot means the pid space has been recycled + /// wholesale; a different machine means the token came from somewhere else + /// entirely; a different namespace means the number is not ours to interpret. + pub fn can_judge(&self, token: &OwnerToken) -> bool { + self.machine == token.machine && self.boot == token.boot && self.pid_ns == token.pid_ns + } +} + +/// Whether a module proved which pid space its pid belongs to. +/// +/// Passed to the liveness callback because the answer changes *how* the question +/// may be asked: an attributed pid needs no guessing about namespaces, while an +/// untagged one is a bare number whose meaning has to be guarded some other way. +#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)] +pub enum Attribution { + /// Carries a token matching this machine, boot and pid namespace. + Tokened, + /// Carries no token; judged by pid alone, and only under + /// [`UntaggedPolicy::CleanByPidAlone`]. + Untagged, +} + +/// What repair may do about modules that carry no token at all. +/// +/// Every module loaded before tokens existed is untagged, and there is no way to +/// establish whose pid it names. The default therefore refuses them: leaving an old +/// orphan behind is recoverable, while unloading a live host's routing is not. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum UntaggedPolicy { + /// Untagged modules are [`Liveness::Unknown`]: reported, never unloaded. + Refuse, + /// Judge untagged modules by pid liveness alone — the pre-token heuristic, + /// available only behind an explicit flag. + CleanByPidAlone, +} + +/// Everything the planner needs in order to decide ownership. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct Policy { + pub local: LocalIdentity, + pub untagged: UntaggedPolicy, +} + +/// 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 — unconditionally, so a +/// future shape that repeats the pid cannot slip through a release build. +const TEMPLATE_SENTINEL_PID: u32 = u32::MAX; + +/// The token used when deriving a template. Its rendering must not contain the pid +/// sentinel's digits, which [`Template::derive`] also asserts. +fn template_sentinel_token() -> OwnerToken { + OwnerToken { + machine: "ffffffffffffffff".to_string(), + boot: "eeeeeeeeeeeeeeee".to_string(), + pid_ns: u64::MAX, + nonce: u64::MAX - 1, + } +} + +/// An exact-match matcher for one shape, derived from that shape's own renderer. +/// +/// The template is only ever a *pre-filter*: it locates the candidate pid and token +/// cheaply, and [`classify`] then re-renders both through the real renderer and +/// demands byte equality. So the authority is always the renderer the loader uses, +/// never these derived strings. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct Template { + pub module_name: &'static str, + prefix: String, + /// Between the pid and the token. `None` for the legacy, token-less form, which + /// has only one hole. + mid: Option, + suffix: String, +} + +impl Template { + /// Split a shape's rendered arguments around its variable parts. + fn derive(shape: Shape, tokened: bool) -> Self { + let sentinel_token = template_sentinel_token(); + let token = tokened.then(|| sentinel_token.clone()); + let rendered = recorded_argument(&shape.render_args(TEMPLATE_SENTINEL_PID, token.as_ref())); + let pid_sentinel = TEMPLATE_SENTINEL_PID.to_string(); + + let (prefix, rest) = rendered + .split_once(&pid_sentinel) + .expect("a rendered shape must contain its pid"); + assert!( + !rest.contains(&pid_sentinel), + "a shape must name its pid exactly once, but {shape:?} rendered {rendered:?}" + ); + + let (mid, suffix) = match &token { + None => (None, rest.to_string()), + Some(token) => { + let token_sentinel = token.render(); + assert!( + !token_sentinel.contains(&pid_sentinel), + "the sentinel token must not contain the sentinel pid's digits" + ); + let (mid, suffix) = rest + .split_once(&token_sentinel) + .expect("a tokened shape must contain its token"); + assert!( + !suffix.contains(&token_sentinel), + "a shape must carry its token exactly once, but {shape:?} rendered \ + {rendered:?}" + ); + (Some(mid.to_string()), suffix.to_string()) + } + }; + + Self { + module_name: shape.module_name(), + prefix: prefix.to_string(), + mid, + suffix, + } + } + + /// The pid and token this argument string names, if it is *exactly* this shape. + /// + /// Total: the argument must equal `prefix ++ pid ++ mid ++ token ++ suffix` with + /// nothing left over. A canonical decimal pid 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 parse(&self, args: &str) -> Option<(u32, Option)> { + let rest = args.strip_prefix(self.prefix.as_str())?; + let (digits, token) = match &self.mid { + None => (rest.strip_suffix(self.suffix.as_str())?, None), + Some(mid) => { + let (digits, after) = rest.split_once(mid.as_str())?; + let raw = after.strip_suffix(self.suffix.as_str())?; + (digits, Some(OwnerToken::parse(raw)?)) + } + }; + if digits.is_empty() || !digits.bytes().all(|b| b.is_ascii_digit()) { + return None; + } + if digits.len() > 1 && digits.starts_with('0') { + return None; + } + Some((digits.parse::().ok()?, token)) + } +} + +impl Shape { + /// The Pulse module this shape loads. Owned here rather than written out at + /// the call site, so the module *name* is as drift-proof as the arguments — + /// the loader asks for it too. + pub fn module_name(self) -> &'static str { + match self { + Shape::LoopbackIntoCapture | Shape::LoopbackOutOfCapture => "module-loopback", + Shape::LegacyCaptureSink => "module-null-sink", + } + } + + /// Which property-list argument this shape carries its owner token in. + fn owner_property_argument(self) -> &'static str { + match self { + // The mirror's stream is a sink-input on our capture sink. + Shape::LoopbackIntoCapture => "sink_input_properties", + // The local monitor's stream is a source-output on our capture monitor. + Shape::LoopbackOutOfCapture => "source_output_properties", + // The sink itself carries the property. + Shape::LegacyCaptureSink => "sink_properties", + } + } + + /// The exact `pactl load-module` arguments for this shape, for `pid`, optionally + /// carrying an owner token. + /// + /// **This is the single source of truth.** `host/audio.rs` loads through it and + /// `--repair` matches through it, so a change here moves both at once. + /// + /// `token: None` renders the **legacy** form — what every pixelpass before + /// ownership tokens loaded. It is still rendered, because repair must be able to + /// recognise those modules in order to report them. + pub fn render_args(self, pid: u32, token: Option<&OwnerToken>) -> Vec { + let sink = sink_name_for(pid); + let mut args = match self { + // The default-sink mirror: the viewer hears system audio. + Shape::LoopbackIntoCapture => vec![ + "source=@DEFAULT_SINK@.monitor".to_string(), + format!("sink={sink}"), + format!("latency_msec={LOOPBACK_LATENCY_MSEC}"), + ], + // The local monitor: the sharer hears the app they are sharing. + Shape::LoopbackOutOfCapture => vec![ + format!("source={sink}.monitor"), + "sink=@DEFAULT_SINK@".to_string(), + format!("latency_msec={LOOPBACK_LATENCY_MSEC}"), + ], + // The legacy capture sink (pre-0c hosts only). + Shape::LegacyCaptureSink => vec![format!("sink_name={sink}")], + }; + if let Some(token) = token { + args.push(format!( + "{}={OWNER_PROPERTY}={}", + self.owner_property_argument(), + token.render() + )); + } + args + } + + /// The exact-match pre-filters for this shape, tokened form first. + /// + /// Both are generated from `render_args`, so neither can drift from the loader. + pub fn templates(self) -> [Template; 2] { + [Template::derive(self, true), Template::derive(self, false)] + } + + /// 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 narrows it is the per-load nonce in [`OwnerToken`] +/// (two loads no longer render identical arguments), and what closes the rest in +/// practice is the liveness recheck nearer the unload. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct Fingerprint { + pub id: u32, + pub module_name: String, + /// The exact recorded argument string. + pub args: String, + pub pid: u32, + pub shape: Shape, + /// The owner token, or `None` for a legacy module loaded before tokens existed. + /// A pid without one cannot be attributed to a namespace — see [`OwnerToken`]. + pub owner: Option, +} + +impl Fingerprint { + /// Does `obs` still present the same observable module this fingerprint was + /// taken from? + /// + /// 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) + } +} + +/// 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, + /// Pids that are still alive and were therefore skipped entirely. Includes + /// this process and any other running pixelpass. + /// + /// These three sets are for *reporting*. A pid claimed by both a tokened and an + /// untagged module can legitimately appear in two of them, because those are two + /// different questions; the unload decision is made per fingerprint against its + /// own attribution, never by looking a pid up in these. + 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, + /// Modules recognised as ours but carrying no owner token, while the policy + /// refuses to judge those. Reported so the user can see what an explicit + /// legacy-cleanup run would act on. + pub untagged: Vec, + /// Modules whose token belongs to another machine, boot or pid namespace. Their + /// pids are numbers this process cannot interpret, so they are never touched. + pub foreign: Vec, +} + +impl Plan { + pub fn is_empty(&self) -> bool { + self.unload.is_empty() + } +} + +/// Recognise one of pixelpass's three module shapes, or `None` for a module that +/// is not ours. +/// +/// 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. +/// +/// The template only proposes a pid; the shape's own renderer decides. Re-rendering +/// and demanding byte equality means the loader is the authority, so a shape that +/// grows an argument, changes a latency, or repeats the pid cannot leave a matcher +/// quietly accepting the old form. +pub fn classify(obs: &ModuleObservation) -> Option { + for shape in ALL_SHAPES { + for template in shape.templates() { + if obs.name != template.module_name { + continue; + } + let Some((pid, owner)) = template.parse(&obs.args) else { + continue; + }; + if recorded_argument(&shape.render_args(pid, owner.as_ref())) != obs.args { + continue; + } + return Some(Fingerprint { + id: obs.id, + module_name: obs.name.clone(), + args: obs.args.clone(), + pid, + shape, + owner, + }); + } + } + None +} + +/// 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() +} + +/// 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. +/// +/// `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], + policy: &Policy, + liveness: impl Fn(u32, Attribution) -> 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. + let mut seen: BTreeMap = BTreeMap::new(); + for obs in observations { + if let Some(fp) = classify(obs) { + seen.entry(fp.id).or_insert(fp); + } + } + + // Ownership first, because it decides whether the pid is even a question we can + // ask. A token from another machine, boot or pid namespace names a process this + // one cannot see, and an absent token names nothing at all. + let mut untagged: Vec = Vec::new(); + let mut foreign: Vec = Vec::new(); + let mut judgeable: Vec<(Fingerprint, Attribution)> = Vec::new(); + for fp in seen.into_values() { + match &fp.owner { + Some(token) if policy.local.can_judge(token) => { + judgeable.push((fp, Attribution::Tokened)) + } + Some(_) => foreign.push(fp), + None => match policy.untagged { + UntaggedPolicy::CleanByPidAlone => judgeable.push((fp, Attribution::Untagged)), + UntaggedPolicy::Refuse => untagged.push(fp), + }, + } + } + + // Liveness is asked once per distinct (pid, attribution), not once per module: + // a host with three modules must not be able to change its own verdict mid-plan. + // The attribution is part of the key because one pid can be two questions — see + // the verdict cache below. + let mut live_pids = BTreeSet::new(); + let mut dead_pids = BTreeSet::new(); + let mut unknown_pids = BTreeSet::new(); + // Keyed on (pid, attribution), NOT on pid alone. The same pid can be claimed by + // a tokened module and an untagged one at once — a host that crashed, was + // restarted by an older build, and reused the number — and those two are not the + // same question: one is answered directly, the other only if the degradation + // signals allow it. Collapsing them lets one module's verdict decide another + // module's fate. + let mut verdicts: BTreeMap<(u32, Attribution), Liveness> = BTreeMap::new(); + for (pid, attribution) in judgeable + .iter() + .map(|(fp, attribution)| (fp.pid, *attribution)) + .collect::>() + { + let verdict = liveness(pid, attribution); + verdicts.insert((pid, attribution), verdict); + match verdict { + Liveness::Alive => live_pids.insert(pid), + Liveness::Dead => dead_pids.insert(pid), + Liveness::Unknown => unknown_pids.insert(pid), + }; + } + + let mut unload: Vec = judgeable + .into_iter() + // Each fingerprint is filtered by *its own* verdict, not by whether the pid + // appears in `dead_pids` — which it might, for the other attribution. + .filter(|(fp, attribution)| verdicts.get(&(fp.pid, *attribution)) == Some(&Liveness::Dead)) + .map(|(fp, _)| fp) + .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)); + untagged.sort_by_key(|fp| fp.id); + foreign.sort_by_key(|fp| fp.id); + + Plan { + unload, + live_pids, + dead_pids, + unknown_pids, + untagged, + foreign, + } +} + +#[cfg(test)] +mod tests { + use super::*; + + /// The identity this test process pretends to be. + fn our_identity() -> LocalIdentity { + LocalIdentity { + machine: "aa11bb22".to_string(), + boot: "cc33dd44".to_string(), + pid_ns: 4_026_531_836, + } + } + + /// A token minted by "us". + fn our_token(nonce: u64) -> OwnerToken { + let local = our_identity(); + OwnerToken { + machine: local.machine, + boot: local.boot, + pid_ns: local.pid_ns, + nonce, + } + } + + /// The default policy: our identity, untagged modules refused. + fn our_policy() -> Policy { + Policy { + local: our_identity(), + untagged: UntaggedPolicy::Refuse, + } + } + + fn obs(id: u32, shape: Shape, pid: u32, token: Option<&OwnerToken>) -> ModuleObservation { + ModuleObservation::new( + id, + shape.module_name(), + &recorded_argument(&shape.render_args(pid, token)), + ) + } + + fn null_sink(id: u32, pid: u32) -> ModuleObservation { + obs(id, Shape::LegacyCaptureSink, pid, Some(&our_token(1))) + } + + fn mirror(id: u32, pid: u32) -> ModuleObservation { + obs(id, Shape::LoopbackIntoCapture, pid, Some(&our_token(2))) + } + + fn local_monitor(id: u32, pid: u32) -> ModuleObservation { + obs(id, Shape::LoopbackOutOfCapture, pid, Some(&our_token(3))) + } + + /// The pre-token form: recognisable as ours, but attributable to nobody. + fn legacy_mirror(id: u32, pid: u32) -> ModuleObservation { + obs(id, Shape::LoopbackIntoCapture, pid, None) + } + + fn nothing_is_alive(_: u32, _: Attribution) -> 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(&Shape::LoopbackIntoCapture.render_args(4242, None)), + "source=@DEFAULT_SINK@.monitor sink=pixelpass_capture_4242 latency_msec=20" + ); + assert_eq!( + recorded_argument(&Shape::LoopbackOutOfCapture.render_args(4242, None)), + "source=pixelpass_capture_4242.monitor sink=@DEFAULT_SINK@ latency_msec=20" + ); + assert_eq!( + recorded_argument(&Shape::LegacyCaptureSink.render_args(4242, None)), + "sink_name=pixelpass_capture_4242" + ); + // And the tokened forms, which is what a host actually loads. These exact + // strings were verified against the live server: all three shapes accept the + // property argument and record it byte-identically. + let token = our_token(7); + assert_eq!( + recorded_argument(&Shape::LoopbackIntoCapture.render_args(4242, Some(&token))), + format!( + "source=@DEFAULT_SINK@.monitor sink=pixelpass_capture_4242 latency_msec=20 \ + sink_input_properties=pixelpass.owner={}", + token.render() + ) + ); + assert_eq!( + recorded_argument(&Shape::LoopbackOutOfCapture.render_args(4242, Some(&token))), + format!( + "source=pixelpass_capture_4242.monitor sink=@DEFAULT_SINK@ latency_msec=20 \ + source_output_properties=pixelpass.owner={}", + token.render() + ) + ); + assert_eq!( + recorded_argument(&Shape::LegacyCaptureSink.render_args(4242, Some(&token))), + format!( + "sink_name=pixelpass_capture_4242 sink_properties=pixelpass.owner={}", + token.render() + ) + ); + } + + /// 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 [tokened, legacy] = shape.templates(); + for pid in [1_u32, 7, 4242, 999_999, u32::MAX - 1] { + let plain = recorded_argument(&shape.render_args(pid, None)); + assert_eq!( + legacy.parse(&plain), + Some((pid, None)), + "{shape:?} failed to round-trip legacy pid {pid}" + ); + + let token = our_token(u64::from(pid)); + let tagged = recorded_argument(&shape.render_args(pid, Some(&token))); + assert_eq!( + tokened.parse(&tagged), + Some((pid, Some(token))), + "{shape:?} failed to round-trip tokened 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 + /// 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, &our_policy(), 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)], + &our_policy(), + nothing_is_alive + )), + vec![11] + ); + assert_eq!( + ids(&plan(&[mirror(10, 7)], &our_policy(), 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, &our_policy(), 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, &our_policy(), |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, &our_policy(), |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] + 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, &our_policy(), |_, _| 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()); + } + + /// 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"), + // `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, &our_policy(), 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, &our_policy(), 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], &our_policy(), 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] + fn a_duplicated_observation_yields_one_action() { + let modules = [mirror(10, 4242), mirror(10, 4242)]; + assert_eq!( + ids(&plan(&modules, &our_policy(), nothing_is_alive)), + vec![10] + ); + } + + /// Liveness is asked once per `(pid, attribution)`. 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. Two pids, counted separately: one pid cannot prove + /// "once *per* pid". (Both modules here are tokened, so one pid is one + /// question; `one_pid_with_two_attributions_gets_two_verdicts` covers the case + /// where it is two.) + #[test] + fn liveness_is_decided_once_per_pid_and_attribution_not_once_per_module() { + 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, &our_policy(), |pid, _| { + *calls.borrow_mut().entry(pid).or_insert(0) += 1; + Liveness::Dead + }); + 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 + /// 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(&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. + assert!(!fp.still_matches(&ModuleObservation::new( + 10, + "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))); + } + + /// The module *name* is part of the identity, and this is provable because the + /// argument grammars overlap: `module-echo-cancel sink_name=pixelpass_capture_9` + /// is byte-identical to the canonical null-sink argument. So a comparator using + /// only `id + args` accepts a foreign module — and unloads it. + /// + /// (I previously argued this case could not be constructed non-vacuously, on the + /// grounds that the name determines which grammar can match. That was wrong: the + /// grammars are not disjoint across names.) + #[test] + fn an_identical_argument_under_another_module_name_is_not_a_match() { + let ours = null_sink(5, 9); + let fp = classify(&ours).expect("ours"); + let impostor = ModuleObservation::new(5, "module-echo-cancel", &ours.args); + + assert_eq!(fp.args, impostor.args, "the arguments really are identical"); + assert_eq!(fp.id, impostor.id, "and so is the index"); + assert!( + !fp.still_matches(&impostor), + "only the module name distinguishes these, so it must be compared" + ); + assert_eq!( + classify(&impostor), + None, + "and it is not ours to begin with" + ); + } + + /// 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 respaced_arguments_do_not_match() { + let fp = classify(&mirror(10, 4242)).expect("ours"); + assert!(!fp.still_matches(&ModuleObservation::new( + 10, + "module-loopback", + "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); + } + + /// A token must survive the round trip through a module argument exactly, and + /// anything this build does not fully understand must not parse at all. + #[test] + fn tokens_round_trip_and_reject_what_they_do_not_understand() { + let token = our_token(42); + assert_eq!(OwnerToken::parse(&token.render()), Some(token.clone())); + + for raw in [ + "", + "1", + "1-aa11bb22", + "1-aa11bb22-cc33dd44", + "1-aa11bb22-cc33dd44-4026531836", + // A version this build does not know: not ours to touch. + "2-aa11bb22-cc33dd44-4026531836-42", + "0-aa11bb22-cc33dd44-4026531836-42", + // Trailing junk. + "1-aa11bb22-cc33dd44-4026531836-42-extra", + // Non-hex identities. + "1-zzzz-cc33dd44-4026531836-42", + "1-aa11bb22-zzzz-4026531836-42", + // Non-numeric namespace or nonce. + "1-aa11bb22-cc33dd44-abc-42", + "1-aa11bb22-cc33dd44-4026531836-abc", + // Empty identity components. + "1--cc33dd44-4026531836-42", + ] { + assert_eq!(OwnerToken::parse(raw), None, "should reject {raw:?}"); + } + } + + /// The defect the token exists for: a pid means different processes in different + /// pid namespaces, so a module whose token names another namespace, boot or + /// machine must never be judged by asking about that number here. + #[test] + fn a_module_from_another_namespace_boot_or_machine_is_never_touched() { + let local = our_identity(); + let elsewhere = [ + // Same machine and boot, different pid namespace: the number is not ours. + OwnerToken { + pid_ns: local.pid_ns + 1, + ..our_token(1) + }, + // Same machine, earlier boot: the whole pid space has been recycled. + OwnerToken { + boot: "ffffffff".to_string(), + ..our_token(1) + }, + // Another machine entirely. + OwnerToken { + machine: "99998888".to_string(), + ..our_token(1) + }, + ]; + + for token in elsewhere { + let modules = [obs(10, Shape::LoopbackIntoCapture, 4242, Some(&token))]; + // `nothing_is_alive` would happily call the pid dead, so a plan that + // consults liveness at all here is already wrong. + let plan = plan(&modules, &our_policy(), nothing_is_alive); + assert!( + plan.is_empty(), + "a module from {token:?} must not be planned: {plan:?}" + ); + assert_eq!(plan.foreign.len(), 1, "and it must be reported: {plan:?}"); + assert!( + plan.dead_pids.is_empty(), + "its pid must not even be considered" + ); + } + } + + /// Liveness must not be consulted at all for a module we cannot attribute — + /// asking is the bug, because the answer is meaningless. + #[test] + fn an_unattributable_module_is_never_asked_about() { + let foreign = OwnerToken { + pid_ns: our_identity().pid_ns + 1, + ..our_token(1) + }; + let modules = [ + obs(10, Shape::LoopbackIntoCapture, 4242, Some(&foreign)), + legacy_mirror(11, 5555), + ]; + let asked = std::cell::RefCell::new(Vec::new()); + let plan = plan(&modules, &our_policy(), |pid, _| { + asked.borrow_mut().push(pid); + Liveness::Dead + }); + assert!( + asked.borrow().is_empty(), + "no pid should have been asked about, but these were: {:?}", + asked.borrow() + ); + assert!(plan.is_empty()); + assert_eq!(plan.foreign.len(), 1); + assert_eq!(plan.untagged.len(), 1); + } + + /// Untagged (pre-token) modules are refused by default and reported, and only an + /// explicit policy judges them by pid alone. + #[test] + fn untagged_modules_are_refused_by_default_and_only_cleaned_on_request() { + let modules = [ + legacy_mirror(10, 4242), + obs(11, Shape::LegacyCaptureSink, 4242, None), + ]; + + let refused = plan(&modules, &our_policy(), nothing_is_alive); + assert!( + refused.is_empty(), + "default must not touch them: {refused:?}" + ); + assert_eq!(refused.untagged.len(), 2); + assert!(refused.dead_pids.is_empty()); + + let opted_in = plan( + &modules, + &Policy { + local: our_identity(), + untagged: UntaggedPolicy::CleanByPidAlone, + }, + nothing_is_alive, + ); + assert_eq!(ids(&opted_in), vec![10, 11], "{opted_in:?}"); + assert!(opted_in.untagged.is_empty()); + // Even opted in, a live pid still wins. + let live = plan( + &modules, + &Policy { + local: our_identity(), + untagged: UntaggedPolicy::CleanByPidAlone, + }, + |_, _| Liveness::Alive, + ); + assert!(live.is_empty(), "{live:?}"); + } + + /// The nonce narrows the ABA window: two loads by the same pid no longer render + /// identical arguments, so a fingerprint taken from one does not match the other. + #[test] + fn the_nonce_distinguishes_two_loads_by_the_same_pid() { + let first = obs(10, Shape::LoopbackIntoCapture, 4242, Some(&our_token(1))); + let second = obs(10, Shape::LoopbackIntoCapture, 4242, Some(&our_token(2))); + assert_ne!(first.args, second.args, "the nonce must reach the argument"); + + let fp = classify(&first).expect("ours"); + assert!(fp.still_matches(&first)); + assert!( + !fp.still_matches(&second), + "a different load must not satisfy the first load's fingerprint" + ); + } + + /// The attribution handed to the liveness callback decides *how* the question + /// may be asked, so it must be right per module. Getting this wrong made the + /// token useless in a container — the one place it exists for — because the + /// probe's namespace guesswork answered `Unknown` for a pid the token had + /// already proven local. + #[test] + fn attribution_is_reported_per_module() { + let modules = [mirror(10, 100), legacy_mirror(11, 200)]; + let asked = std::cell::RefCell::new(Vec::new()); + let plan = plan( + &modules, + &Policy { + local: our_identity(), + untagged: UntaggedPolicy::CleanByPidAlone, + }, + |pid, attribution| { + asked.borrow_mut().push((pid, attribution)); + Liveness::Dead + }, + ); + let asked = asked.into_inner(); + assert_eq!( + asked, + vec![(100, Attribution::Tokened), (200, Attribution::Untagged)], + "each pid must be asked about with its own module's attribution" + ); + assert_eq!(ids(&plan), vec![10, 11]); + } + + /// One pid, two attributions. A host crashes, an older build restarts and reuses + /// the number, and now a tokened module and an untagged one both claim it. Those + /// are two different questions — one answerable directly, the other only if the + /// degradation signals allow it — so each module must be filtered by *its own* + /// verdict. Keying the verdict cache on the pid alone let whichever module came + /// last decide both. + /// + /// Run in both module-id orders, because the bug was order-dependent. + #[test] + fn one_pid_with_two_attributions_gets_two_verdicts() { + let policy = Policy { + local: our_identity(), + untagged: UntaggedPolicy::CleanByPidAlone, + }; + // The tokened module's owner is dead; the untagged claim on the same number + // cannot be judged, which is what a degraded probe would say. + let verdict = |_pid: u32, attribution: Attribution| match attribution { + Attribution::Tokened => Liveness::Dead, + Attribution::Untagged => Liveness::Unknown, + }; + + for (tokened_id, untagged_id) in [(10, 11), (11, 10)] { + let modules = [ + obs( + tokened_id, + Shape::LoopbackIntoCapture, + 4242, + Some(&our_token(1)), + ), + obs(untagged_id, Shape::LoopbackOutOfCapture, 4242, None), + ]; + let plan = plan(&modules, &policy, verdict); + assert_eq!( + ids(&plan), + vec![tokened_id], + "only the attributable module may be planned (ids {tokened_id}/{untagged_id}): \ + {plan:?}" + ); + } + + // And the mirror image: the untagged claim is judged dead while the tokened + // owner is alive. The live owner's module must survive. + let inverted = |_pid: u32, attribution: Attribution| match attribution { + Attribution::Tokened => Liveness::Alive, + Attribution::Untagged => Liveness::Dead, + }; + for (tokened_id, untagged_id) in [(10, 11), (11, 10)] { + let modules = [ + obs( + tokened_id, + Shape::LoopbackIntoCapture, + 4242, + Some(&our_token(1)), + ), + obs(untagged_id, Shape::LoopbackOutOfCapture, 4242, None), + ]; + let plan = plan(&modules, &policy, inverted); + assert_eq!( + ids(&plan), + vec![untagged_id], + "a live tokened owner must not have its module unloaded because an \ + untagged claim on the same pid looked dead: {plan:?}" + ); + } + } + + /// Tokened and legacy forms must both classify, and carry the difference. + #[test] + fn both_forms_classify_and_record_whether_they_are_attributable() { + let tokened = classify(&mirror(10, 4242)).expect("tokened is ours"); + assert_eq!(tokened.owner, Some(our_token(2))); + + let legacy = classify(&legacy_mirror(11, 4242)).expect("legacy is still ours"); + assert_eq!(legacy.owner, None); + assert_eq!(legacy.shape, Shape::LoopbackIntoCapture); + } + + /// 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, &our_policy(), nothing_is_alive), + plan(&b, &our_policy(), nothing_is_alive) + ); + } +}