peerspeak/tests/screenshare_host_fault.rs

//! S2 exit gate: a pixelpass host that dies mid-share must be torn down —
//! reaped, pulled off presence, `ScreenShareStopped` emitted **before** the
//! explanatory error — and a host stopped *deliberately* must NOT produce that
//! error when its stdout EOF arrives late (the staleness gate).
//!
//! Drives the real core loop end to end through `CoreController`, with the
//! pixelpass override pointed at fake shell scripts: one that emits a ticket
//! and dies, one that emits a ticket and lives until signalled. This is the
//! only harness that reaches the core's fault handler — the command loop has
//! no unit seam — so these two halves are what kill the "forwarder drops the
//! Eof" and "handler ignores the generation" mutants.
//!
//! Live: joins a real (solo) room, so it needs a working audio backend and
//! network access for the endpoint bind.
//! `cargo test --test screenshare_host_fault -- --ignored`

#![cfg(unix)]

use std::os::unix::fs::PermissionsExt;
use std::path::PathBuf;
use std::time::Duration;

use peerspeak::core::CoreController;
use peerspeak::core::messages::{CoreCommand, UiEvent};

const EVENT_TIMEOUT: Duration = Duration::from_secs(20);
/// How long to listen for events that must NOT arrive. Comfortably past the
/// fake host's exit plus the drain/forwarder hop, so a stale fault that WOULD
/// be mishandled has arrived by the end of it.
const QUIET_WINDOW: Duration = Duration::from_secs(3);

/// Removes the fake-pixelpass dir even when an assertion panics mid-test
/// (a plain trailing `remove_dir_all` never runs on an unwind).
struct TempDir(PathBuf);

impl Drop for TempDir {
    fn drop(&mut self) {
        std::fs::remove_dir_all(&self.0).ok();
    }
}

fn write_fake_pixelpass(dir: &std::path::Path, name: &str, body: &str) -> PathBuf {
    let path = dir.join(name);
    std::fs::write(&path, body).expect("write fake pixelpass");
    std::fs::set_permissions(&path, std::fs::Permissions::from_mode(0o755))
        .expect("chmod fake pixelpass");
    path
}

/// Skip events until `pick` matches, panicking after [`EVENT_TIMEOUT`].
/// Unrelated events (identity, presence, chat plumbing) flow on this channel
/// too, so gates scan rather than assert exact sequences.
async fn wait_for<T>(
    rx: &mut tokio::sync::mpsc::Receiver<UiEvent>,
    what: &str,
    mut pick: impl FnMut(&UiEvent) -> Option<T>,
) -> T {
    let deadline = tokio::time::Instant::now() + EVENT_TIMEOUT;
    loop {
        let ev = tokio::time::timeout_at(deadline, rx.recv())
            .await
            .unwrap_or_else(|_| panic!("timed out waiting for {what}"))
            .unwrap_or_else(|| panic!("ui channel closed waiting for {what}"));
        if let Some(v) = pick(&ev) {
            return v;
        }
    }
}

