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/repair/introspect.rs b/src/repair/introspect.rs new file mode 100644 index 0000000..99dc5a2 --- /dev/null +++ b/src/repair/introspect.rs @@ -0,0 +1,286 @@ +//! 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. +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 IO + // events with is still alive, then let the mainloop go. Bounded and + // best-effort: this runs on the way out, with nobody left to report to. + if let Some(mut context) = self.context.take() { + context.disconnect(); + drop(context); + } + for _ in 0..8 { + if matches!(self.mainloop.iterate(false), IterateResult::Success(_)) { + continue; + } + break; + } + } +} + +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 index e107e1c..2a96352 100644 --- a/src/repair/mod.rs +++ b/src/repair/mod.rs @@ -16,47 +16,21 @@ //! 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: one listing, atomically +//! # Where the observations come from //! -//! `pactl list short modules`, in a single invocation, and nothing else. Every -//! observation's `(index, name, argument)` therefore comes from **one** server -//! response and cannot be mis-assembled. -//! -//! This replaced a two-listing scheme (indices from the short listing, exact -//! arguments from `pactl -f json list modules`, correlated by position). The JSON -//! listing was needed because it carries the exact argument where the short -//! listing's is tab-delimited text — but on pactl 17 its records carry **no module -//! index at all** (`"index": null`) while `unload-module` accepts only an index, so -//! it can never stand alone. Correlating the two by position is unsound with -//! repeated module names: if another client loads one module and unloads another -//! between the two calls, the counts and names still line up while the arguments -//! have shifted by one, and a *foreign* module can inherit a canonical -//! fingerprint. The name check cannot see that, and the retry never fires because -//! correlation "succeeded". -//! -//! Two consequences of using the short listing alone, both handled rather than -//! hoped away: -//! -//! - **A tab inside an argument is invisible to this format.** Such a row is marked -//! `args_complete: false`; [`plan::classify`] refuses it outright (a truncation -//! could otherwise coincide with a canonical form) while the reference gate can -//! still see that the row names a sink. -//! - **A crafted argument containing a newline can fabricate a row.** A fabricated -//! row only does damage if it names a *real* module's index — which makes that -//! index appear twice — so a duplicated index refuses the whole run. -//! -//! The remaining exact route is libpulse introspection (`pa_module_info` carries -//! index, name and argument in one record). That is a new dependency plus a -//! mainloop in a one-shot CLI path, so it is recorded as the upgrade rather than -//! taken now; see the deferred item in the impl plan. +//! 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 std::process::Command; -use plan::{Fingerprint, Liveness, ModuleObservation, Shape}; +use introspect::PulseSession; +use plan::{Fingerprint, Liveness, Shape}; pub async fn run() -> Result<()> { let liveness = LivenessProbe::new(); @@ -67,7 +41,10 @@ pub async fn run() -> Result<()> { ); } - let modules = snapshot().context("failed to observe pactl modules")?; + 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 @@ -123,7 +100,9 @@ pub async fn run() -> Result<()> { // Fresh snapshot per action. Deliberately not hoisted out of the loop: // each unload changes the module table, and the point is to decide // against the table as it is *now*, not as it was when the plan was made. - let current = snapshot().context("failed to re-observe pactl modules")?; + 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", @@ -182,7 +161,7 @@ pub async fn run() -> Result<()> { } } - match unload_module(fp.id) { + match pulse.unload_module(fp.id) { Ok(()) => { println!("[pixelpass] --repair: {}", describe(fp)); unloaded += 1; @@ -234,26 +213,36 @@ fn describe(fp: &Fingerprint) -> String { /// Answers "is this pid still around", and knows when it must refuse to answer. /// -/// Two separate hazards, and the first one is the dangerous one: +/// **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. /// -/// 1. **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. Repair there can reach the host's Pulse socket, -/// see a live host's modules, conclude its pid is dead, and unload a running -/// host's audio. `hidepid` has the same self-visible/others-invisible shape. So -/// a "can I see myself" preflight proves nothing; what is needed is positive -/// confidence that our pid numbers mean the same thing as the ones in the -/// module names. `NSpid` in `/proc/self/status` answers that directly: more than -/// one entry means we are nested, and every verdict becomes `Unknown`. -/// 2. **An error is not an absence.** `Path::exists()` maps permission errors and a -/// missing `/proc` to `false`, which here reads as "dead, go ahead". 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 remote Pulse server also fails closed: the module table then belongs to -/// another machine's processes, where our pids mean nothing at all. +/// 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 pid numbers here are trustworthy; `Some(reason)` when every - /// answer must be [`Liveness::Unknown`]. + /// `None` when no signal says pid numbers are untrustworthy; `Some(reason)` + /// when every answer must be [`Liveness::Unknown`]. degraded: Option, } @@ -265,18 +254,12 @@ impl LivenessProbe { } fn detect_degradation() -> Option { - // A remote server's modules belong to another machine's pids. - if let Some(server) = std::env::var_os("PULSE_SERVER") { - let server = server.to_string_lossy().to_string(); - let local = - server.starts_with("unix:") || server.starts_with('/') || server.starts_with("{"); - if !local { - return Some(format!("PULSE_SERVER={server} is not a local socket")); - } - } - // The authoritative namespace question: NSpid lists this process's pid in - // every namespace it is visible in, outermost first. More than one entry - // means our pid numbers are not the ones the outer namespace uses. + // 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 @@ -290,6 +273,8 @@ impl LivenessProbe { 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( @@ -336,191 +321,20 @@ impl LivenessProbe { } } -// ────────────────────────────────────────────────────────────────────── -// Observation -// ────────────────────────────────────────────────────────────────────── - -/// One observation of the whole module table, from a single `pactl` invocation. -fn snapshot() -> Result> { - let output = Command::new("pactl") - .args(["list", "short", "modules"]) - .output() - .context("failed to run pactl")?; - 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")?; - parse_short(&text) -} - -/// Parse `pactl list short modules`: `\t\t` per module, one -/// server response, so a row's three fields always belong together. -/// -/// Two format hazards, both handled rather than assumed away: -/// -/// - **Tabs.** pactl emits a trailing tab after the argument, so a well-formed row -/// has at most one empty field after it. Any *non-empty* content past the -/// argument means the argument itself contained a tab and this format cannot -/// show it in full: the row is marked `args_complete: false`, never classified, -/// but still visible to the reference gate. -/// - **Newlines.** Some modules render arguments as a multi-line `{ … }` block; the -/// continuation lines never parse as a u32 index, so they are dropped. A crafted -/// argument *could* still fabricate a row that does parse — but to do damage it -/// must claim a real module's index, which then appears twice. A duplicated index -/// therefore refuses the whole run. -fn parse_short(text: &str) -> Result> { - let mut rows: Vec = Vec::new(); - for line in text.lines() { - let fields: Vec<&str> = line.split('\t').collect(); - let Some(id_str) = fields.first() else { - continue; - }; - let Ok(id) = id_str.parse::() else { - continue; - }; - let Some(name) = fields.get(1) else { continue }; - let args = fields.get(2).copied().unwrap_or(""); - let complete = fields.iter().skip(3).all(|extra| extra.is_empty()); - rows.push(if complete { - ModuleObservation::new(id, name, args) - } else { - ModuleObservation::truncated(id, name, args) - }); - } - // A repeated index is the one shape a fabricated row needs in order to make us - // unload a module that exists. It is also never legitimate. - for (i, row) in rows.iter().enumerate() { - if rows[..i].iter().any(|earlier| earlier.id == row.id) { - bail!( - "module index #{} appears more than once in `pactl list short modules`; \ - refusing to unload anything", - row.id - ); - } - } - Ok(rows) -} - -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::*; - /// The whole row comes from one response: index, name and the exact argument, - /// including pactl's trailing tab. - #[test] - fn parses_index_name_and_exact_argument() { - let text = "5\tmodule-null-sink\tsink_name=pixelpass_capture_42\t\n\ - 10\tmodule-loopback\tsource=@DEFAULT_SINK@.monitor \ - sink=pixelpass_capture_42 latency_msec=20\t\n\ - 7\tmodule-always-sink\n"; - let rows = parse_short(text).expect("well-formed"); - assert_eq!(rows.len(), 3); - assert_eq!(rows[0].id, 5); - assert_eq!(rows[0].args, "sink_name=pixelpass_capture_42"); - assert!(rows[0].args_complete, "a trailing tab is not truncation"); - assert!(rows[1].args.ends_with("latency_msec=20")); - assert_eq!(rows[2].args, "", "a module without arguments still parses"); - } + // 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. - /// An argument containing a tab cannot be shown in full by this format. Such a - /// row must never be classified — a truncation could coincide with a canonical - /// form — but it must still be *present*, or the reference gate would not see - /// that something names our sink. - #[test] - fn a_tab_inside_an_argument_marks_the_row_incomplete() { - let canonical = "source=@DEFAULT_SINK@.monitor sink=pixelpass_capture_42 latency_msec=20"; - let text = format!("10\tmodule-loopback\t{canonical}\tremix=false\n"); - let rows = parse_short(&text).expect("well-formed"); - assert_eq!(rows.len(), 1); - assert!(!rows[0].args_complete); - assert_eq!( - plan::classify(&rows[0]), - None, - "a truncated argument that happens to match must not be ours" - ); - // Still visible to the gate that protects a sink from being destroyed. - assert_eq!( - plan::sink_still_referenced(&rows, 42, 999), - Some(10), - "an unreadable argument still counts as a reference" - ); - } - - /// Modules whose arguments render as a `{ … }` block wrap onto continuation - /// lines; swallowing one as a module would fabricate an entry — and a - /// fabricated entry is an index we might later unload. - #[test] - fn ignores_continuation_lines_of_multi_line_argument_blocks() { - let text = "1\tlibpipewire-module-rt\t{\n\ - \x20 nice.level = -11\n\ - \x20 rt.prio = 88\n\ - }\n\ - 5\tmodule-null-sink\tsink_name=pixelpass_capture_42\n"; - let rows = parse_short(text).expect("well-formed"); - assert_eq!(rows.len(), 2, "{rows:?}"); - assert_eq!(rows[1].id, 5); - } - - /// A crafted argument can fabricate a row that parses. It only does damage if - /// it claims a *real* module's index — which makes that index appear twice — so - /// a duplicate index refuses the entire run rather than unloading a stranger. - #[test] - fn a_duplicated_index_refuses_the_whole_run() { - // As it would arrive: a module whose argument embeds a newline and a row - // that re-uses index 5, which really belongs to something else. - let text = "5\tmodule-real-thing\tconfig={\n\ - 5\tmodule-null-sink\tsink_name=pixelpass_capture_4242\n"; - let err = parse_short(text).expect_err("a repeated index is never legitimate"); - assert!( - format!("{err:#}").contains("more than once"), - "unexpected error: {err:#}" - ); - } - - /// End-to-end from raw pactl text: the parser feeding the planner. Without - /// this, a parser that dropped every argument would leave the entire planner - /// suite green while real `--repair` recognised nothing at all. - #[test] - fn raw_pactl_output_becomes_a_plan() { - let text = "1\tlibpipewire-module-rt\t{\n\ - \x20 nice.level = -11\n\ - }\n\ - 10\tmodule-loopback\tsource=@DEFAULT_SINK@.monitor \ - sink=pixelpass_capture_4242 latency_msec=20\t\n\ - 11\tmodule-loopback\tsource=pixelpass_capture_4242.monitor \ - sink=@DEFAULT_SINK@ latency_msec=20\t\n"; - let observations = parse_short(text).expect("well-formed"); - - let planned = plan::plan(&observations, |_| Liveness::Dead); - assert_eq!( - planned.unload.iter().map(|fp| fp.id).collect::>(), - vec![10, 11], - "both orphan loopbacks must be planned from raw output: {planned:?}" - ); - assert_eq!(planned.dead_pids.len(), 1); - } - - /// The probe must refuse to call anything dead when pid numbers here may not - /// mean what the module names mean. On this host it should be confident; the - /// point of the assertion is that `of()` never returns `Dead` while degraded. + /// 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 { @@ -530,14 +344,16 @@ mod tests { assert_eq!(degraded.of(u32::MAX), Liveness::Unknown); let probe = LivenessProbe::new(); - // Our own pid is alive by construction — unless this test itself runs in a - // pid namespace, which is exactly when the probe must abstain. + // 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), } - // pid 0 would signal our whole process group; it is never askable. + // `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); } } diff --git a/src/repair/plan.rs b/src/repair/plan.rs index c0f67d0..e575cdd 100644 --- a/src/repair/plan.rs +++ b/src/repair/plan.rs @@ -97,24 +97,16 @@ pub fn recorded_argument(args: &[String]) -> String { /// One module exactly as the server reported it. /// -/// `args` is the **exact** recorded argument string, not a normalised one. Within -/// a single repair invocation every snapshot comes from the same server, so -/// re-rendering is not a thing that happens, and normalising would only make two -/// genuinely different arguments compare equal (whitespace inside a quoted -/// property value is not layout). +/// `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, - /// False when the observation format could not carry the whole argument (the - /// short listing is tab-delimited, and an argument may itself contain a tab). - /// - /// A truncated argument is never ours — [`classify`] refuses it outright, - /// because a truncation could otherwise coincide with a canonical form. It is - /// still kept in the snapshot: [`sink_still_referenced`] must be able to see - /// that *something* names a sink even when it cannot read the whole argument. - pub args_complete: bool, } impl ModuleObservation { @@ -123,15 +115,6 @@ impl ModuleObservation { id, name: name.to_string(), args: args.to_string(), - args_complete: true, - } - } - - /// An observation whose argument the format could not fully carry. - pub fn truncated(id: u32, name: &str, args: &str) -> Self { - Self { - args_complete: false, - ..Self::new(id, name, args) } } } @@ -357,12 +340,6 @@ impl Plan { /// 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 { - // An argument we could not read in full is never ours: a truncation could - // coincide with a canonical form, and unloading on a coincidence is the one - // outcome none of these rules tolerate. - if !obs.args_complete { - return None; - } for shape in ALL_SHAPES { let template = shape.template(); if obs.name != template.module_name {