Merge phase 0b + the peerspeak half of 0c: teardown ordering, reaping, graceful stop
Two invariants land here, both of which phase 6 depends on: 1. The echo-cancel module cannot unload while a pixelpass host is alive. The ordering-critical fields moved into `ScreenshareTeardown` with `echo_cancel` declared LAST, and killing is no longer taken for reaping — `kill_on_drop` only signals, so `ReapOnDrop` blocks on a bounded poll until the child is actually gone. The defect was real, not theoretical: `echo_cancel` sat ahead of `screenshare_host` in declaration order, so any unwind unloaded the AEC first, and unwind is reachable (no `panic=abort`, many `unwrap()`s). 2. Stop Share asks before it insists — SIGINT, a bounded grace, then SIGKILL — so pixelpass runs its own cleanup instead of leaking a null-sink module every time. SIGINT specifically: pixelpass installs only a `ctrl_c()` handler. Reviewed by Codex across two rounds: changes-requested (two blocking findings, both real, both the same shape — a defence disarmed exactly when it was needed) then approve-with-follow-ups (five P3s, all applied). The mutation matrix was revised from five to four after one pinned mutation was proved unreachable by construction, and teardown was hoisted to one unconditional post-loop site so every loop exit is covered structurally. Owed and recorded: the live Stop Share SIGINT gate has never been field-run, and the hoisted call site's live proof belongs to the phase-9 lifecycle row. 638 lib tests, clippy clean, fmt clean.
This commit is contained in:
Generated
+1
@@ -4883,6 +4883,7 @@ dependencies = [
|
||||
"image",
|
||||
"iroh",
|
||||
"iroh-gossip",
|
||||
"libc",
|
||||
"opus",
|
||||
"pipewire",
|
||||
"rand 0.10.1",
|
||||
|
||||
@@ -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"
|
||||
|
||||
@@ -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_<pid> 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 **`<node.name>.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=<node.name>.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<StopOutcome>`, 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
|
||||
|
||||
+80
-36
@@ -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<IrohTransport>,
|
||||
/// Loaded PipeWire echo-cancel module (if enabled); unloads on drop.
|
||||
#[cfg(target_os = "linux")]
|
||||
echo_cancel: Option<crate::audio::echo_cancel::EchoCancelGuard>,
|
||||
/// 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<tokio::process::Child>,
|
||||
/// 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<tokio::process::Child, std::convert::Infallible>;
|
||||
|
||||
impl ActiveSession {
|
||||
async fn shutdown(mut self, audio_backend: Arc<PlatformAudioBackend>) {
|
||||
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(())
|
||||
}
|
||||
|
||||
|
||||
@@ -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<Output = std::io::Result<()>> + 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<C: ChildProcess> {
|
||||
/// `None` once the child has been reaped through the explicit path.
|
||||
child: Option<C>,
|
||||
/// Names the child in the reap-timeout log line.
|
||||
label: &'static str,
|
||||
}
|
||||
|
||||
impl<C: ChildProcess> ReapOnDrop<C> {
|
||||
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<C: ChildProcess> Drop for ReapOnDrop<C> {
|
||||
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<C: ChildProcess, G> {
|
||||
/// Our pixelpass screen-share host child while sharing.
|
||||
host: Option<ReapOnDrop<C>>,
|
||||
/// 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<C>)>,
|
||||
/// 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<G>,
|
||||
}
|
||||
|
||||
impl<C: ChildProcess, G> ScreenshareTeardown<C, G> {
|
||||
pub(super) fn new(echo_cancel: Option<G>) -> 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<StopOutcome> {
|
||||
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<Mutex<Vec<String>>>;
|
||||
|
||||
fn log() -> Log {
|
||||
Arc::new(Mutex::new(Vec::new()))
|
||||
}
|
||||
|
||||
fn entries(log: &Log) -> Vec<String> {
|
||||
log.lock().unwrap().clone()
|
||||
}
|
||||
|
||||
fn position(log: &Log, entry: &str) -> Option<usize> {
|
||||
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<Output = std::io::Result<()>> + 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<FakeChild, FakeAec> {
|
||||
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);
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user