#[tokio::test]
#[ignore = "live: joins a real solo room (audio backend + network bind)"]
async fn a_dead_host_is_torn_down_and_a_clean_stop_stays_clean() {
    let dir_guard =
        TempDir(std::env::temp_dir().join(format!("peerspeak-hostfault-{}", std::process::id())));
    let dir = dir_guard.0.clone();
    std::fs::create_dir_all(&dir).unwrap();

    // Half 1's host: emits its ticket, then dies on its own — the S2 defect
    // scenario. Plain `sleep` (no exec) so the shell itself exits and closes
    // stdout with no orphan holding the pipe.
    let dying_host = write_fake_pixelpass(
        &dir,
        "pixelpass-dies",
        "#!/bin/sh\necho '{\"event\":\"ticket\",\"value\":\"fake-ticket-dies\"}'\nsleep 1\n",
    );
    // Half 2's host: lives until signalled. `exec` so the SIGINT from Stop
    // Share hits the sleep itself — the process dies AND its stdout closes,
    // which is exactly what makes the late Eof arrive and exercise the
    // staleness gate rather than vacuously never sending a fault.
    let living_host = write_fake_pixelpass(
        &dir,
        "pixelpass-lives",
        "#!/bin/sh\necho '{\"event\":\"ticket\",\"value\":\"fake-ticket-lives\"}'\nexec sleep 600\n",
    );

    let (ui_tx, mut ui_rx) = tokio::sync::mpsc::channel(256);
    let controller = CoreController::new(ui_tx);

    assert!(controller.send(CoreCommand::SetPixelpassPath(Some(
        dying_host.to_string_lossy().into_owned()
    ))));
    assert!(controller.send(CoreCommand::Join {
        name: "host-fault-gate".into(),
        ticket: "create".into(),
        room_name: "s2".into(),
        input_device: None,
        output_device: None,
        echo_cancellation: false,
        avatar: Default::default(),
    }));
    wait_for(&mut ui_rx, "RoomJoined", |ev| match ev {
        UiEvent::RoomJoined { .. } => Some(()),
        UiEvent::Error(e) => panic!("join failed: {e}"),
        _ => None,
    })
    .await;

    // ── Half 1: the host dies mid-share ─────────────────────────────────────
    assert!(controller.send(CoreCommand::StartScreenShare {
        audio_app: None,
        settings: Default::default(),
        quality: Default::default(),
    }));
    wait_for(
        &mut ui_rx,
        "ScreenShareStarted (dying host)",
        |ev| match ev {
            UiEvent::ScreenShareStarted => Some(()),
            UiEvent::Error(e) => panic!("share start failed: {e}"),
            _ => None,
        },
    )
    .await;

    // The fake host exits ~1s in. The contract: ScreenShareStopped FIRST (it
    // clears the UI's sharing state), the explanatory error only after.
    wait_for(&mut ui_rx, "ScreenShareStopped after host death", |ev| {
        match ev {
            UiEvent::ScreenShareStopped => Some(()),
            // An error arriving first is the exact ordering defect S2 fixes:
            // the UI would show "sharing" next to the explanation.
            UiEvent::Error(e) => panic!("error arrived before ScreenShareStopped: {e}"),
            _ => None,
        }
    })
    .await;
    let err = wait_for(&mut ui_rx, "the host-death error", |ev| match ev {
        UiEvent::Error(e) => Some(e.clone()),
        _ => None,
    })
    .await;
    assert!(
        err.contains("unexpectedly"),
        "the error should say the share ended unexpectedly, got: {err}"
    );

    // ── Half 2: a deliberate stop must stay clean ───────────────────────────
    assert!(controller.send(CoreCommand::SetPixelpassPath(Some(
        living_host.to_string_lossy().into_owned()
    ))));
    assert!(controller.send(CoreCommand::StartScreenShare {
        audio_app: None,
        settings: Default::default(),
        quality: Default::default(),
    }));
    wait_for(
        &mut ui_rx,
        "ScreenShareStarted (living host)",
        |ev| match ev {
            UiEvent::ScreenShareStarted => Some(()),
            UiEvent::Error(e) => panic!("second share start failed: {e}"),
            _ => None,
        },
    )
    .await;

    assert!(controller.send(CoreCommand::StopScreenShare));
    wait_for(
        &mut ui_rx,
        "ScreenShareStopped after Stop Share",
        |ev| match ev {
            UiEvent::ScreenShareStopped => Some(()),
            UiEvent::Error(e) => panic!("clean stop produced an error: {e}"),
            _ => None,
        },
    )
    .await;

    // The stopped host's stdout EOF is arriving about now as a *stale* fault
    // (its generation was retired when Stop Share cleared the share). Without
    // the staleness gate the handler would emit a second ScreenShareStopped
    // and a spurious "ended unexpectedly" error — listen long enough for that
    // mishandling to have shown up, and require silence.
    let deadline = tokio::time::Instant::now() + QUIET_WINDOW;
    while let Ok(Some(ev)) = tokio::time::timeout_at(deadline, ui_rx.recv()).await {
        match ev {
            UiEvent::ScreenShareStopped => {
                panic!("stale host fault re-emitted ScreenShareStopped after a clean stop")
            }
            UiEvent::Error(e) if e.contains("unexpectedly") => {
                panic!("stale host fault surfaced as an error after a clean stop: {e}")
            }
            _ => {}
        }
    }

    // ── Half 3: a failed room switch while sharing must not cry "crash" ─────
    // Join tears the old session down (killing the host, deliberately) BEFORE
    // it validates the ticket, so an invalid ticket exits the Join arm early.
    // The share must be retired at the teardown itself — left advertised, the
    // killed host's EOF passes the staleness gate and a spurious "ended
    // unexpectedly" lands on top of the ticket error (Gemini review, P2-1).
    assert!(controller.send(CoreCommand::StartScreenShare {
        audio_app: None,
        settings: Default::default(),
        quality: Default::default(),
    }));
    wait_for(
        &mut ui_rx,
        "ScreenShareStarted (before failed switch)",
        |ev| match ev {
            UiEvent::ScreenShareStarted => Some(()),
            UiEvent::Error(e) => panic!("third share start failed: {e}"),
            _ => None,
        },
    )
    .await;
    assert!(controller.send(CoreCommand::Join {
        name: "host-fault-gate".into(),
        ticket: "definitely-not-a-ticket".into(),
        room_name: "s2".into(),
        input_device: None,
        output_device: None,
        echo_cancellation: false,
        avatar: Default::default(),
    }));
    wait_for(&mut ui_rx, "the invalid-ticket error", |ev| match ev {
        UiEvent::Error(e) if e.contains("invalid room ticket") => Some(()),
        UiEvent::Error(e) => panic!("unexpected error before the ticket error: {e}"),
        _ => None,
    })
    .await;
    // The deliberately-killed host's EOF is arriving about now; it must be
    // dropped as stale, not reported as a crash.
    let deadline = tokio::time::Instant::now() + QUIET_WINDOW;
    while let Ok(Some(ev)) = tokio::time::timeout_at(deadline, ui_rx.recv()).await {
        match ev {
            UiEvent::ScreenShareStopped => {
                panic!("failed room switch re-emitted ScreenShareStopped for the torn-down share")
            }
            UiEvent::Error(e) if e.contains("unexpectedly") => {
                panic!("deliberate teardown during a failed room switch reported as a crash: {e}")
            }
            _ => {}
        }
    }
}

