From ac3dc4323f0f3e972969efc29be3361669e08d23 Mon Sep 17 00:00:00 2001 From: enricobuehler Date: Fri, 24 Jul 2026 01:21:55 +0200 Subject: [PATCH] fix(native/session): make the stop flag enforceable so a session can't outlive its client MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Native sessions could survive long after the client was gone, in two independent ways. HOST: `stop` was only advisory. The one thing that ends a session is the stream thread returning — every teardown (conn.close, the joins, and the RAII drops of the session permit, admission entry and stream marker) sits after that await, and nothing forced the issue. The encode loop checks `stop` between iterations, so any unbounded call INSIDE one never reaches the check, and one stuck syscall became a permanent zombie: it held its semaphore slot (four of those and the host stops accepting QUIC entirely), its admission entry (a later client gets "host busy" forever), and not even the console's Stop button could clear it — that button sets this same flag. * Bound the wait: once the session has been told to stop, the thread gets STREAM_STOP_GRACE (90 s, well past the 40 s capture-rebuild budget) to return, then teardown runs anyway. The thread is detached, not killed — Rust can't cancel a blocking thread — so it keeps its capturer/encoder until the stuck call returns, but the session's slot and admission entry come back and the host keeps serving. It logs at ERROR as the host wedge it is. * Bound the audio/input joins too — the last unbounded await in teardown. * Take the session permit AFTER the QUIC handshake instead of before `accept()`, so a host at its concurrency cap still accepts and the waiting client sees a live path instead of a silent dial timeout. * Bound the compositor helpers that caused the wedge in the first place: new pf-vdisplay `proc::{status_within, output_within}` kill a child that outlives its budget. `kscreen-doctor` is a Wayland client of the very compositor it configures, so against a wedged KWin it never returned; same for systemctl/dbus against a stuck session bus. CLIENTS: the connection was never closed, so the host was right to keep the session — it still had a live, keep-alive-answering peer. * Android: backgrounding did no teardown at all, and Android doesn't suspend the process, so the worker kept answering keep-alives until the OS reclaimed it (on a TV box, never). End the session on ON_STOP, via the existing onDispose path; a plain close, not a quit, so the host lingers the display for a fast return. * Apple: the .background arm was iOS-only AND gated on an opt-in that defaults off, so backgrounding did nothing — while the `audio` background mode kept the app (and its connection) alive indefinitely. Act unconditionally, and cover tvOS. * Core: `conn.close()` only queues the frame, and run_pump is the body of a block_on whose runtime is dropped the instant it returns, so the driver could never put it on the wire — a deliberate quit reached the host as silence (8 s idle timeout, no quit code, and the linger meant for an unwanted disconnect). Carry the endpoint out of the handshake and flush with wait_idle(), the same discipline the pairing and probe paths already use. Linux check/clippy/tests green: 262 host, 71 pf-vdisplay (incl. new bounded-process tests), 231 core. Co-Authored-By: Claude Fable 5 --- .../kotlin/io/unom/punktfunk/StreamScreen.kt | 23 ++++ .../Sources/PunktfunkClient/ContentView.swift | 26 +++- crates/pf-vdisplay/src/lib.rs | 5 + crates/pf-vdisplay/src/vdisplay/linux/kwin.rs | 88 +++++++------- crates/pf-vdisplay/src/vdisplay/proc.rs | 114 ++++++++++++++++++ crates/pf-vdisplay/src/vdisplay/session.rs | 59 ++++++--- crates/punktfunk-core/src/client/pump.rs | 8 ++ .../src/client/pump/handshake.rs | 7 ++ crates/punktfunk-host/src/native.rs | 92 ++++++++++++-- 9 files changed, 338 insertions(+), 84 deletions(-) create mode 100644 crates/pf-vdisplay/src/vdisplay/proc.rs 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.