core/teardown: reap the screen-share children before the AEC unloads
Phase 0b, fixes 1-3 of design v3.4 §7.2 (decision D4). The invariant is that the echo-cancel module must not unload while a pixelpass host is alive and fanning out; two paths have to honour it and only one is code we get to run. The explicit path: `ActiveSession::shutdown` now awaits `ScreenshareTeardown::shutdown_children`, and the reliable command channel's close arm tears the session down explicitly instead of letting it drop on the way out of `run_core_loop`. The drop/unwind path: the ordering-critical fields move out of `ActiveSession` into `core::teardown::ScreenshareTeardown`, where `echo_cancel` is the LAST declared field and therefore the last dropped. Previously it was declared first (`:682`, ahead of `screenshare_host` at `:685`), so an unwind unloaded the AEC while the host was still live — and unwind is reachable, the core is full of `unwrap()` and has no `panic=abort` profile. Killing is not enough. `kill_on_drop(true)` only signals: it hands the child to the runtime's orphan queue and returns, which an unwinding runtime may never drain. `ReapOnDrop` blocks on a bounded 250 ms budget until the child is really gone, because a bounded stall beats unloading the AEC out from under a live pixelpass. Everything is generic over a narrow `ChildProcess` trait and over the guard type, so ordering is unit-testable without spawning processes or loading PipeWire modules — the seam idiom already used by `replace_viewer_index`. Mutation-verified, and the plan's demand that mutations 4 and 5 prove *different* defenses holds: reversing the field order fails only the AEC-ordering tests and leaves the reap test green; removing the reap loop fails only the reap tests and leaves the ordering test green. Removing the explicit wait fails the explicit-path tests. 631 lib tests, clippy clean, fmt clean. ⚠️ Mutations 1 and 2 of the pinned matrix do not both exist: the best-effort wake arm is unreachable by construction, twice over. Documented at the site; adjudication owed in the impl plan. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
+57
-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,7 +1515,17 @@ async fn run_core_loop(
|
||||
biased;
|
||||
maybe_cmd = reliable_rx.recv() => match maybe_cmd {
|
||||
Some(cmd) => cmd,
|
||||
None => break,
|
||||
// Every `CoreController`/`CoreCommandSender` is gone — the UI has
|
||||
// dropped the core. Tear the session down explicitly instead of
|
||||
// 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).
|
||||
None => {
|
||||
if let Some(session) = active_session.take() {
|
||||
session.shutdown(audio_backend.clone()).await;
|
||||
}
|
||||
break;
|
||||
}
|
||||
},
|
||||
maybe_wake = besteffort_wake_rx.recv() => match maybe_wake {
|
||||
Some(()) => {
|
||||
@@ -1529,6 +1543,18 @@ async fn run_core_loop(
|
||||
None => continue,
|
||||
}
|
||||
}
|
||||
// ⚠️ UNREACHABLE BY CONSTRUCTION, twice over — do not mistake this
|
||||
// for a tested 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.
|
||||
// The teardown therefore lives in the reliable arm above. If this
|
||||
// arm is ever made reachable, it needs the same `shutdown().await`.
|
||||
None => break,
|
||||
},
|
||||
game_change = next_game_change(&mut game_rx) => {
|
||||
@@ -2726,9 +2752,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 +3429,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 +3481,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,8 +3502,7 @@ 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;
|
||||
if session.teardown.stop_host().await {
|
||||
crate::log_msg("Screen share host stopped");
|
||||
}
|
||||
let self_state = presence.to_state(
|
||||
@@ -3505,16 +3530,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 +3543,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) => {
|
||||
|
||||
@@ -0,0 +1,485 @@
|
||||
//! 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);
|
||||
|
||||
/// 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 {
|
||||
/// 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.
|
||||
fn wait_reaped(&mut self) -> impl Future<Output = ()> + Send;
|
||||
}
|
||||
|
||||
impl ChildProcess for tokio::process::Child {
|
||||
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) {
|
||||
let _ = self.wait().await;
|
||||
}
|
||||
}
|
||||
|
||||
/// A child that is killed **and reaped** when it is dropped.
|
||||
///
|
||||
/// The explicit path calls [`shutdown`](Self::shutdown), which takes the child
|
||||
/// out, so the `Drop` below is a no-op afterwards. `Drop` is the last-ditch
|
||||
/// protection for the panic/unwind path only.
|
||||
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,
|
||||
}
|
||||
}
|
||||
|
||||
/// Kill the child and wait for it to be reaped. Idempotent.
|
||||
///
|
||||
/// The wait is the point: returning after `start_kill` would let the caller
|
||||
/// proceed to unload the AEC while the child is still running.
|
||||
pub(super) async fn shutdown(&mut self) {
|
||||
let Some(mut child) = self.child.take() else {
|
||||
return;
|
||||
};
|
||||
let _ = child.start_kill();
|
||||
child.wait_reaped().await;
|
||||
}
|
||||
}
|
||||
|
||||
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.
|
||||
pub(super) async fn stop_host(&mut self) -> bool {
|
||||
let Some(mut host) = self.host.take() else {
|
||||
return false;
|
||||
};
|
||||
host.shutdown().await;
|
||||
true
|
||||
}
|
||||
|
||||
/// 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);
|
||||
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) {
|
||||
if let Some(host) = &mut self.host {
|
||||
host.shutdown().await;
|
||||
}
|
||||
self.host = None;
|
||||
for (_, viewer) in self.viewers.iter_mut() {
|
||||
viewer.shutdown().await;
|
||||
}
|
||||
self.viewers.clear();
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::{ChildProcess, ReapOnDrop, ScreenshareTeardown};
|
||||
use std::future::Future;
|
||||
use std::sync::{Arc, Mutex};
|
||||
|
||||
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 two events the ordering rules turn on. `killed` gates
|
||||
/// reaping so the double cannot report a reap that never followed a kill.
|
||||
struct FakeChild {
|
||||
log: Log,
|
||||
label: &'static str,
|
||||
killed: bool,
|
||||
reaped: bool,
|
||||
/// When true the child is already dead before anyone kills it — the
|
||||
/// closed-player-window case that `sweep_exited_viewers` looks for.
|
||||
exited_on_its_own: bool,
|
||||
}
|
||||
|
||||
impl FakeChild {
|
||||
fn new(log: &Log, label: &'static str) -> Self {
|
||||
Self {
|
||||
log: log.clone(),
|
||||
label,
|
||||
killed: false,
|
||||
reaped: false,
|
||||
exited_on_its_own: false,
|
||||
}
|
||||
}
|
||||
|
||||
fn already_exited(log: &Log, label: &'static str) -> Self {
|
||||
Self {
|
||||
exited_on_its_own: true,
|
||||
..Self::new(log, label)
|
||||
}
|
||||
}
|
||||
|
||||
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 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.killed || self.exited_on_its_own {
|
||||
self.mark_reaped();
|
||||
return true;
|
||||
}
|
||||
false
|
||||
}
|
||||
|
||||
fn wait_reaped(&mut self) -> impl Future<Output = ()> + Send {
|
||||
self.mark_reaped();
|
||||
std::future::ready(())
|
||||
}
|
||||
}
|
||||
|
||||
/// 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 ---
|
||||
|
||||
/// 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:kill", "host:reap", "viewer:kill", "viewer:reap",],
|
||||
"children must be killed 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 kill and one reap: the drop path must not re-signal a
|
||||
// child the explicit path already took.
|
||||
assert_eq!(entries(&log), vec!["host:kill", "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!(!t.stop_host().await, "not sharing: nothing to stop");
|
||||
|
||||
t.set_host(FakeChild::new(&log, "host"));
|
||||
assert!(t.is_sharing());
|
||||
assert!(t.stop_host().await);
|
||||
assert!(!t.is_sharing());
|
||||
assert_eq!(entries(&log), vec!["host:kill", "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:kill", "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