/// S2 presence gate: a host fault must pull the share ticket off PRESENCE —
/// what remote peers actually see — and must do it BEFORE the reap wait, not
/// after. Nothing on the sharer's own `UiEvent` channel can witness either
/// half (presence is only observable from another node), so this test runs a
/// real second core as an OBSERVER and asserts the sharer's `PeerState.sharing`
/// goes `Some` → `None` on fault.
///
/// The observer runs in a SEPARATE PROCESS (`presence_probe_helper`, this same
/// test binary re-invoked): two in-process cores would load the same
/// `identity.key` and collapse into one node id, and swapping `XDG_CONFIG_HOME`
/// between spawns in-process races other threads' getenv.
///
/// The fake host is a WEDGE — it closes stdout (the fault) but ignores SIGINT
/// and lives until the SIGKILL fallback — so `stop_host` burns the full 2 s
/// grace and TIME becomes the discriminator, exactly like the SIGINT gate:
/// with presence-removal-first the observer sees the ticket clear ~1 s after
/// it appeared (the wedge's pre-fault lifetime); with the old
/// reap-then-presence ordering, only after ~3 s. The bound also makes the
/// "presence removal deleted" mutant fail by timeout instead of passing
/// vacuously.
///
/// Live: two real solo-room cores (audio backend + network bind each).
#[tokio::test]
#[ignore = "live: two real cores in one room (audio backend + network bind), observer subprocess"]
async fn a_host_fault_pulls_the_ticket_off_presence_within_the_grace() {
    /// Mirrors `core::teardown::STOP_GRACE` (private): the wait the wedge
    /// forces before the SIGKILL fallback reaps it.
    const STOP_GRACE_MS: u128 = 2000;

    let dir_guard = TempDir(
        std::env::temp_dir().join(format!("peerspeak-presence-gate-{}", std::process::id())),
    );
    let dir = dir_guard.0.clone();
    std::fs::create_dir_all(&dir).unwrap();

    // Emits its ticket, shares for ~1 s, then closes stdout (the fault) while
    // staying alive and ignoring SIGINT, so the reap must wait out the grace.
    // The trailing sleep is NOT exec'd on purpose: it forks after stdout is
    // closed, so it holds no pipe (the vacuous-staleness trap doesn't apply),
    // and it merely idles out after the SIGKILL reaps the shell.
    //
    // The fake ticket must pass `screenshare::sanitize_ticket` (`endpoint` +
    // alphanumerics): the OBSERVER's gossip ingest sanitizes peer-advertised
    // tickets, and a garbage one is nulled to `sharing: None` there — the
    // probe would never see the share appear and the gate would go vacuous.
    let wedged_host = write_fake_pixelpass(
        &dir,
        "pixelpass-wedges",
        "#!/bin/sh\ntrap '' INT\n\
         echo '{\"event\":\"ticket\",\"value\":\"endpointaabwxjexzensznfvuudiapn5tyzws3angd2merarm\"}'\n\
         sleep 1\nexec 1>&-\nsleep 30\n",
    );

    let (ui_tx, mut ui_rx) = tokio::sync::mpsc::channel(256);
    let controller = CoreController::new(ui_tx);

    assert!(controller.send(CoreCommand::SetPixelpassPath(Some(
        wedged_host.to_string_lossy().into_owned()
    ))));
    assert!(controller.send(CoreCommand::Join {
        name: "presence-gate".into(),
        ticket: "create".into(),
        room_name: "s2-presence".into(),
        input_device: None,
        output_device: None,
        echo_cancellation: false,
        avatar: Default::default(),
    }));
    let room_ticket = wait_for(&mut ui_rx, "RoomJoined", |ev| match ev {
        UiEvent::RoomJoined { ticket, .. } => Some(ticket.clone()),
        UiEvent::Error(e) => panic!("join failed: {e}"),
        _ => None,
    })
    .await;

    // The observer, in its own process with its own config dir (fresh
    // identity). It prints `PROBE …` lines this test parses.
    let probe_config = dir.join("probe-config");
    std::fs::create_dir_all(&probe_config).unwrap();
    let probe = tokio::process::Command::new(std::env::current_exe().unwrap())
        .kill_on_drop(true)
        .args([
            "presence_probe_helper",
            "--exact",
            "--ignored",
            "--nocapture",
        ])
        .env("PEERSPEAK_PROBE_TICKET", &room_ticket)
        .env("XDG_CONFIG_HOME", &probe_config)
        .stdout(std::process::Stdio::piped())
        .stderr(std::process::Stdio::piped())
        .spawn()
        .expect("spawn the presence probe");

    // Only share once the probe is in the room, so it witnesses the ticket
    // APPEARING before the fault clears it (otherwise `Some` → `None` could
    // both predate its join and the gate would go vacuous).
    wait_for(&mut ui_rx, "the probe's PeerJoined", |ev| match ev {
        UiEvent::PeerJoined { .. } => Some(()),
        UiEvent::Error(e) => panic!("waiting for the probe: {e}"),
        _ => None,
    })
    .await;

    assert!(controller.send(CoreCommand::StartScreenShare {
        audio_app: None,
        settings: Default::default(),
        quality: Default::default(),
    }));
    wait_for(
        &mut ui_rx,
        "ScreenShareStarted (wedged host)",
        |ev| match ev {
            UiEvent::ScreenShareStarted => Some(()),
            UiEvent::Error(e) => panic!("share start failed: {e}"),
            _ => None,
        },
    )
    .await;

    // Sharer-side contract, unchanged by the reorder: Stopped first, the
    // explanatory error only after.
    wait_for(
        &mut ui_rx,
        "ScreenShareStopped after the wedge faults",
        |ev| match ev {
            UiEvent::ScreenShareStopped => Some(()),
            UiEvent::Error(e) => panic!("error arrived before ScreenShareStopped: {e}"),
            _ => None,
        },
    )
    .await;
    let err = wait_for(&mut ui_rx, "the host-death error", |ev| match ev {
        UiEvent::Error(e) => Some(e.clone()),
        _ => None,
    })
    .await;
    assert!(
        err.contains("unexpectedly"),
        "the error should say the share ended unexpectedly, got: {err}"
    );

    let out = tokio::time::timeout(Duration::from_secs(60), probe.wait_with_output())
        .await
        .expect("probe process outlived its budget")
        .expect("probe process wait");
    let stdout = String::from_utf8_lossy(&out.stdout);
    let stderr = String::from_utf8_lossy(&out.stderr);
    assert!(
        out.status.success(),
        "probe failed ({}).\nstdout:\n{stdout}\nstderr:\n{stderr}",
        out.status
    );
    let cleared_ms: u128 = stdout
        .lines()
        .find_map(|l| l.strip_prefix("PROBE sharing-cleared "))
        .unwrap_or_else(|| {
            panic!("probe never saw the ticket clear from presence.\nstdout:\n{stdout}")
        })
        .trim()
        .parse()
        .expect("probe delta should be integer millis");
    // Presence-removal-first: ~1000 ms (the wedge's pre-fault lifetime).
    // Reap-then-presence: ~3000 ms (lifetime + the full stop grace). The
    // grace itself splits them with ~1 s of jitter headroom on each side.
    assert!(
        cleared_ms < STOP_GRACE_MS,
        "presence kept advertising the dead share for {cleared_ms} ms after it appeared — \
         at or past the wedge lifetime + stop grace, i.e. the ticket was only removed \
         AFTER the reap wait instead of before it"
    );

    assert!(controller.send(CoreCommand::Leave));
}

