diff --git a/Cargo.lock b/Cargo.lock index 95dffd3..7a8568d 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4883,6 +4883,7 @@ dependencies = [ "image", "iroh", "iroh-gossip", + "libc", "opus", "pipewire", "rand 0.10.1", diff --git a/Cargo.toml b/Cargo.toml index dfd50bc..5cb4483 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -109,3 +109,10 @@ windows-sys = { version = "0.61", features = [ "Win32_System_Diagnostics_ToolHelp", "Win32_System_Threading", ] } + +# Unix-only. Used for exactly one thing: sending SIGINT to our own +# pixelpass child so it can run its cleanup before we resort to SIGKILL +# (src/core/teardown.rs). Already in the tree via alsa/cpal/tokio, so +# declaring it directly adds no new code to the build. +[target.'cfg(unix)'.dependencies] +libc = "0.2.186" diff --git a/docs/screenshare-audio-exclusion-impl-plan.md b/docs/screenshare-audio-exclusion-impl-plan.md index 5d0bc3f..f4a661c 100644 --- a/docs/screenshare-audio-exclusion-impl-plan.md +++ b/docs/screenshare-audio-exclusion-impl-plan.md @@ -76,7 +76,7 @@ if it differs, failing closed. | # | Phase | Repo | Mutates graph? | Exit gate | | --- | --- | --- | --- | --- | | 0a | `object.serial` u64 fix | pixelpass | no | boundary parse tests | -| 0b | Explicit teardown + drop ordering | peerspeak | no | **five independent mutations** (plan §2) | +| 0b | Explicit teardown + drop ordering | peerspeak | no | **four independent mutations** (plan §2; revised from five, §10 r14) ✅ built | | 0c | Graceful stop + connection-owned capture sink | both | sink ownership | two-host SIGKILL live gate **+ SIGINT-first gate** | | 0d | **Typed capture plan + internal mode input** | pixelpass | no | mode matrix; neither unsafe source nor unsafe sink input constructible | | 1 | peerspeak ownership tagging | peerspeak | no | tag on live nodes; literal pinned in plan §3 | @@ -151,7 +151,19 @@ v3.4 §7.2, decision D4. All **three** of v3.4's fixes: AEC guard unloads. This is the *only* protection on the panic/unwind path, and unwind is reachable — the core has numerous `unwrap()` sites and no `panic=abort` profile. -⚠️ **Mutation testing: five mutations, each independently breaking a named test.** v1 demanded +> ✅ **0b IMPLEMENTED 2026-07-26** (peerspeak branch `phase-0b-teardown`). The ordering +> defect was live: `echo_cancel` was declared *ahead* of `screenshare_host`, so any unwind +> unloaded the AEC while the host was still fanning out. Fields moved into +> `src/core/teardown.rs` with `echo_cancel` declared last, and `ReapOnDrop` added because +> `kill_on_drop(true)` only *signals* — it hands the child to the runtime's orphan queue, +> which an unwinding runtime may never drain. Matrix revised to four mutations; see §10 +> round 14, and round 15 for the two blocking review findings that followed. +> +> **Owed to phase 9:** an explicit lifecycle row — *drop the controller / close the command +> channel while sharing* — which is the live proof for the hoisted teardown call site. + +⚠️ **Mutation testing: five mutations, each independently breaking a named test.** +⚠️ **SUPERSEDED by §10 round 14 — mutation 2 is vacuous and the matrix is now four.** v1 demanded a mutation that targeted the wrong defense; v2 fixed that but bundled two defenses into one combined mutant, which proves neither. Final form: @@ -167,6 +179,45 @@ Both channel-close arms get their own test; a single "closes the command channel exercise one arm and leave the other unsafe. ### 0c. Graceful stop + connection-owned capture sink — both + +> ✅ **MECHANISM PROBE PASSED on this host, 2026-07-26** (PipeWire 1.6.8). Run *before* any +> structural work, on the reviewer's insistence, because a single unverified assumption could +> have invalidated the entire approach: whether a hand-created adapter is visible to +> pipewire-pulse under the name pixelpass's capture path depends on. It is. +> +> ``` +> pw-cli> create-node adapter factory.name=support.null-audio-sink \ +> node.name=pixelpass_probe_ media.class=Audio/Sink \ +> audio.channels=2 audio.position=[FL,FR] node.virtual=true \ +> monitor.channel-volumes=true object.linger=false +> ``` +> +> Five gates, all green: +> 1. `pactl list short sinks` shows the sink under the **exact** `node.name`. +> 2. `pactl list short sources` shows **`.monitor`** — the derived monitor name is +> a pipewire-pulse contract, not a property of Pulse-created sinks. This was the one that +> could have sunk the approach. +> 3. `gst-launch-1.0 pulsesrc device=.monitor num-buffers=40 ! fakesink` pulled its +> buffers and exited clean, and a real recording stream attached — so the pixelpass capture +> path works against it unchanged. +> 4. **No null-sink module was loaded** (`pactl list short modules | grep -c null-sink` stayed +> at its baseline of 3). It is genuinely not a Pulse module. +> 5. **SIGKILL of the owning connection removed both Pulse-visible names**, with zero residue +> anywhere in `pw-dump`. That is the entire point of 0c, demonstrated on the real graph. +> +> The default sink never moved, so this is also safe to run on a live desktop. +> **Every O1 stop condition listed for 0c is retired.** `object.linger=false` is load-bearing: +> the bundled pipewire-rs example sets `linger=1` for the opposite behaviour. +> +> ⚠️ **`--repair`'s job does not shrink — it BREAKS.** Discovery derives dead host PIDs +> **only** from `module-null-sink` entries (`pixelpass/src/repair.rs`), and only then matches +> loopbacks against that PID set. The native sink is scoped to **every mode that owns a +> capture sink**, not just `DesktopExcluding`, so legacy Pulse loopbacks will coexist with a +> connection-owned sink; when that host dies the sink vanishes automatically and its loopbacks +> become **undiscoverable orphans**. Candidate PIDs must be derived independently from all +> three module shapes (`null-sink sink_name=`, `loopback sink=`, `loopback source=…monitor`), +> with a liveness recheck immediately before each destructive unload. This makes the repair +> rework **load-bearing, not defensive**. v3.4 §7.4. The **largest hidden cost in Phase 0**: moving the null sink off `pactl load-module` (`pixelpass/src/host/audio.rs:69`, cleaned up only in `Routing::cleanup` at `:259-260`, which SIGKILL skips) onto a connection-owned PipeWire object. @@ -758,6 +809,88 @@ mid-share load stays a **synthetic** test until a second `enable` site or hot re ## 10. Adjudication record +**Round 14 (2026-07-26) — 0b's five-mutation matrix is revised to four, and one pinned +mutation is retired as vacuous.** Reached independently by both reviewers, then agreed. + +- **Mutation 2 cannot be killed by any test, because its site cannot execute.** The + best-effort wake arm (`core/mod.rs`, the `besteffort_wake_rx` close arm) is unreachable + **by construction, twice over**: (i) `run_core_loop` owns a clone of `besteffort_wake_tx` + — created at `CoreController::new` and used for the `has_more` re-arm inside the loop — + and a tokio `Receiver::recv()` yields `None` only once *every* sender is dropped; (ii) + even without that clone, both `CoreController` and `CoreCommandSender` hold `reliable_tx` + alongside the wake sender, and the `select!` is `biased` with the reliable arm first, so + the reliable arm always wins the race to exit. Writing teardown there would be code that + provably never runs, dressed as a tested path. +- **Mutation 1's site is reachable but not unit-testable.** It sits inside `run_core_loop`, + which builds a real iroh endpoint and loads identity; no unit test can drive it. +- **Decision: (b) + (c).** Teardown is **hoisted to one unconditional site after the loop**, + so every `break` is covered structurally, including any added later — strictly better than + duplicating teardown across one live arm and one dead one. The **seam-level mutation gates + are the real ordering proof**, and the call site's live proof is owed to **phase 9**, which + gains an explicit row: *drop the controller / close the command channel while sharing*. + "UI crash" is not precise enough to serve as that row. +- **Rejected: an integration test built to preserve the number five.** It would pay for a + full iroh core plus test-only observability and prove only that a method was called — not + the ordering invariant, which is the thing that actually breaks. +- The 0b gate is therefore **four mutations**, enumerated exactly (round 16 P3-5 — the earlier + wording said "three plus the 0c pair", which reads as five and blurred what 0b owns): + + | # | mutation | killed by | status | + |---|----------|-----------|--------| + | 1 (old gate 3) | remove the wait after the host kill | `explicit_shutdown_reaps_the_host_before_the_aec_can_unload` | killed now | + | 2 (old gate 4) | reverse `ScreenshareTeardown`'s field declaration order | `the_aec_unloads_after_the_children_on_the_drop_path` | killed now | + | 3 (old gate 5) | remove the reap loop from `ReapOnDrop::drop` | `dropping_a_guard_kills_and_then_reaps_the_child` | killed now | + | 4 (old 1) | remove the teardown at the hoisted post-loop call site | — | **deferred to the phase-9 row** *drop the controller / close the command channel while sharing* | + + 4-vs-5 separation verified: reversing the field order leaves the reap test green, and + removing the reap loop leaves the ordering test green. **0c's own pair (no SIGINT · no + SIGKILL fallback) is counted under 0c, not here**, along with the round-15/16 additions + (disarm the wrapper at entry · disarm it between the waits · treat a wait error as a reap · + report an unconfirmed stop as clean · zero the grace). + +**Round 15 (2026-07-26) — review of the 0b/0c-peerspeak implementation returned two blocking +findings, both accepted.** Recorded because both are the same shape: a defence that existed +but was disarmed exactly when it was needed. + +- **The drop fallback was disarmed across its own wait.** `shutdown` took the child out of + the wrapper before the first `.await`; a cancellation or unwind during the wait left the + raw child to drop with `kill_on_drop` (which signals without reaping) while `Drop` found + `None`. The child now stays owned until the reap is **confirmed**. +- **A failed wait was reported as a reap, and the hard-kill wait was unbounded.** The + `io::Result` was discarded, so a wait error returned "reaped"; and a process in + uninterruptible sleep after SIGKILL could wedge the core loop forever. Both waits are now + bounded and the conflict case has a written policy: availability wins, the child stays + owned so the bounded `Drop` retry stays armed, and the residual risk is logged. +- **A gate of mine was vacuous and the review's fourth test-double point caught it.** The + elapsed-time assertion compared against `STOP_GRACE` itself, so zeroing the constant left + it trivially true. `the_grace_is_a_real_interval` now pins the constant to a band. + +**Round 16 (2026-07-26) — the re-review of the 0b/0c-peerspeak fixes returned *approve with +follow-ups*: no blocking findings, five P3s, all five applied before the merge.** The two that +carry design content: + +- **An unconfirmed stop was reported to the user as a clean one.** `stop_host` returned a bare + "was sharing" bool, so the one case where availability-first gives up (SIGKILL queued, reap + never confirmed) still emitted `ScreenShareStopped` with no warning — the UI would say + sharing ended while pixelpass might still be fanning out. `ReapOnDrop::shutdown` now returns + `StopOutcome`, `stop_host` returns `Option`, and an `Unconfirmed` user-initiated + stop raises a UI error naming the stray process. Session/viewer teardown discards the outcome + on purpose: no user is waiting on an answer there and the risk is already logged. +- **Cancellation coverage only reached the graceful wait.** The mid-wait test could not kill a + mutant that disarmed the wrapper *between* the two waits. Verified: the naive form of that + mutant does not compile (the child is borrowed from `self`), but the restructured form — + `self.child.take()` once cooperation has failed — compiles, and the pre-existing test passes + it. `cancelling_shutdown_after_the_kill_leaves_the_fallback_armed` kills it. + +**Deferred item — aggregate teardown latency (round 16 P3-4).** Bounds are per child, not per +teardown. Sequential drain gives `2 × STOP_GRACE` per unconfirmed child inline (≈4 s), plus +`REAP_BUDGET` (250 ms) per child on the `Drop` path: three wedged children ≈6 s of command-loop +stall, ≈12.75 s worst case including drop retries. Accepted as-is for 0b — one host plus one or +two viewers is the real shape, and concurrency here would mean detaching children from the +session that owns the AEC's lifetime. **Trigger to revisit: a fourth tracked child becomes +routine, or a measured teardown exceeds 5 s.** The fix, when triggered, is to drain viewers +concurrently while still owned by `shutdown_children` — not to detach them. + **Round 1 — 13 items, 12 accepted.** Phase reorder (AEC machine before dry-run); typed capture plan (accepted, moved *earlier* than proposed); Phase 3 five-part gate; tag-consumption gating; Phase 6 matrix mandatory; 0b unwind backstop restored **and my mutation test corrected — it diff --git a/src/core/mod.rs b/src/core/mod.rs index b0bc4e7..6a3c78e 100644 --- a/src/core/mod.rs +++ b/src/core/mod.rs @@ -4,6 +4,7 @@ pub mod fetchbudget; pub mod jitter; pub mod messages; mod recovery; +mod teardown; use crate::audio::eq::{Eq, EqSettings}; use crate::audio::{AudioBackend, PlatformAudioBackend}; @@ -677,31 +678,33 @@ struct ActiveSession { recovery_terminal_task: tokio::task::JoinHandle<()>, grace_timers: GraceTimers, transport: Arc, - /// Loaded PipeWire echo-cancel module (if enabled); unloads on drop. - #[cfg(target_os = "linux")] - echo_cancel: Option, - /// Our pixelpass screen-share host child while sharing (`kill_on_drop`, so it - /// also dies if the session is dropped without an explicit stop). - screenshare_host: Option, - /// pixelpass viewer children we spawned to watch peers' shares, each paired - /// with the share ticket it's viewing so a re-watch of the same share can - /// replace (not stack) its player. Killed on session teardown (each also - /// self-exits when its player window closes). - screenshare_viewers: Vec<(String, tokio::process::Child)>, + /// The screen-share children and the echo-cancel module, held together + /// because their **destruction order** is load-bearing: the AEC module must + /// not unload while a pixelpass host is alive and fanning out (design v3.4 + /// §7.1). `teardown` owns that ordering; see `core::teardown`. + teardown: SessionTeardown, } +/// The session's teardown set, with the echo-cancel guard the platform actually +/// has. On non-Linux there is no AEC module, and `Infallible` makes that +/// structural — the `Option` cannot be `Some`. +#[cfg(target_os = "linux")] +type SessionTeardown = teardown::ScreenshareTeardown< + tokio::process::Child, + crate::audio::echo_cancel::EchoCancelGuard, +>; +#[cfg(not(target_os = "linux"))] +type SessionTeardown = + teardown::ScreenshareTeardown; + impl ActiveSession { async fn shutdown(mut self, audio_backend: Arc) { crate::log_msg("ActiveSession::shutdown started"); // Tear down any screen-share children first so the host stops streaming - // promptly (kill_on_drop is the backstop, but kill explicitly so viewers - // see the stream end without waiting on drop ordering). - if let Some(mut host) = self.screenshare_host.take() { - let _ = host.kill().await; - } - for (_, mut viewer) in self.screenshare_viewers.drain(..) { - let _ = viewer.kill().await; - } + // promptly, and so they are dead *and reaped* well before the AEC guard + // unloads at the end of this function (design v3.4 §7.1). Drop ordering + // is the backstop for the unwind path; this is the path we control. + self.teardown.shutdown_children().await; self.datagram_task.abort(); self.mixer_task.abort(); self.event_task.abort(); @@ -726,8 +729,9 @@ impl ActiveSession { // Unload the echo-cancel module now that the audio streams releasing its // virtual nodes have stopped. (Dropping the guard runs `pactl unload`.) - #[cfg(target_os = "linux")] - drop(self.echo_cancel); + // The screen-share children were killed *and reaped* at the top of this + // function, so nothing pixelpass-side is alive to see the module vanish. + drop(self.teardown); crate::log_msg("Leaving room..."); let _ = self.room_state.leave().await; @@ -1511,6 +1515,8 @@ async fn run_core_loop( biased; maybe_cmd = reliable_rx.recv() => match maybe_cmd { Some(cmd) => cmd, + // Every `CoreController`/`CoreCommandSender` is gone — the UI has + // dropped the core. Teardown happens once, after the loop. None => break, }, maybe_wake = besteffort_wake_rx.recv() => match maybe_wake { @@ -1529,6 +1535,18 @@ async fn run_core_loop( None => continue, } } + // ⚠️ UNREACHABLE BY CONSTRUCTION, twice over — do not mistake this + // for a live teardown path (phase 0b finding, 2026-07-26): + // 1. this function owns `besteffort_wake_tx` (cloned at the + // `CoreController::new` spawn site, used just above for the + // `has_more` re-arm), so the channel can never close while + // this loop is running; + // 2. even without that, every holder of a wake sender — + // `CoreController` and `CoreCommandSender` — holds + // `reliable_tx` too, and the `biased` select polls that one + // first, so the reliable arm always wins the race to exit. + // Teardown is hoisted after the loop, so if this arm is ever made + // reachable it is already covered — nothing to add here. None => break, }, game_change = next_game_change(&mut game_rx) => { @@ -2726,9 +2744,9 @@ async fn run_core_loop( grace_timers, transport: transport.clone(), #[cfg(target_os = "linux")] - echo_cancel: echo_cancel_guard, - screenshare_host: None, - screenshare_viewers: Vec::<(String, tokio::process::Child)>::new(), + teardown: SessionTeardown::new(echo_cancel_guard), + #[cfg(not(target_os = "linux"))] + teardown: SessionTeardown::new(None), }; let self_id = endpoint.id().to_string(); @@ -3403,7 +3421,7 @@ async fn run_core_loop( .await; continue; }; - if session.screenshare_host.is_some() { + if session.teardown.is_sharing() { continue; // already sharing } let bin = match crate::screenshare::pixelpass_path(pixelpass_override.as_deref()) { @@ -3455,7 +3473,7 @@ async fn run_core_loop( { Ok((child, ticket)) => { crate::log_msg("Screen share host started"); - session.screenshare_host = Some(child); + session.teardown.set_host(child); current_sharing = Some(ticket.clone()); let self_state = presence.to_state( is_muted.load(Ordering::Relaxed), @@ -3476,9 +3494,24 @@ async fn run_core_loop( CoreCommand::StopScreenShare => { current_sharing = None; if let Some(session) = &mut active_session { - if let Some(mut child) = session.screenshare_host.take() { - let _ = child.kill().await; - crate::log_msg("Screen share host stopped"); + match session.teardown.stop_host().await { + None => {} + Some(teardown::StopOutcome::Reaped) => { + crate::log_msg("Screen share host stopped"); + } + // We gave up waiting rather than freeze the client, so + // pixelpass may still be alive and serving. Saying + // "stopped" and nothing else would be a lie the user + // cannot see through (round-16 review, P3-2). + Some(teardown::StopOutcome::Unconfirmed) => { + let _ = ui_tx + .send(UiEvent::Error( + "Couldn't confirm the screen-share process exited — \ + it may still be sharing. Check for a stray pixelpass." + .into(), + )) + .await; + } } let self_state = presence.to_state( is_muted.load(Ordering::Relaxed), @@ -3505,16 +3538,12 @@ async fn run_core_loop( if let Some(session) = &mut active_session { // Drop viewers whose player window has already closed so the // list only tracks live players. - session - .screenshare_viewers - .retain_mut(|(_, child)| !matches!(child.try_wait(), Ok(Some(_)))); + session.teardown.sweep_exited_viewers(); // One player per share: a second Watch click on a share we're // already viewing is a retry (usually because the first window // froze), so replace the existing player rather than stacking a // second mpv — two players would double the shared audio. - if let Some(pos) = replace_viewer_index(&session.screenshare_viewers, &ticket) { - let (_, mut old) = session.screenshare_viewers.remove(pos); - let _ = old.kill().await; + if session.teardown.replace_viewer(&ticket).await { crate::log_msg("Screen share viewer replaced (re-watch)"); } } @@ -3522,7 +3551,7 @@ async fn run_core_loop( Ok(child) => { crate::log_msg("Screen share viewer started"); if let Some(session) = &mut active_session { - session.screenshare_viewers.push((ticket, child)); + session.teardown.push_viewer(ticket, child); } } Err(e) => { @@ -3535,6 +3564,21 @@ async fn run_core_loop( } } + // The command loop has exited, by any route. Tear the session down + // explicitly rather than letting it drop on the way out of this function: + // an implicit drop unloads the echo-cancel module without first reaping the + // pixelpass host (design v3.4 §7.2, decision D4). + // + // This sits *after* the loop rather than in the close arm on purpose. The + // impl plan pinned one teardown per channel-close arm, but the best-effort + // wake arm is unreachable by construction (see the comment at that arm), so + // that shape would have duplicated teardown to cover one live path and one + // dead one. Here every `break` is covered structurally, including any added + // later. Adjudication: impl plan §10, 2026-07-26. + if let Some(session) = active_session.take() { + session.shutdown(audio_backend.clone()).await; + } + Ok(()) } diff --git a/src/core/teardown.rs b/src/core/teardown.rs new file mode 100644 index 0000000..2a1caa3 --- /dev/null +++ b/src/core/teardown.rs @@ -0,0 +1,889 @@ +//! Destruction-order guarantees for the screen-share children and the +//! echo-cancel module (phase 0b of the screenshare audio-exclusion plan; +//! design v3.4 §7.1–§7.2, decision D4). +//! +//! # The invariant +//! +//! > **The echo-cancel module must not unload while a pixelpass host is alive +//! > and fanning out.** +//! +//! If it does, the AEC's virtual nodes vanish from under a live pixelpass that +//! still holds link proxies and a stale module index. Phase 6 makes this sharp +//! — it is the first phase whose objects live only as long as pixelpass does — +//! so the ordering guarantee has to exist *before* it. +//! +//! Two paths have to honour it, and only one of them is code we get to run: +//! +//! 1. **The explicit path** — [`ScreenshareTeardown::shutdown_children`], awaited +//! by `ActiveSession::shutdown` before the guard is dropped. +//! 2. **The drop/unwind path** — nobody calls anything. The core has numerous +//! `unwrap()` sites and no `panic=abort` profile, so unwind is reachable, and +//! on that path the only thing standing between us and a violated invariant +//! is *field declaration order* plus [`ReapOnDrop`]. +//! +//! Hence the two structural rules enforced here: +//! +//! - `echo_cancel` is the **last declared field** of [`ScreenshareTeardown`]. +//! Rust drops fields in declaration order, so last-declared is last-dropped. +//! This is not a style choice; reversing it reintroduces the bug. +//! - Killing is not enough — a child must be **reaped**. `kill_on_drop(true)` +//! only *signals*; it hands the child to the runtime's orphan queue and +//! returns, which on an unwinding runtime may never be drained. [`ReapOnDrop`] +//! therefore blocks, briefly and boundedly, until the child is actually gone. +//! +//! Everything here is generic over [`ChildProcess`] and over the guard type so +//! the ordering is unit-testable without spawning processes or loading PipeWire +//! modules — the same seam idiom as `replace_viewer_index` and +//! `rebuild_with_fallback` in the parent module. + +use std::future::Future; +use std::time::{Duration, Instant}; + +/// How long [`ReapOnDrop::drop`] will block waiting for a killed child to be +/// reaped before giving up and logging. This runs on the unwind path, so it is +/// a deliberate trade: a bounded stall is preferable to unloading the AEC out +/// from under a live pixelpass, and unbounded blocking in a `Drop` is not. +const REAP_BUDGET: Duration = Duration::from_millis(250); + +/// Poll interval while waiting out [`REAP_BUDGET`]. +const REAP_POLL: Duration = Duration::from_millis(5); + +/// How long a child gets to honour the graceful stop before it is killed. +/// +/// A healthy pixelpass exits in well under this, so the normal path never +/// spends it; only a wedged child does. It is awaited inline in the core +/// command loop, so it is also how long a wedged child can delay other +/// commands — hence seconds, not tens of seconds. +const STOP_GRACE: Duration = Duration::from_secs(2); + +/// The child-process operations the teardown ordering actually depends on. +/// +/// Deliberately narrow, and deliberately not `ExitStatus`-shaped: the ordering +/// rules care only about *whether* a child has been signalled and *whether* it +/// has been reaped, so the test double is a few lines instead of a fabricated +/// exit status. +pub(super) trait ChildProcess { + /// Ask the child to exit **gracefully**, so it can run its own cleanup. + /// Does **not** wait, and is not guaranteed to be honoured. + fn request_stop(&mut self) -> std::io::Result<()>; + + /// Signal the child to die. Does **not** wait. + fn start_kill(&mut self) -> std::io::Result<()>; + + /// Poll once. `true` once the child has exited **and been reaped**. + fn try_reap(&mut self) -> bool; + + /// Wait until the child has exited and been reaped. + /// + /// The `io::Result` is load-bearing and must not be discarded by callers: + /// a failed wait is *not* a confirmed reap, and treating it as one is how + /// the AEC ends up unloading over a live child. + fn wait_reaped(&mut self) -> impl Future> + Send; +} + +impl ChildProcess for tokio::process::Child { + /// **SIGINT, not SIGTERM.** pixelpass installs only a `tokio::signal::ctrl_c()` + /// handler (`pixelpass/src/common/signal.rs`), so SIGTERM would be the default + /// disposition — instant death, no cleanup — which is indistinguishable from + /// SIGKILL for our purposes. + /// + /// Signalling by pid is safe against pid reuse here because we have not + /// reaped this child: an exited-but-unreaped child is a zombie whose pid the + /// kernel reserves until we `wait` it, so the pid cannot name a stranger. + #[cfg(unix)] + fn request_stop(&mut self) -> std::io::Result<()> { + let Some(pid) = self.id() else { + // Already reaped — nothing to signal. + return Ok(()); + }; + // SAFETY: `kill` is async-signal-safe and takes no pointers; the pid is + // this process's own unreaped child (see above). + if unsafe { libc::kill(pid as libc::pid_t, libc::SIGINT) } == 0 { + Ok(()) + } else { + Err(std::io::Error::last_os_error()) + } + } + + /// Windows has no SIGINT to send to another process without attaching to its + /// console, so the graceful request degrades to the hard kill and the + /// bounded wait below simply returns early. + #[cfg(not(unix))] + fn request_stop(&mut self) -> std::io::Result<()> { + tokio::process::Child::start_kill(self) + } + + fn start_kill(&mut self) -> std::io::Result<()> { + tokio::process::Child::start_kill(self) + } + + fn try_reap(&mut self) -> bool { + matches!(self.try_wait(), Ok(Some(_))) + } + + async fn wait_reaped(&mut self) -> std::io::Result<()> { + self.wait().await.map(|_| ()) + } +} + +/// Did the explicit stop path actually confirm the child was reaped? +/// +/// The distinction is not cosmetic: on [`Unconfirmed`](Self::Unconfirmed) we +/// deliberately stopped waiting (see [`ReapOnDrop::shutdown`]), so pixelpass may +/// still be alive and fanning out. A user-initiated Stop Share must not report +/// that as a clean stop. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +#[must_use = "an unconfirmed stop means the child may still be sharing"] +pub(super) enum StopOutcome { + /// The child is gone and has been reaped. + Reaped, + /// We could not confirm the reap within the bound and gave up waiting. + Unconfirmed, +} + +/// A child that is killed **and reaped** when it is dropped. +/// +/// The explicit path calls [`shutdown`](Self::shutdown), which releases the +/// child only once its reap is *confirmed*, so the `Drop` below is a no-op +/// afterwards but stays armed through every await until then. `Drop` is the +/// last-ditch protection for the panic/unwind/cancellation paths. +pub(super) struct ReapOnDrop { + /// `None` once the child has been reaped through the explicit path. + child: Option, + /// Names the child in the reap-timeout log line. + label: &'static str, +} + +impl ReapOnDrop { + pub(super) fn new(child: C, label: &'static str) -> Self { + Self { + child: Some(child), + label, + } + } + + /// Poll once, without killing. `true` if the child has exited on its own — + /// used to sweep player windows the user has already closed. + pub(super) fn has_exited(&mut self) -> bool { + match &mut self.child { + Some(child) => { + if child.try_reap() { + self.child = None; + true + } else { + false + } + } + // Already reaped through the explicit path. + None => true, + } + } + + /// Stop the child gracefully if it will go, and by force if it will not. + /// Waits for it to be reaped either way. Idempotent. + /// + /// Ask, then insist (design v3.4 §7.4): a pixelpass host that gets SIGINT + /// unloads its capture sink on the way out, whereas SIGKILL skips that and + /// leaks a null-sink module on every Stop Share. + /// + /// The wait is the point: returning after signalling would let the caller + /// proceed to unload the AEC while the child is still running. + /// + /// ⚠️ The child stays owned by `self` across every `.await`, and is released + /// **only after a confirmed reap**. Taking it out first would disarm the + /// `Drop` fallback for exactly as long as the wait lasts: cancel or unwind + /// this future at that moment and the raw child would drop with nothing but + /// `kill_on_drop` (which signals without reaping) while `Drop` below found + /// `None` and did nothing — the precise hole this type exists to close. + pub(super) async fn shutdown(&mut self) -> StopOutcome { + let Some(child) = self.child.as_mut() else { + return StopOutcome::Reaped; + }; + + // Three different things can go wrong here and they want three + // different operator diagnoses: the signal never left (a runtime or + // permission fault), the child ignored it (a wedged pixelpass), or the + // wait itself broke (we no longer know anything about the child). + // Collapsing them into one line was P3-1 of the round-16 review. + if let Err(e) = child.request_stop() { + crate::log_msg(&format!( + "teardown: could not ask {} to stop: {e}", + self.label + )); + } + match tokio::time::timeout(STOP_GRACE, child.wait_reaped()).await { + Ok(Ok(())) => { + self.child = None; + return StopOutcome::Reaped; + } + Ok(Err(e)) => crate::log_msg(&format!( + "teardown: waiting for {} failed ({e}); killing it", + self.label + )), + Err(_) => crate::log_msg(&format!( + "teardown: {} ignored the graceful stop within {STOP_GRACE:?}; killing it", + self.label + )), + } + + if let Err(e) = child.start_kill() { + crate::log_msg(&format!( + "teardown: {} could not be killed: {e}", + self.label + )); + } + + // The second wait is bounded too. An unbounded one lets a process stuck + // in uninterruptible sleep wedge the core command loop forever, and a + // permanently frozen app is a worse failure than the risk below. + if let Ok(Ok(())) = tokio::time::timeout(STOP_GRACE, child.wait_reaped()).await { + self.child = None; + return StopOutcome::Reaped; + } + + // Explicit policy for the one case where the two guarantees conflict: + // we could not confirm the reap and will NOT block indefinitely, so we + // give up availability-first and leave the child owned — `Drop`'s + // bounded retry stays armed, and the AEC may unload over a child that + // is still somehow alive. That residual risk is logged, not silent — + // and, for a user-initiated stop, reported to the caller rather than + // dressed up as success. + crate::log_msg(&format!( + "teardown: {} could not be confirmed dead; the echo-cancel module \ + may unload while it lives", + self.label + )); + StopOutcome::Unconfirmed + } + + /// Is the `Drop` fallback still armed? Test-only: the arming rule is the + /// whole point of holding the child across the waits. + #[cfg(test)] + fn is_armed(&self) -> bool { + self.child.is_some() + } +} + +impl Drop for ReapOnDrop { + fn drop(&mut self) { + let Some(child) = self.child.as_mut() else { + return; + }; + let _ = child.start_kill(); + // `Drop` cannot await, so poll on a bounded budget. See `REAP_BUDGET`. + let deadline = Instant::now() + REAP_BUDGET; + loop { + if child.try_reap() { + return; + } + if Instant::now() >= deadline { + crate::log_msg(&format!( + "teardown: {} did not exit within the reap budget; \ + continuing (the echo-cancel module may unload while it lives)", + self.label + )); + return; + } + std::thread::sleep(REAP_POLL); + } + } +} + +/// Everything in an `ActiveSession` whose **destruction order** is load-bearing. +/// +/// ⚠️ Field order below **is** the invariant. `echo_cancel` is declared last so +/// it is dropped last, after every screen-share child has been killed and +/// reaped. Do not reorder these fields. +pub(super) struct ScreenshareTeardown { + /// Our pixelpass screen-share host child while sharing. + host: Option>, + /// pixelpass viewer children we spawned to watch peers' shares, each paired + /// with the share ticket it is viewing so a re-watch of the same share can + /// replace (not stack) its player. + viewers: Vec<(String, ReapOnDrop)>, + /// Loaded PipeWire echo-cancel module (if enabled); unloads on drop. + /// + /// ⚠️ **LAST FIELD ON PURPOSE** — see the module docs and the struct note. + /// + /// Never read, and that is the design: the guard is held only so that its + /// `Drop` runs, and only so that it runs *here*, last. `dead_code` is right + /// that nothing reads it and wrong that it does nothing. + #[allow(dead_code)] + echo_cancel: Option, +} + +impl ScreenshareTeardown { + pub(super) fn new(echo_cancel: Option) -> Self { + Self { + host: None, + viewers: Vec::new(), + echo_cancel, + } + } + + pub(super) fn is_sharing(&self) -> bool { + self.host.is_some() + } + + pub(super) fn set_host(&mut self, child: C) { + self.host = Some(ReapOnDrop::new(child, "screen-share host")); + } + + /// Stop sharing: kill the host and wait for it to be reaped. `None` if we + /// were not sharing; otherwise whether the reap was actually confirmed — + /// the caller owns telling the user, since an unconfirmed stop may leave + /// pixelpass fanning out after the UI says sharing ended. + pub(super) async fn stop_host(&mut self) -> Option { + let mut host = self.host.take()?; + Some(host.shutdown().await) + } + + /// Drop viewers whose player window has already closed, so the list only + /// tracks live players. + pub(super) fn sweep_exited_viewers(&mut self) { + self.viewers.retain_mut(|(_, child)| !child.has_exited()); + } + + /// Kill and reap the viewer already showing `ticket`, if any, so a re-watch + /// replaces its player instead of stacking a second one. + pub(super) async fn replace_viewer(&mut self, ticket: &str) -> bool { + let Some(pos) = super::replace_viewer_index(&self.viewers, ticket) else { + return false; + }; + let (_, mut old) = self.viewers.remove(pos); + // A viewer is our own player window, not the thing peers are watching: + // an unconfirmed reap is already logged, and there is no user decision + // riding on it the way there is for Stop Share. + let _ = old.shutdown().await; + true + } + + pub(super) fn push_viewer(&mut self, ticket: String, child: C) { + self.viewers + .push((ticket, ReapOnDrop::new(child, "screen-share viewer"))); + } + + /// Kill and reap **every** screen-share child, host first so viewers see the + /// stream end promptly. + /// + /// The caller must await this before the echo-cancel guard is dropped. On + /// the drop/unwind path nothing calls it and field order carries the + /// invariant instead. + pub(super) async fn shutdown_children(&mut self) { + // Outcomes are discarded on purpose: this runs on the session/teardown + // path, where the policy is already availability-first and the residual + // risk is logged by `shutdown` itself. There is no user still waiting + // on an answer here, unlike `stop_host`. + if let Some(host) = &mut self.host { + let _ = host.shutdown().await; + } + self.host = None; + for (_, viewer) in self.viewers.iter_mut() { + let _ = viewer.shutdown().await; + } + self.viewers.clear(); + } +} + +#[cfg(test)] +mod tests { + use super::{ChildProcess, ReapOnDrop, STOP_GRACE, ScreenshareTeardown, StopOutcome}; + use std::future::Future; + use std::sync::{Arc, Mutex}; + use std::time::Duration; + + type Log = Arc>>; + + fn log() -> Log { + Arc::new(Mutex::new(Vec::new())) + } + + fn entries(log: &Log) -> Vec { + log.lock().unwrap().clone() + } + + fn position(log: &Log, entry: &str) -> Option { + entries(log).iter().position(|e| e == entry) + } + + /// Records the events the ordering rules turn on. Death is gated on an + /// actual signal, so the double cannot report a reap that nothing caused. + struct FakeChild { + log: Log, + label: &'static str, + interrupted: bool, + killed: bool, + reaped: bool, + /// A well-behaved child exits on SIGINT. A wedged one ignores it and + /// dies only to SIGKILL. + honours_interrupt: bool, + /// When true the child is already dead before anyone signals it — the + /// closed-player-window case that `sweep_exited_viewers` looks for. + exited_on_its_own: bool, + /// Death is not instantaneous: `try_reap` reports the child alive this + /// many more times before it goes. + polls_before_death: u32, + /// `wait` reports an error instead of a reap. + wait_fails: bool, + } + + impl FakeChild { + /// A well-behaved child: exits when asked. + fn new(log: &Log, label: &'static str) -> Self { + Self { + log: log.clone(), + label, + interrupted: false, + killed: false, + reaped: false, + honours_interrupt: true, + exited_on_its_own: false, + polls_before_death: 0, + wait_fails: false, + } + } + + /// A child that ignores the graceful stop entirely. + fn wedged(log: &Log, label: &'static str) -> Self { + Self { + honours_interrupt: false, + ..Self::new(log, label) + } + } + + /// A child that does not die the instant it is signalled: `try_reap` + /// reports it alive for `polls` calls first. Without this the `Drop` + /// polling loop could be replaced by a single `try_reap` and no test + /// would notice. + fn reaps_after_polls(log: &Log, label: &'static str, polls: u32) -> Self { + Self { + polls_before_death: polls, + ..Self::new(log, label) + } + } + + /// A child that ignores SIGINT *and* does not die the instant it is + /// killed — the only shape that lets a test reach the post-SIGKILL + /// wait and still be reaped by the `Drop` poll loop afterwards. + fn wedged_then_dies_after_polls(log: &Log, label: &'static str, polls: u32) -> Self { + Self { + honours_interrupt: false, + polls_before_death: polls, + ..Self::new(log, label) + } + } + + /// A child whose `wait` fails. A failed wait is not a confirmed reap, + /// so it must not be reported as one. + fn wait_fails(log: &Log, label: &'static str) -> Self { + Self { + wait_fails: true, + ..Self::new(log, label) + } + } + + fn already_exited(log: &Log, label: &'static str) -> Self { + Self { + exited_on_its_own: true, + ..Self::new(log, label) + } + } + + /// Has anything actually made this child exit yet? A signalled child + /// still has to burn through `polls_before_death` first. + fn is_dead(&self) -> bool { + let signalled = self.killed + || self.exited_on_its_own + || (self.interrupted && self.honours_interrupt); + signalled && self.polls_before_death == 0 + } + + /// One observation of a dying-but-not-yet-dead child. + fn tick(&mut self) { + self.polls_before_death = self.polls_before_death.saturating_sub(1); + } + + fn record(&self, event: &str) { + self.log + .lock() + .unwrap() + .push(format!("{}:{event}", self.label)); + } + + fn mark_reaped(&mut self) { + if !self.reaped { + self.reaped = true; + self.record("reap"); + } + } + } + + impl ChildProcess for FakeChild { + fn request_stop(&mut self) -> std::io::Result<()> { + if !self.interrupted { + self.interrupted = true; + self.record("sigint"); + } + Ok(()) + } + + fn start_kill(&mut self) -> std::io::Result<()> { + if !self.killed { + self.killed = true; + self.record("kill"); + } + Ok(()) + } + + fn try_reap(&mut self) -> bool { + if self.is_dead() { + self.mark_reaped(); + return true; + } + self.tick(); + false + } + + /// Pending until something actually kills the child, so a wedged child + /// really does make the caller wait out `STOP_GRACE`. No waker is + /// registered: under `start_paused` the runtime auto-advances its clock + /// when every task is idle, which is exactly what fires the timeout. + fn wait_reaped(&mut self) -> impl Future> + Send { + std::future::poll_fn(move |_cx| { + if self.wait_fails { + return std::task::Poll::Ready(Err(std::io::Error::other("wait failed"))); + } + if self.is_dead() { + self.mark_reaped(); + std::task::Poll::Ready(Ok(())) + } else { + std::task::Poll::Pending + } + }) + } + } + + /// Stands in for `EchoCancelGuard`, whose real `Drop` runs `pactl unload`. + struct FakeAec(Log); + + impl Drop for FakeAec { + fn drop(&mut self) { + self.0.lock().unwrap().push("aec:unload".to_string()); + } + } + + fn teardown(log: &Log) -> ScreenshareTeardown { + ScreenshareTeardown::new(Some(FakeAec(log.clone()))) + } + + // --- The drop/unwind path: field order + ReapOnDrop carry the invariant --- + + /// Mutation gate #5 (remove the reap loop from `ReapOnDrop::drop`). + /// + /// Asserts only that dropping a guard reaps, and reaps *after* killing — + /// deliberately says nothing about the AEC, so reversing the struct's field + /// order leaves this test green and only the ordering test below fails. + #[test] + fn dropping_a_guard_kills_and_then_reaps_the_child() { + let log = log(); + drop(ReapOnDrop::new(FakeChild::new(&log, "host"), "host")); + assert_eq!(entries(&log), vec!["host:kill", "host:reap"]); + } + + /// Mutation gate #4 (reverse the field order of `ScreenshareTeardown`). + /// + /// Asserts only kill-before-unload, so removing the reap loop leaves this + /// test green and only the reap test above fails. + #[test] + fn the_aec_unloads_after_the_children_on_the_drop_path() { + let log = log(); + let mut t = teardown(&log); + t.set_host(FakeChild::new(&log, "host")); + t.push_viewer("ticket-A".to_string(), FakeChild::new(&log, "viewer")); + drop(t); + + let unload = position(&log, "aec:unload").expect("the AEC guard must be dropped"); + let host_kill = position(&log, "host:kill").expect("the host must be killed"); + let viewer_kill = position(&log, "viewer:kill").expect("the viewer must be killed"); + assert!( + host_kill < unload, + "the AEC unloaded while the host was alive: {:?}", + entries(&log) + ); + assert!( + viewer_kill < unload, + "the AEC unloaded while a viewer was alive: {:?}", + entries(&log) + ); + } + + /// The whole invariant in one sequence, as documentation. + #[test] + fn the_drop_path_reaps_every_child_before_unloading_the_aec() { + let log = log(); + let mut t = teardown(&log); + t.set_host(FakeChild::new(&log, "host")); + drop(t); + assert_eq!(entries(&log), vec!["host:kill", "host:reap", "aec:unload"]); + } + + // --- The explicit path: ask, then insist --- + + /// A healthy child must be *asked*, never killed. If Stop Share went + /// straight to SIGKILL, pixelpass would skip its own cleanup and leak a + /// null-sink module every time (design v3.4 §7.4). + #[tokio::test] + async fn a_healthy_child_is_asked_to_stop_and_never_killed() { + let log = log(); + let mut t = teardown(&log); + t.set_host(FakeChild::new(&log, "host")); + + assert_eq!(t.stop_host().await, Some(StopOutcome::Reaped)); + + assert_eq!(entries(&log), vec!["host:sigint", "host:reap"]); + assert!( + !entries(&log).contains(&"host:kill".to_string()), + "a child that honoured the graceful stop must not be killed: {:?}", + entries(&log) + ); + } + + /// ...but a child that ignores the request must not be able to hold the + /// session open forever: the grace is bounded and SIGKILL follows. + #[tokio::test(start_paused = true)] + async fn a_wedged_child_is_killed_once_the_grace_expires() { + let log = log(); + let mut t = teardown(&log); + t.set_host(FakeChild::wedged(&log, "host")); + + // The outer bound turns "the fallback was removed" into a failure + // rather than a hung test. Under `start_paused` no real time passes. + let start = tokio::time::Instant::now(); + tokio::time::timeout(Duration::from_secs(60), t.stop_host()) + .await + .expect("a wedged child must not block teardown indefinitely"); + + assert_eq!(entries(&log), vec!["host:sigint", "host:kill", "host:reap"]); + assert!( + start.elapsed() >= STOP_GRACE, + "the child must actually be given the grace period, waited {:?}", + start.elapsed() + ); + } + + /// The assertion above compares elapsed time against `STOP_GRACE` itself, + /// so it stays vacuously true if the constant is set to zero — both sides + /// move together. Pin the constant independently: the whole point of the + /// graceful stop is that pixelpass gets a real interval in which to unload + /// its capture sink, and zero is not one. + #[test] + fn the_grace_is_a_real_interval() { + assert!( + STOP_GRACE >= Duration::from_millis(500), + "too short to let pixelpass tear its pipeline down: {STOP_GRACE:?}" + ); + // ...and short enough that a wedged child cannot visibly stall the core + // command loop, which awaits this inline. + assert!( + STOP_GRACE <= Duration::from_secs(5), + "long enough to freeze the UI's command handling: {STOP_GRACE:?}" + ); + } + + /// The hole the whole type exists to close, and the one place the old + /// implementation left open: if `shutdown` is cancelled while waiting, the + /// child must still be owned, so dropping the guard still kills and reaps. + #[tokio::test(start_paused = true)] + async fn cancelling_shutdown_mid_wait_leaves_the_fallback_armed() { + let log = log(); + let mut guard = ReapOnDrop::new(FakeChild::wedged(&log, "host"), "host"); + + // Cancel well inside the grace, while it is still waiting. + assert!( + tokio::time::timeout(STOP_GRACE / 4, guard.shutdown()) + .await + .is_err(), + "the wedged child should still have been waiting when we cancelled" + ); + assert!( + guard.is_armed(), + "a cancelled shutdown must not disarm the drop fallback" + ); + + drop(guard); + assert_eq!(entries(&log), vec!["host:sigint", "host:kill", "host:reap"]); + } + + /// The test above only ever cancels during the *graceful* wait, so a + /// mutation that disarmed the wrapper between the two waits would survive + /// it (round-16 review, P3-3). This one cancels during the post-SIGKILL + /// wait — the window where we have already given up on cooperation and the + /// `Drop` fallback is the only thing left. + #[tokio::test(start_paused = true)] + async fn cancelling_shutdown_after_the_kill_leaves_the_fallback_armed() { + let log = log(); + // Ignores SIGINT, so the grace expires and we reach the kill; then + // survives three polls, so the second wait is still pending when we + // cancel, and the drop loop still gets to reap it. + let mut guard = ReapOnDrop::new( + FakeChild::wedged_then_dies_after_polls(&log, "host", 3), + "host", + ); + + assert!( + tokio::time::timeout(STOP_GRACE + STOP_GRACE / 4, guard.shutdown()) + .await + .is_err(), + "we should have been cancelled inside the post-kill wait" + ); + assert_eq!( + entries(&log), + vec!["host:sigint", "host:kill"], + "the graceful stop must have expired and escalated before we cancelled" + ); + assert!( + guard.is_armed(), + "cancelling after the kill must not disarm the drop fallback either" + ); + + drop(guard); + // The fake's `start_kill` is idempotent, so `Drop` re-signalling an + // already-killed child adds no entry; the *reap* is what proves the + // fallback ran to completion after we abandoned the wait. + assert_eq!( + entries(&log), + vec!["host:sigint", "host:kill", "host:reap"], + "Drop must poll until the child is actually gone" + ); + } + + /// A failed wait is not a reap. Reporting it as one is how the AEC ends up + /// unloading over a child that is still alive. + #[tokio::test(start_paused = true)] + async fn a_failed_wait_is_not_treated_as_a_confirmed_reap() { + let log = log(); + let mut guard = ReapOnDrop::new(FakeChild::wait_fails(&log, "host"), "host"); + + assert_eq!( + guard.shutdown().await, + StopOutcome::Unconfirmed, + "a stop we could not confirm must not be reported as a clean one" + ); + + assert!( + !entries(&log).contains(&"host:reap".to_string()), + "nothing confirmed the reap: {:?}", + entries(&log) + ); + assert!( + entries(&log).contains(&"host:kill".to_string()), + "a child that would not stop must still be escalated: {:?}", + entries(&log) + ); + assert!( + guard.is_armed(), + "an unconfirmed reap must leave the drop fallback armed" + ); + } + + /// Death is not instantaneous, so the drop path has to keep polling. A + /// single `try_reap` in place of the loop must not pass. + #[test] + fn the_drop_path_polls_until_the_child_is_actually_gone() { + let log = log(); + drop(ReapOnDrop::new( + FakeChild::reaps_after_polls(&log, "host", 3), + "host", + )); + assert_eq!(entries(&log), vec!["host:kill", "host:reap"]); + } + + /// Mutation gate #3 (remove the wait after the host kill). + #[tokio::test] + async fn explicit_shutdown_reaps_the_host_before_the_aec_can_unload() { + let log = log(); + let mut t = teardown(&log); + t.set_host(FakeChild::new(&log, "host")); + t.push_viewer("ticket-A".to_string(), FakeChild::new(&log, "viewer")); + + t.shutdown_children().await; + + // Reaped by the explicit path — before the guard is anywhere near dropped. + assert_eq!( + entries(&log), + vec!["host:sigint", "host:reap", "viewer:sigint", "viewer:reap"], + "children must be stopped and reaped by the explicit path" + ); + + drop(t); + let unload = position(&log, "aec:unload").expect("the AEC guard must be dropped"); + let host_reap = position(&log, "host:reap").expect("the host must be reaped"); + assert!(host_reap < unload); + } + + #[tokio::test] + async fn explicit_shutdown_is_idempotent_with_the_drop_path() { + let log = log(); + let mut t = teardown(&log); + t.set_host(FakeChild::new(&log, "host")); + t.shutdown_children().await; + drop(t); + // Exactly one stop and one reap: the drop path must not re-signal a + // child the explicit path already took. + assert_eq!( + entries(&log), + vec!["host:sigint", "host:reap", "aec:unload"] + ); + } + + // --- Host/viewer bookkeeping --- + + #[tokio::test] + async fn stop_host_reports_whether_it_was_sharing() { + let log = log(); + let mut t = teardown(&log); + assert!(!t.is_sharing()); + assert_eq!(t.stop_host().await, None, "not sharing: nothing to stop"); + + t.set_host(FakeChild::new(&log, "host")); + assert!(t.is_sharing()); + assert_eq!(t.stop_host().await, Some(StopOutcome::Reaped)); + assert!(!t.is_sharing()); + assert_eq!(entries(&log), vec!["host:sigint", "host:reap"]); + } + + #[test] + fn sweeping_drops_only_the_players_that_already_closed() { + let log = log(); + let mut t = teardown(&log); + t.push_viewer( + "closed".to_string(), + FakeChild::already_exited(&log, "closed"), + ); + t.push_viewer("live".to_string(), FakeChild::new(&log, "live")); + + t.sweep_exited_viewers(); + + // The live player survives the sweep; only the closed one is dropped, + // and dropping it must not kill anything (it was already gone). + assert_eq!(t.viewers.len(), 1); + assert_eq!(t.viewers[0].0, "live"); + assert_eq!(entries(&log), vec!["closed:reap"]); + } + + #[tokio::test] + async fn re_watching_a_share_replaces_that_player_only() { + let log = log(); + let mut t = teardown(&log); + t.push_viewer("ticket-A".to_string(), FakeChild::new(&log, "a")); + t.push_viewer("ticket-B".to_string(), FakeChild::new(&log, "b")); + + assert!(t.replace_viewer("ticket-A").await); + assert_eq!(entries(&log), vec!["a:sigint", "a:reap"]); + assert_eq!(t.viewers.len(), 1); + assert_eq!(t.viewers[0].0, "ticket-B"); + + // A share we are not watching has nothing to replace. + assert!(!t.replace_viewer("ticket-C").await); + } +}