diff --git a/clients/android/app/src/main/kotlin/io/unom/punktfunk/StreamScreen.kt b/clients/android/app/src/main/kotlin/io/unom/punktfunk/StreamScreen.kt index 72d93977..eaec7000 100644 --- a/clients/android/app/src/main/kotlin/io/unom/punktfunk/StreamScreen.kt +++ b/clients/android/app/src/main/kotlin/io/unom/punktfunk/StreamScreen.kt @@ -50,6 +50,9 @@ import androidx.core.content.ContextCompat import androidx.core.view.WindowCompat import androidx.core.view.WindowInsetsCompat import androidx.core.view.WindowInsetsControllerCompat +import androidx.lifecycle.Lifecycle +import androidx.lifecycle.LifecycleEventObserver +import androidx.lifecycle.LifecycleOwner import io.unom.punktfunk.kit.GamepadFeedback import io.unom.punktfunk.kit.GamepadRouter import io.unom.punktfunk.kit.deviceBodyVibrator @@ -384,6 +387,26 @@ fun StreamScreen(handle: Long, micEnabled: Boolean, onDisconnect: () -> Unit) { // Back gesture = a deliberate exit → signal the quit so the host tears down now (no linger). BackHandler { NativeBridge.nativeDisconnectQuit(handle); onDisconnect() } + // Leaving the app (Home, task switch, screen off) MUST end the session. Android does not + // suspend a process for going to background, so without this the native worker kept running and + // its QUIC connection kept answering the host's keep-alives — the user was long gone but the + // host still saw a live client and held the session (and its display + encoder) open until the + // OS eventually reclaimed the process, which on a TV box is effectively never. + // + // Route it through `onDisconnect()` so the composable's `onDispose` above runs the one real + // teardown path. Deliberately NOT a `nativeDisconnectQuit`: backgrounding isn't a user "quit", + // so the host should linger the display and make coming straight back a fast reconnect. + DisposableEffect(handle) { + val lifecycle = (context as? LifecycleOwner)?.lifecycle + val obs = LifecycleEventObserver { _, event -> + if (event == Lifecycle.Event.ON_STOP) { + onDisconnect() + } + } + lifecycle?.addObserver(obs) + onDispose { lifecycle?.removeObserver(obs) } + } + // Auto-engage pointer capture at stream start (setting on + a mouse actually present). // Delayed a beat: the grab needs window focus and the capture view attached. LaunchedEffect(handle) { diff --git a/clients/apple/Sources/PunktfunkClient/ContentView.swift b/clients/apple/Sources/PunktfunkClient/ContentView.swift index 043a43b1..9937cb73 100644 --- a/clients/apple/Sources/PunktfunkClient/ContentView.swift +++ b/clients/apple/Sources/PunktfunkClient/ContentView.swift @@ -144,14 +144,26 @@ struct ContentView: View { // tap uses, so trust policy / WoL / the approval sheet all come along. Never starts a // parallel session — this drives the one `model` ContentView owns. .onOpenURL { handleDeepLink($0) } - #if os(iOS) - // Background keep-alive driver (opt-in). Only .background/.active matter; .inactive (a - // transient peek) is ignored so the disconnect timer never starts for a Control-Center pull. + #if os(iOS) || os(tvOS) + // Backgrounding driver. Only .background/.active matter; .inactive (a transient peek) is + // ignored so neither branch fires for a Control-Center pull. + // + // Backgrounding MUST end the session one way or the other: the app keeps running while + // streaming (the `audio` background mode plus a live audio session), so its QUIC connection + // keeps answering the host's keep-alives with the user long gone — the host has no way to + // tell that apart from someone watching, and the session survived indefinitely. Either hold + // it under the opt-in keep-alive (bounded by that path's own auto-disconnect timer) or end + // it here. .onChange(of: scenePhase) { _, phase in switch phase { case .background: - if backgroundKeepAlive, model.phase == .streaming { + guard model.phase == .streaming else { break } + if backgroundKeepAlive { model.enterBackground(timeoutMinutes: backgroundTimeoutMinutes) + } else { + // Not deliberate: the user may come straight back, so let the host linger the + // display for a fast reconnect instead of tearing it down. + model.disconnect(deliberate: false) } case .active: model.exitBackground() @@ -159,7 +171,11 @@ struct ContentView: View { break } } - // Live Activity lifecycle, driven from the model's published state. + #endif + #if os(iOS) + // Live Activity lifecycle, driven from the model's published state. iPhone/iPad only — + // ActivityKit (and so `liveActivity`) does not exist on tvOS, which is why this stays in its + // own os(iOS) block rather than riding the backgrounding driver's. .onChange(of: model.phase) { _, phase in switch phase { case .streaming: diff --git a/crates/pf-vdisplay/src/lib.rs b/crates/pf-vdisplay/src/lib.rs index e7c09ca7..f8b58385 100644 --- a/crates/pf-vdisplay/src/lib.rs +++ b/crates/pf-vdisplay/src/lib.rs @@ -58,6 +58,11 @@ pub(crate) fn emit_display_event(ev: DisplayEvent) { pub(crate) mod backend; pub use backend::{DisplayOwnership, VirtualDisplay, VirtualOutput}; +/// Time-bounded child-process helpers — every compositor query shells out, and an unbounded one +/// can wedge the calling (session) thread forever. +#[path = "vdisplay/proc.rs"] +pub(crate) mod proc; + /// Live-session detection + session-epoch + env retargeting (plan §W3). #[path = "vdisplay/session.rs"] pub(crate) mod session; diff --git a/crates/pf-vdisplay/src/vdisplay/linux/kwin.rs b/crates/pf-vdisplay/src/vdisplay/linux/kwin.rs index f86b62df..809cb1aa 100644 --- a/crates/pf-vdisplay/src/vdisplay/linux/kwin.rs +++ b/crates/pf-vdisplay/src/vdisplay/linux/kwin.rs @@ -149,11 +149,7 @@ impl VirtualDisplay for KwinDisplay { return; }; // kscreen-doctor position syntax: `output..position.,`. - let ok = std::process::Command::new("kscreen-doctor") - .arg(format!("output.{output}.position.{x},{y}")) - .status() - .map(|s| s.success()) - .unwrap_or(false); + let ok = kscreen_ok(&[format!("output.{output}.position.{x},{y}")]); if ok { tracing::info!(output, x, y, "KWin: placed output in the desktop layout"); } else { @@ -342,9 +338,7 @@ fn reenable_outputs(outputs: &[(String, String)]) { .iter() .map(|(name, _)| format!("output.{name}.enable")) .collect(); - let _ = std::process::Command::new("kscreen-doctor") - .args(&enable_args) - .status(); + let _ = kscreen_ok(&enable_args); // THEN re-assert each captured mode, best-effort — a bare re-enable lets KWin fall back to the // EDID-preferred mode (a 120 Hz panel returns at ~60 Hz); this restores the exact refresh. The // output is enabled now, so the mode set is valid; a rejected mode just leaves KWin's default. @@ -354,9 +348,7 @@ fn reenable_outputs(outputs: &[(String, String)]) { .map(|(name, mode)| format!("output.{name}.mode.{mode}")) .collect(); if !mode_args.is_empty() { - let _ = std::process::Command::new("kscreen-doctor") - .args(&mode_args) - .status(); + let _ = kscreen_ok(&mode_args); } std::thread::sleep(Duration::from_millis(200)); tracing::info!(reenabled = ?outputs, "KWin: restored the physical/bootstrap outputs at their captured modes (group empty)"); @@ -404,13 +396,37 @@ fn resolve_kscreen_addr(name: &str, w: u32, h: u32) -> String { fallback } +/// Budget for one `kscreen-doctor` call. +/// +/// It is a Wayland client of the very compositor it configures, so against a wedged KWin it blocks +/// in its own connect and never returns — and these calls run on the session's stream thread, whose +/// only way to end a session is to return. Generous next to a healthy call (tens of ms). +const KSCREEN_BUDGET: Duration = Duration::from_secs(5); + +/// `kscreen-doctor ` run for its exit status, bounded by [`KSCREEN_BUDGET`]. A timeout reads +/// as a failed apply — the same best-effort path a rejected argument already takes. +fn kscreen_ok(args: &[String]) -> bool { + crate::proc::status_within( + std::process::Command::new("kscreen-doctor").args(args), + KSCREEN_BUDGET, + ) + .map(|s| s.success()) + .unwrap_or(false) +} + +/// `kscreen-doctor -j` stdout, bounded by [`KSCREEN_BUDGET`]; `None` on any failure. +fn kscreen_json_bytes() -> Option> { + crate::proc::output_within( + std::process::Command::new("kscreen-doctor").arg("-j"), + KSCREEN_BUDGET, + ) + .ok() + .map(|o| o.stdout) +} + /// `kscreen-doctor -j` parsed, `None` on any failure. fn kscreen_json() -> Option { - let out = std::process::Command::new("kscreen-doctor") - .arg("-j") - .output() - .ok()?; - serde_json::from_slice(&out.stdout).ok() + serde_json::from_slice(&kscreen_json_bytes()?).ok() } /// The `(width, height)` of an output's CURRENT mode from its `kscreen-doctor -j` entry. @@ -545,13 +561,7 @@ fn pick_custom_mode<'a>( fn set_custom_refresh(width: u32, height: u32, hz: u32, output: &str) -> Option<(u32, u32, u32)> { let output = output.to_string(); let mhz = hz.saturating_mul(1000); - let run = |arg: String| { - std::process::Command::new("kscreen-doctor") - .arg(arg) - .status() - .map(|s| s.success()) - .unwrap_or(false) - }; + let run = |arg: String| kscreen_ok(&[arg]); // Install the mode only if the output doesn't already carry a usable one: kscreen-doctor // APPENDS to the output's custom-mode list and KWin PERSISTS that list per output name // (`kwinoutputconfig.json`, which is why the same per-slot name is reused across sessions) — so @@ -710,14 +720,11 @@ fn output_current_mode_spec(o: &serde_json::Value) -> Option { /// session's own output — so a 2nd `exclusive` session (with a distinct per-slot name) never disables /// the 1st session's live output. Parsed from `kscreen-doctor -j` (same source as [`read_active_mode`]). fn other_enabled_outputs() -> Vec<(String, String)> { - let out = match std::process::Command::new("kscreen-doctor") - .arg("-j") - .output() - { - Ok(o) => o, - Err(_) => return Vec::new(), + let out = match kscreen_json_bytes() { + Some(o) => o, + None => return Vec::new(), }; - let doc: serde_json::Value = match serde_json::from_slice(&out.stdout) { + let doc: serde_json::Value = match serde_json::from_slice(&out) { Ok(d) => d, Err(_) => return Vec::new(), }; @@ -746,13 +753,10 @@ fn other_enabled_outputs() -> Vec<(String, String)> { /// then sets itself primary — the pre-group behavior). Recent kscreen marks the primary with /// `"priority": 1`; older builds used a `"primary": true` bool — accept either. fn a_managed_output_is_primary() -> bool { - let Ok(out) = std::process::Command::new("kscreen-doctor") - .arg("-j") - .output() - else { + let Some(out) = kscreen_json_bytes() else { return false; }; - let Ok(doc) = serde_json::from_slice::(&out.stdout) else { + let Ok(doc) = serde_json::from_slice::(&out) else { return false; }; doc.get("outputs") @@ -777,13 +781,7 @@ fn a_managed_output_is_primary() -> bool { /// the keepalive to re-enable on teardown. Best-effort: on failure, streaming continues (just possibly /// showing only the wallpaper) rather than failing the session. fn apply_virtual_primary(ours: &str) -> Vec<(String, String)> { - let kscreen = |args: &[String]| { - std::process::Command::new("kscreen-doctor") - .args(args) - .status() - .map(|s| s.success()) - .unwrap_or(false) - }; + let kscreen = |args: &[String]| kscreen_ok(args); // First-slot-wins (§6.1): only grab primary if no managed group member is primary yet — so a 2nd // exclusive session joins as a secondary monitor of the shared desktop instead of stealing the // shell off the 1st session's output. KWin usually then re-homes the desktop + disables the @@ -815,11 +813,7 @@ fn apply_virtual_primary(ours: &str) -> Vec<(String, String)> { /// (don't disable the bootstrap/physical) — so the shell re-homes onto the streamed surface while a /// physical screen stays usable. Nothing to restore on teardown (we disabled nothing). fn apply_virtual_primary_only(ours: &str) { - let ok = std::process::Command::new("kscreen-doctor") - .arg(format!("output.{ours}.primary")) - .status() - .map(|s| s.success()) - .unwrap_or(false); + let ok = kscreen_ok(&[format!("output.{ours}.primary")]); if ok { tracing::info!("KWin: streamed output set primary (physical outputs kept)"); } else { diff --git a/crates/pf-vdisplay/src/vdisplay/proc.rs b/crates/pf-vdisplay/src/vdisplay/proc.rs new file mode 100644 index 00000000..7a8fdf3e --- /dev/null +++ b/crates/pf-vdisplay/src/vdisplay/proc.rs @@ -0,0 +1,114 @@ +//! Time-bounded child-process helpers. +//! +//! Every compositor query this crate makes shells out to a helper (`kscreen-doctor`, `systemctl`, +//! `pw-dump`, …), and most of them are *clients of the very thing being diagnosed*: `kscreen-doctor` +//! is a Wayland client, so against a wedged KWin it blocks in its own connect and **never returns**. +//! `Command::status()` / `Command::output()` have no timeout, so one hung helper pinned the calling +//! thread forever — and on the host that thread is the session's stream thread, whose only way to +//! end a session is to return. A stuck query therefore became a permanently stuck session. +//! +//! These wrappers bound the wait: poll for exit until the budget runs out, then kill the child and +//! report [`std::io::ErrorKind::TimedOut`], so callers see a plain "the helper failed" error and +//! take their existing failure path instead of hanging. + +use std::io::{Error, ErrorKind, Result}; +use std::process::{Command, ExitStatus, Output}; +use std::time::{Duration, Instant}; + +/// Poll interval while waiting for a child to exit. Short enough that a fast helper (the normal +/// case — `kscreen-doctor` answers in tens of ms) isn't measurably delayed. +const POLL: Duration = Duration::from_millis(20); + +/// Run `cmd` to completion, killing it if it outlives `budget`. +/// +/// Stdout/stderr are left as the caller configured them (inherited by default), so this is for +/// commands run for their exit status alone — see [`output_within`] when the output is read. +pub(crate) fn status_within(cmd: &mut Command, budget: Duration) -> Result { + let mut child = cmd.spawn()?; + let deadline = Instant::now() + budget; + loop { + match child.try_wait()? { + Some(status) => return Ok(status), + None if Instant::now() >= deadline => { + let _ = child.kill(); + let _ = child.wait(); // reap it — never leave a zombie behind + return Err(timed_out(cmd, budget)); + } + None => std::thread::sleep(POLL), + } + } +} + +/// Run `cmd` to completion and capture its stdout/stderr, killing it if it outlives `budget`. +/// +/// The output is read only after the child has exited, so a helper that fills the pipe buffer and +/// stalls is caught by the budget rather than deadlocking the reader (these helpers emit at most a +/// few hundred KiB, well under any real pipe pressure). +pub(crate) fn output_within(cmd: &mut Command, budget: Duration) -> Result { + let mut child = cmd + .stdout(std::process::Stdio::piped()) + .stderr(std::process::Stdio::piped()) + .spawn()?; + let deadline = Instant::now() + budget; + loop { + match child.try_wait()? { + // Exited: `wait_with_output` now only drains already-buffered pipes. + Some(_) => return child.wait_with_output(), + None if Instant::now() >= deadline => { + let _ = child.kill(); + let _ = child.wait(); + return Err(timed_out(cmd, budget)); + } + None => std::thread::sleep(POLL), + } + } +} + +fn timed_out(cmd: &Command, budget: Duration) -> Error { + let program = cmd.get_program().to_string_lossy().to_string(); + tracing::warn!( + program, + budget_ms = budget.as_millis() as u64, + "helper did not exit within its budget — killed it (a wedged compositor/session bus is the \ + usual cause); treating it as a failed query" + ); + Error::new( + ErrorKind::TimedOut, + format!("`{program}` did not exit within {budget:?}"), + ) +} + +#[cfg(test)] +mod tests { + use super::*; + + /// A helper that never exits must be killed at the budget and reported as `TimedOut` — the + /// whole point of the module (an unbounded `status()` here is what wedged a whole session). + #[test] + fn a_hung_child_is_killed_at_the_budget() { + let started = Instant::now(); + let err = status_within(Command::new("sleep").arg("30"), Duration::from_millis(150)) + .expect_err("must time out"); + assert_eq!(err.kind(), ErrorKind::TimedOut); + assert!( + started.elapsed() < Duration::from_secs(5), + "must return at its budget, not the child's lifetime (took {:?})", + started.elapsed() + ); + } + + /// The normal path is unaffected: a quick command still yields its status and its output. + #[test] + fn a_quick_child_returns_normally() { + let st = status_within(&mut Command::new("true"), Duration::from_secs(5)).expect("ran"); + assert!(st.success()); + + let out = output_within( + Command::new("echo").arg("punktfunk"), + Duration::from_secs(5), + ) + .expect("ran"); + assert!(out.status.success()); + assert_eq!(String::from_utf8_lossy(&out.stdout).trim(), "punktfunk"); + } +} diff --git a/crates/pf-vdisplay/src/vdisplay/session.rs b/crates/pf-vdisplay/src/vdisplay/session.rs index 3a27b5b4..498813ad 100644 --- a/crates/pf-vdisplay/src/vdisplay/session.rs +++ b/crates/pf-vdisplay/src/vdisplay/session.rs @@ -6,6 +6,15 @@ use super::*; +/// Budget for one `systemctl --user` / `dbus-update-activation-environment` call. +/// +/// These talk to the session bus, and a bus that is itself restarting or wedged answers nothing — +/// unbounded, that pinned the caller (on the host, the session's stream thread) forever. A restart +/// of the portal units is the slowest legitimate case, hence the generous window; missing it just +/// means the portal env settles late, which the callers already treat as best-effort. +#[cfg(target_os = "linux")] +const SYSTEMD_BUDGET: std::time::Duration = std::time::Duration::from_secs(10); + /// The **session epoch** — bumped whenever session detection observes a different compositor /// *instance*: an [`ActiveKind`] change, **or** a new compositor PID for the same kind (the /// Desktop→Game→Desktop bounce that brings up a fresh KWin/gamescope with an unrelated node-id space). @@ -86,9 +95,15 @@ pub fn observe_session_instance(active: &ActiveSession) { /// via the next [`settle_desktop_portal`], so scrubbing on a bounce is harmless.) #[cfg(target_os = "linux")] fn scrub_desktop_manager_env() { - let _ = std::process::Command::new("systemctl") - .args(["--user", "unset-environment", "WAYLAND_DISPLAY", "DISPLAY"]) - .status(); + let _ = crate::proc::status_within( + std::process::Command::new("systemctl").args([ + "--user", + "unset-environment", + "WAYLAND_DISPLAY", + "DISPLAY", + ]), + SYSTEMD_BUDGET, + ); } #[cfg(not(target_os = "linux"))] @@ -499,40 +514,46 @@ pub fn settle_desktop_portal(chosen: Compositor) { ]; // Push our (correct) env into the systemd --user manager + the D-Bus activation environment so a // re-activated portal/backend inherits the live session. - let _ = std::process::Command::new("systemctl") - .args(["--user", "import-environment"]) - .args(VARS) - .status(); - let _ = std::process::Command::new("dbus-update-activation-environment") - .arg("--systemd") - .args(VARS) - .status(); + let _ = crate::proc::status_within( + std::process::Command::new("systemctl") + .args(["--user", "import-environment"]) + .args(VARS), + SYSTEMD_BUDGET, + ); + let _ = crate::proc::status_within( + std::process::Command::new("dbus-update-activation-environment") + .arg("--systemd") + .args(VARS), + SYSTEMD_BUDGET, + ); // KWin input goes through the xdg RemoteDesktop portal; the frontend routes RemoteDesktop to a // backend by its OWN startup XDG_CURRENT_DESKTOP, so restart it (+ the KDE backend) to re-read // the now-live session, then let it settle before the injector reopens against it. if chosen == Compositor::Kwin { - let _ = std::process::Command::new("systemctl") - .args([ + let _ = crate::proc::status_within( + std::process::Command::new("systemctl").args([ "--user", "try-restart", "xdg-desktop-portal-kde.service", "xdg-desktop-portal.service", - ]) - .status(); + ]), + SYSTEMD_BUDGET, + ); std::thread::sleep(std::time::Duration::from_millis(600)); } // Hyprland capture rides the xdg ScreenCast portal serviced by xdph (G5): on a mid-stream switch // xdph may still hold the old session's Wayland/instance env, so restart it (+ the frontend) to // re-read the now-live session, mirroring the KWin settling above. if chosen == Compositor::Hyprland { - let _ = std::process::Command::new("systemctl") - .args([ + let _ = crate::proc::status_within( + std::process::Command::new("systemctl").args([ "--user", "try-restart", "xdg-desktop-portal-hyprland.service", "xdg-desktop-portal.service", - ]) - .status(); + ]), + SYSTEMD_BUDGET, + ); std::thread::sleep(std::time::Duration::from_millis(600)); } tracing::info!( diff --git a/crates/punktfunk-core/src/client/pump.rs b/crates/punktfunk-core/src/client/pump.rs index 9d14e945..74fa2055 100644 --- a/crates/punktfunk-core/src/client/pump.rs +++ b/crates/punktfunk-core/src/client/pump.rs @@ -36,6 +36,7 @@ pub(super) async fn run_pump(args: WorkerArgs) { }; let handshake::HandshakeOut { conn, + ep, session, ctrl_send, ctrl_recv, @@ -192,4 +193,11 @@ pub(super) async fn run_pump(args: WorkerArgs) { 0 }; conn.close(close_code.into(), b"client closed"); + // Flush the CONNECTION_CLOSE before the runtime is dropped (the same discipline as the pairing + // + probe paths). `close` only queues the frame — the endpoint driver puts it on the wire, and + // this fn is the body of a `block_on` whose runtime is dropped the instant it returns, so + // without this the driver could simply never be polled again. The host then saw a deliberate + // quit as silence: no `QUIT_CLOSE_CODE`, an 8 s idle timeout, and the keep-alive linger meant + // for an UNWANTED disconnect. Bounded — a host already gone must not delay the client's exit. + let _ = tokio::time::timeout(std::time::Duration::from_millis(300), ep.wait_idle()).await; } diff --git a/crates/punktfunk-core/src/client/pump/handshake.rs b/crates/punktfunk-core/src/client/pump/handshake.rs index 674b457b..4b5a7f94 100644 --- a/crates/punktfunk-core/src/client/pump/handshake.rs +++ b/crates/punktfunk-core/src/client/pump/handshake.rs @@ -9,6 +9,12 @@ use super::*; /// Everything [`run_pump`](super::run_pump) needs from a successful connect + handshake. pub(super) struct HandshakeOut { pub(super) conn: quinn::Connection, + /// The dialing endpoint, kept alive for the session so the pump can FLUSH its + /// `CONNECTION_CLOSE` before the runtime is dropped ([`super::run_pump`]). Dropping it here + /// left nothing to drive the close onto the wire, so a deliberate quit reached the host as + /// silence — an 8 s idle timeout with no quit code, which the host reads as an unwanted + /// disconnect and lingers the display for. + pub(super) ep: quinn::Endpoint, pub(super) session: Session, pub(super) ctrl_send: quinn::SendStream, pub(super) ctrl_recv: io::MsgReader, @@ -238,6 +244,7 @@ pub(super) async fn connect_and_handshake(args: &WorkerArgs) -> Result Ok(HandshakeOut { conn, + ep, session, ctrl_send: send, ctrl_recv: recv, diff --git a/crates/punktfunk-host/src/native.rs b/crates/punktfunk-host/src/native.rs index 6d940f1b..102cf098 100644 --- a/crates/punktfunk-host/src/native.rs +++ b/crates/punktfunk-host/src/native.rs @@ -392,11 +392,6 @@ pub(crate) async fn serve( ); loop { - let permit = sem - .clone() - .acquire_owned() - .await - .expect("session semaphore is never closed"); let incoming = match ep.accept().await { Some(i) => i, None => break, // endpoint closed @@ -409,9 +404,18 @@ pub(crate) async fn serve( Ok(c) => c, Err(e) => { tracing::warn!(error = %e, "QUIC accept failed"); - continue; // `permit` drops here → slot freed; not counted toward max_sessions + continue; // not counted toward max_sessions } }; + // Take the session slot only AFTER the handshake, so a full host still ACCEPTS the + // connection and the waiting client sees a live path (quinn's keep-alive holds it) instead + // of a silent dial timeout — previously the loop parked on this await before `accept()`, so + // a host at its concurrency cap looked simply unreachable. + let permit = sem + .clone() + .acquire_owned() + .await + .expect("session semaphore is never closed"); let peer = conn.remote_address(); tracing::info!(%peer, "punktfunk/1 client connected"); let opts = opts.clone(); @@ -486,6 +490,31 @@ pub(crate) async fn serve( /// connects and never finishes the handshake would otherwise wedge the host for everyone. const HANDSHAKE_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10); +/// How long the stream thread may still run AFTER its session was told to stop before +/// [`serve_session`] gives up waiting for it. +/// +/// Must exceed every legitimate post-stop path so a slow-but-healthy teardown is never abandoned: +/// the capture-loss rebuild budget is 40 s and one pipeline-build attempt can take ~10 s on a cold +/// compositor, so 90 s leaves generous headroom. +const STREAM_STOP_GRACE: std::time::Duration = std::time::Duration::from_secs(90); + +/// How long teardown waits for the audio + input threads once the connection is closed. They exit +/// promptly by construction (the audio loop checks `stop` every ≤5 s; the input thread's channel +/// drops with the connection), so this only catches a genuine wedge. +const SIDE_THREAD_JOIN_GRACE: std::time::Duration = std::time::Duration::from_secs(10); + +/// Resolves once `stop` has been set for [`STREAM_STOP_GRACE`] — i.e. the session was told to end +/// and its stream thread *still* hasn't returned. +/// +/// Polled rather than notified: `stop` is a plain flag shared with blocking threads, and the poll +/// only runs while a session is live (every 500 ms, one relaxed atomic load). +async fn stop_overdue(stop: &AtomicBool) { + while !stop.load(Ordering::SeqCst) { + tokio::time::sleep(std::time::Duration::from_millis(500)).await; + } + tokio::time::sleep(STREAM_STOP_GRACE).await; +} + /// QUIC application error code the host closes with on a `mode_conflict = reject` admission refusal, /// carrying the human-readable busy reason (live mode + client label) the client surfaces. A distinct /// code lets a client tell "host busy" apart from a transport failure. Shared with the clients via @@ -1330,7 +1359,7 @@ async fn serve_session( let bringup_dp = bringup.clone(); let resize_ms_dp = resize_ms.clone(); let result: Result<()> = async { - tokio::task::spawn_blocking(move || -> Result<()> { + let stream_thread = tokio::task::spawn_blocking(move || -> Result<()> { // Bring up the (already-bound) data-plane socket. Default: hole-punch — wait briefly // for the client's punch, then stream to its OBSERVED source, so video traverses a // NAT / stateful inter-VLAN firewall (control + side planes ride the client-initiated @@ -1444,9 +1473,34 @@ async fn serve_session( } } } - }) - .await - .context("stream thread")??; + }); + // `stop` is only ADVISORY: the stream thread observes it between iterations, so a call that + // blocks without a bound INSIDE one (a compositor CLI that never returns, a D-Bus round-trip + // on a stuck bus, a driver wait on a hung GPU) never reaches the check — and nothing else + // can end the session, because every teardown below runs only once this await resolves. That + // made one stuck syscall a permanent zombie: it kept its semaphore slot (four of them and the + // host stops accepting entirely), its admission entry (a later client gets "host busy" + // forever) and its stream marker, and even the console's Stop button — which just sets this + // same flag — could not clear it. + // + // So bound the wait: once the session HAS been told to stop, give the thread + // `STREAM_STOP_GRACE` to return, then stop waiting for it and let teardown run. The thread is + // detached, not killed (a blocking thread can't be cancelled in Rust) — it keeps its capturer + // and encoder until the stuck call returns, and its own guards unwind if it ever does. That + // is a leak, but a bounded one: the session's slot and admission entry come back, so the rest + // of the host keeps serving. + tokio::select! { + joined = stream_thread => joined.context("stream thread")??, + () = stop_overdue(&stop) => { + tracing::error!( + grace_s = STREAM_STOP_GRACE.as_secs(), + "stream thread has not returned since the session was stopped — abandoning it so \ + the session slot is freed. Its capture/encoder stay held until the stuck call \ + returns; this is a HOST WEDGE — please report it with the log above" + ); + anyhow::bail!("stream thread wedged after stop"); + } + } // Give the client a moment to drain before the close. tokio::time::sleep(std::time::Duration::from_secs(1)).await; Ok(()) @@ -1462,13 +1516,25 @@ async fn serve_session( if result.is_ok() { 0u32 } else { 1u32 }.into(), if result.is_ok() { b"done" } else { b"error" }, ); - let _ = tokio::task::spawn_blocking(move || { + // Bounded, for the same reason the stream-thread wait is: the input thread exits only when the + // datagram task drops its channel, which the `conn.close()` above forces — but a join is the + // last unbounded await in teardown, and one stuck side thread must not hold the session's + // permit/admission entry (released when this fn returns) hostage. + let side_threads = tokio::task::spawn_blocking(move || { if let Some(h) = audio_handle { let _ = h.join(); } let _ = input_handle.join(); - }) - .await; + }); + if tokio::time::timeout(SIDE_THREAD_JOIN_GRACE, side_threads) + .await + .is_err() + { + tracing::warn!( + grace_s = SIDE_THREAD_JOIN_GRACE.as_secs(), + "audio/input threads did not exit after the connection closed — detaching them" + ); + } // The capture (and our gamescope session's VirtualOutput) are gone by here. If this was the // host-managed gamescope path on a box that autologs into gaming mode (Bazzite default), put the // TV's gaming session back so it's the default when no one is streaming.