/// Observer half of `a_host_fault_pulls_the_ticket_off_presence_within_the_grace`,
/// run BY that test as a subprocess. Standalone (no `PEERSPEAK_PROBE_TICKET` in
/// the env — e.g. a plain `--ignored` sweep) it is a no-op pass.
#[tokio::test]
#[ignore = "helper: spawned by the presence gate as a subprocess; standalone it no-ops"]
async fn presence_probe_helper() {
    let Ok(room_ticket) = std::env::var("PEERSPEAK_PROBE_TICKET") else {
        return;
    };

    let (ui_tx, mut ui_rx) = tokio::sync::mpsc::channel(256);
    let controller = CoreController::new(ui_tx);
    assert!(controller.send(CoreCommand::Join {
        name: "presence-probe".into(),
        ticket: room_ticket,
        room_name: String::new(),
        input_device: None,
        output_device: None,
        echo_cancellation: false,
        avatar: Default::default(),
    }));
    wait_for(&mut ui_rx, "RoomJoined (probe)", |ev| match ev {
        UiEvent::RoomJoined { .. } => Some(()),
        UiEvent::Error(e) => panic!("probe join failed: {e}"),
        _ => None,
    })
    .await;

    // Watch the sharer's presence: record when its `sharing` ticket appears,
    // report the delta when it clears. Timings on both ends are local-loopback
    // arrival times, so the parent's bound compares like with like.
    let deadline = tokio::time::Instant::now() + Duration::from_secs(30);
    let mut seen_at: Option<std::time::Instant> = None;
    loop {
        let ev = tokio::time::timeout_at(deadline, ui_rx.recv())
            .await
            .expect("probe timed out watching for the sharing transition")
            .expect("probe ui channel closed");
        let sharing = match &ev {
            UiEvent::PeerJoined { state, .. } | UiEvent::PeerUpdated { state, .. } => {
                state.sharing.is_some()
            }
            _ => continue,
        };
        match (&seen_at, sharing) {
            (None, true) => {
                seen_at = Some(std::time::Instant::now());
                println!("PROBE sharing-seen");
            }
            (Some(t0), false) => {
                println!("PROBE sharing-cleared {}", t0.elapsed().as_millis());
                break;
            }
            _ => {}
        }
    }
    assert!(controller.send(CoreCommand::Leave));
}

/// The long-owed Stop Share SIGINT gate (0c half (ii)), against the REAL
/// pixelpass binary: a Stop Share must end the host through the graceful
/// SIGINT path — child exits within [`STOP_GRACE`], no SIGKILL fallback, no
/// "couldn't confirm" warning — because SIGKILL would skip pixelpass's own
/// teardown (it unloads its capture sink on the way out in sink-owning modes).
///
/// The fallback is indistinguishable from success in the event stream (both
/// end in a confirmed reap), so the discriminator is TIME: the fallback path
/// first waits out the full 2 s grace, while a host honouring SIGINT exits in
/// milliseconds. The bound asserts the stop completed inside the grace.
///
/// Live: needs `pixelpass` on `$PATH` plus a real solo room (audio + network).
#[tokio::test]
#[ignore = "live: real pixelpass host + a real solo room (audio backend, network bind)"]
async fn stop_share_ends_the_real_host_via_sigint_within_the_grace() {
    /// Mirrors `core::teardown::STOP_GRACE` (private): the graceful wait
    /// before the SIGKILL fallback.
    const STOP_GRACE: Duration = Duration::from_secs(2);

    let (ui_tx, mut ui_rx) = tokio::sync::mpsc::channel(256);
    let controller = CoreController::new(ui_tx);

    // No override: resolve the real binary from $PATH.
    assert!(controller.send(CoreCommand::SetPixelpassPath(None)));
    assert!(controller.send(CoreCommand::Join {
        name: "sigint-gate".into(),
        ticket: "create".into(),
        room_name: "s2".into(),
        input_device: None,
        output_device: None,
        echo_cancellation: false,
        avatar: Default::default(),
    }));
    wait_for(&mut ui_rx, "RoomJoined", |ev| match ev {
        UiEvent::RoomJoined { .. } => Some(()),
        UiEvent::Error(e) => panic!("join failed: {e}"),
        _ => None,
    })
    .await;

    // Whole-desktop share: no viewers ever connect, so the real host sits idle
    // after its ticket (capture starts on first viewer) — exactly the state a
    // Stop Share most often hits.
    assert!(controller.send(CoreCommand::StartScreenShare {
        audio_app: None,
        settings: Default::default(),
        quality: Default::default(),
    }));
    wait_for(
        &mut ui_rx,
        "ScreenShareStarted (real pixelpass)",
        |ev| match ev {
            UiEvent::ScreenShareStarted => Some(()),
            UiEvent::Error(e) => panic!("real pixelpass host failed to start: {e}"),
            _ => None,
        },
    )
    .await;

    let stop_started = std::time::Instant::now();
    assert!(controller.send(CoreCommand::StopScreenShare));
    wait_for(
        &mut ui_rx,
        "ScreenShareStopped (real pixelpass)",
        |ev| match ev {
            UiEvent::ScreenShareStopped => Some(()),
            // An Unconfirmed reap surfaces exactly this way; it means the
            // SIGINT AND the SIGKILL both failed to end the host.
            UiEvent::Error(e) => panic!("stop of the real host was not clean: {e}"),
            _ => None,
        },
    )
    .await;
    let elapsed = stop_started.elapsed();
    assert!(
        elapsed < STOP_GRACE,
        "stop took {elapsed:?} — at or past the {STOP_GRACE:?} grace, i.e. the \
         SIGKILL fallback fired instead of pixelpass honouring SIGINT"
    );

    // And the late stdout EOF from the SIGINTed host must stay silent (same
    // staleness contract the fake-host half pins).
    let deadline = tokio::time::Instant::now() + QUIET_WINDOW;
    while let Ok(Some(ev)) = tokio::time::timeout_at(deadline, ui_rx.recv()).await {
        match ev {
            UiEvent::ScreenShareStopped => {
                panic!("stale fault from the SIGINTed real host re-emitted ScreenShareStopped")
            }
            UiEvent::Error(e) if e.contains("unexpectedly") => {
                panic!("stale fault from the SIGINTed real host surfaced as an error: {e}")
            }
            _ => {}
        }
    }

    assert!(controller.send(CoreCommand::Leave));
}