fix(gamestream/session): end the session when the client disconnects or vanishes

GameStream sessions outlived their client: the only complete teardowns
were the explicit ones (RTSP TEARDOWN, nvhttp /cancel, mgmt DELETE
/session), and the only automatic detector was a media-UDP send error —
which needs an ICMP port-unreachable, so a true vanish (Wi-Fi drop,
sleep, power off, crash) left video+audio encoding into the void
forever, and even a plain Moonlight quit (which sends neither TEARDOWN
nor /cancel) leaked the session. The stale state then cascaded: a
lingering launch 503-blocked a different client under
mode_conflict=reject, and streaming=true made a reconnect's PLAY take
its "stream already running" branch — no new threads, old threads still
aimed at the dead endpoint, the reconnect got no media.

ENet already detects all of this — the control peer's reliable-ping
timeout (or clean disconnect) fires Event::Disconnect within ~5-30 s —
but the handler only reset input state. Wire the real teardown into it:

* AppState::end_session — THE compat-plane session teardown: stops both
  media threads (their flags), clears launch + negotiated stream config;
  idempotent. /cancel and mgmt stop_session now share it.
* control.rs Disconnect → end_session. Gated on the TRACKED session
  peer, and Connect only tracks a peer from the /launch owner's IP (the
  same source-IP bind the RTSP/media plane uses), so an unauthenticated
  LAN peer connect+disconnect can't end a live session, and a fast
  reconnect's stale-peer timeout can't kill its successor.
* Client-unreachable UDP send errors now end the whole session via an
  OnSessionLost callback (built at PLAY) instead of stopping only the
  plane that noticed — audio no longer keeps streaming after video
  detects the dead client, and vice versa.

Linux check/clippy/tests green (53 gamestream tests incl. the new
end_session regression test).

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

crates/punktfunk-host/src/gamestream/audio.rs

@@ -220,12 +220,13 @@ pub fn start(
     rikeyid: i32,
     params: AudioParams,
     audio_cap: AudioCapSlot,
+    on_lost: super::OnSessionLost,
 ) {
     let _ = std::thread::Builder::new()
         .name("punktfunk-audio".into())
         .spawn(move || {
             tracing::info!(?params, "audio stream starting");
-            if let Err(e) = run(&running, &gcm_key, rikeyid, params, &audio_cap) {
+            if let Err(e) = run(&running, &gcm_key, rikeyid, params, &audio_cap, &on_lost) {
                 tracing::error!(error = %format!("{e:#}"), "audio stream failed");
             }
             running.store(false, Ordering::SeqCst);
@@ -243,6 +244,7 @@ pub fn start(
     _rikeyid: i32,
     _params: AudioParams,
     _audio_cap: AudioCapSlot,
+    _on_lost: super::OnSessionLost,
 ) {
     tracing::error!("GameStream audio requires Linux (PipeWire) or Windows (WASAPI) + libopus");
     running.store(false, std::sync::atomic::Ordering::SeqCst);
@@ -255,6 +257,7 @@ fn run(
     rikeyid: i32,
     params: AudioParams,
     audio_cap: &std::sync::Mutex<Option<Box<dyn AudioCapturer>>>,
+    on_lost: &super::OnSessionLost,
 ) -> Result<()> {
     let sock = UdpSocket::bind(("0.0.0.0", AUDIO_PORT)).context("bind audio UDP")?;
     // Grow SO_SNDBUF/RCVBUF; the opt-in DSCP/QoS tag happens after connect below (Windows
@@ -296,7 +299,7 @@ fn run(
         }
         None => audio::open_audio_capture(want).context("open audio capture")?,
     };
-    let result = audio_body(&mut *cap, &sock, gcm_key, rikeyid, params, running);
+    let result = audio_body(&mut *cap, &sock, gcm_key, rikeyid, params, running, on_lost);
     cap.idle(); // parked between sessions — release the routing claim (Linux stream sink)
     audio::park_audio_capture(audio_cap, cap); // drop on Windows (restores the default), keep on Linux
     result
@@ -355,6 +358,7 @@ impl SessionEncoder {
 }
 
 #[cfg(any(target_os = "linux", target_os = "windows"))]
+#[allow(clippy::too_many_arguments)]
 fn audio_body(
     cap: &mut dyn AudioCapturer,
     sock: &UdpSocket,
@@ -362,6 +366,9 @@ fn audio_body(
     rikeyid: i32,
     params: AudioParams,
     running: &AtomicBool,
+    // Whole-session teardown for the client-unreachable send errors below — video would
+    // otherwise keep streaming at the dead endpoint (see `AppState::end_session`).
+    on_lost: &super::OnSessionLost,
 ) -> Result<()> {
     let layout = layout_for(&params);
     let mut enc = SessionEncoder::new(layout)?;
@@ -427,7 +434,8 @@ fn audio_body(
                 .encrypt_padded_vec_mut::<Pkcs7>(&out[..n]);
             let pkt = build_rtp(seq, timestamp, &ct);
             if sock.send(&pkt).is_err() {
-                tracing::info!(sent, "audio: client unreachable — stopping");
+                tracing::info!(sent, "audio: client unreachable — ending session");
+                on_lost();
                 return Ok(());
             }
             // Surround FEC: accumulate the encrypted payloads of the aligned 4-packet block;
@@ -449,7 +457,11 @@ fn audio_body(
                                 let fp =
                                     build_fec_rtp(rtp_seq, x as u8, fec_base_seq, fec_base_ts, par);
                                 if sock.send(&fp).is_err() {
-                                    tracing::info!(sent, "audio: client unreachable — stopping");
+                                    tracing::info!(
+                                        sent,
+                                        "audio: client unreachable — ending session"
+                                    );
+                                    on_lost();
                                     return Ok(());
                                 }
                             }

crates/punktfunk-host/src/gamestream/control.rs

@@ -83,10 +83,34 @@ pub fn spawn(state: Arc<AppState>) -> Result<()> {
                     match host.service() {
                         Ok(Some(event)) => match event {
                             Event::Connect { peer: p, .. } => {
-                                tracing::info!("control: client connected");
-                                peer = Some(p.id());
+                                // Track this peer as THE session peer only if it comes from the
+                                // `/launch` owner's IP (when captured — `None` falls back to
+                                // trusting the connect, the pre-teardown behavior). The tracked
+                                // peer's disconnect now ENDS the session, so an unauthenticated
+                                // LAN peer that connects+disconnects on 47999 must not be able to
+                                // steal the slot and tear a live session down. Same source-IP
+                                // bind the RTSP/media plane uses (security-review #4).
+                                let owner_ip = state.launch.lock().unwrap().and_then(|s| s.peer_ip);
+                                let from = p.address().map(|a| a.ip());
+                                if owner_ip.is_some() && from.is_some() && owner_ip != from {
+                                    tracing::warn!(
+                                        ?from,
+                                        "control: peer connected from a non-owner IP — ignoring"
+                                    );
+                                } else {
+                                    tracing::info!("control: client connected");
+                                    peer = Some(p.id());
+                                }
                             }
-                            Event::Disconnect { .. } => {
+                            Event::Disconnect { peer: p, .. } => {
+                                // Gate on the TRACKED session peer: a stray probe peer (or the
+                                // OLD peer's late timeout after a fast reconnect replaced it in
+                                // the Connect arm) must neither clobber the live session's input
+                                // state nor end its session.
+                                if peer != Some(p.id()) {
+                                    tracing::debug!("control: non-session peer disconnected");
+                                    continue;
+                                }
                                 tracing::info!("control: client disconnected");
                                 detected = None;
                                 decrypt_fails = 0;
@@ -97,6 +121,16 @@ pub fn spawn(state: Arc<AppState>) -> Result<()> {
                                 // uinput pen releases any held tool/tip kernel-side).
                                 pads = GamepadManager::new();
                                 pointer = super::pen::GsPointer::new();
+                                // The control stream is the session's liveness anchor — Moonlight
+                                // holds it for the whole stream, and ENet detects a vanished peer
+                                // via its reliable-ping timeout (~5–30 s), which ALSO lands here.
+                                // End the session: without this, a client that disconnects without
+                                // an explicit RTSP TEARDOWN / nvhttp `/cancel` (a network drop,
+                                // sleep, crash — or just a plain Moonlight quit, which sends
+                                // neither) left the media threads streaming at the dead endpoint
+                                // forever (a UDP send only errors on an ICMP port-unreachable) and
+                                // the stale launch/streaming state wedged every reconnect.
+                                state.end_session("control stream disconnected");
                             }
                             Event::Receive {
                                 channel_id, packet, ..

crates/punktfunk-host/src/gamestream/mod.rs

@@ -182,7 +182,46 @@ pub struct AppState {
     pub stats: Arc<crate::stats_recorder::StatsRecorder>,
 }
 
+/// Session-lost callback the media threads invoke when they detect the client is unreachable
+/// (a UDP send error): ends the WHOLE GameStream session via [`AppState::end_session`], not just
+/// the thread that noticed — video and audio otherwise stop independently and leave the launch
+/// state behind. Built by the RTSP PLAY handler (the one place with the `Arc<AppState>`).
+pub(crate) type OnSessionLost = Arc<dyn Fn() + Send + Sync>;
+
 impl AppState {
+    /// End the GameStream session as one unit: signal BOTH media threads to stop (they observe
+    /// their `streaming`/`audio_streaming` flags) and clear the launch + negotiated stream
+    /// config. Idempotent — safe to call from every "the client is gone" site.
+    ///
+    /// This is THE teardown for the compat plane. Anything less leaves a stale session behind:
+    /// a lingering `launch` 503-blocks a different client's `/launch` under
+    /// `mode_conflict = reject`, and a stale `streaming = true` makes a reconnect's RTSP PLAY
+    /// take its "stream already running" branch while the old threads still stream at the
+    /// vanished client's endpoint (no new threads are started — the reconnect gets no media).
+    /// Returns whether the video stream was live (for the caller's log line).
+    pub(crate) fn end_session(&self, reason: &str) -> bool {
+        use std::sync::atomic::Ordering;
+        let was_streaming = self.streaming.swap(false, Ordering::SeqCst);
+        let was_audio = self.audio_streaming.swap(false, Ordering::SeqCst);
+        let had_launch = self
+            .launch
+            .lock()
+            .unwrap_or_else(|e| e.into_inner())
+            .take()
+            .is_some();
+        self.stream.lock().unwrap_or_else(|e| e.into_inner()).take();
+        if was_streaming || was_audio || had_launch {
+            tracing::info!(
+                reason,
+                was_streaming,
+                was_audio,
+                had_launch,
+                "gamestream: session ended"
+            );
+        }
+        was_streaming
+    }
+
     /// Fresh control-plane state: no active session; the pairing allow-list is loaded from
     /// disk (pairings persist across restarts). `stats` is the shared recorder handed to both the
     /// mgmt API and the streaming loops.
@@ -429,6 +468,69 @@ pub(crate) fn save_paired(paired: &[Vec<u8>]) {
     }
 }
 
+#[cfg(test)]
+mod session_tests {
+    use super::*;
+
+    fn test_state() -> AppState {
+        let host = Host {
+            hostname: "test-host".into(),
+            uniqueid: "deadbeef".into(),
+            local_ip: IpAddr::V4(Ipv4Addr::LOCALHOST),
+            http_port: HTTP_PORT,
+            https_port: HTTPS_PORT,
+        };
+        let identity = cert::ServerIdentity::ephemeral().expect("ephemeral identity");
+        let stats = crate::stats_recorder::StatsRecorder::new(std::env::temp_dir().join(format!(
+            "pf-gs-endsession-{}-{:p}",
+            std::process::id(),
+            &0u8 as *const u8
+        )));
+        AppState::new(host, identity, stats)
+    }
+
+    /// `end_session` is THE compat-plane teardown: one call must clear the whole session — both
+    /// media-thread flags, the launch, and the negotiated stream config — and be idempotent.
+    /// Guards the ENet-Disconnect / client-unreachable paths that previously stopped nothing
+    /// (the "session stays alive after the client disconnects" bug).
+    #[test]
+    fn end_session_clears_the_whole_session() {
+        use std::sync::atomic::Ordering;
+        let state = test_state();
+        state.streaming.store(true, Ordering::SeqCst);
+        state.audio_streaming.store(true, Ordering::SeqCst);
+        *state.launch.lock().unwrap() = Some(LaunchSession {
+            gcm_key: [0; 16],
+            rikeyid: 0,
+            width: 1920,
+            height: 1080,
+            fps: 60,
+            appid: 1,
+            peer_ip: None,
+            owner_fp: None,
+        });
+        *state.stream.lock().unwrap() = Some(stream::StreamConfig {
+            width: 1920,
+            height: 1080,
+            fps: 60,
+            packet_size: 1024,
+            bitrate_kbps: 20_000,
+            codec: crate::encode::Codec::H265,
+            min_fec: 0,
+            hdr: false,
+        });
+
+        assert!(state.end_session("test"), "video was live");
+        assert!(!state.streaming.load(Ordering::SeqCst));
+        assert!(!state.audio_streaming.load(Ordering::SeqCst));
+        assert!(state.launch.lock().unwrap().is_none());
+        assert!(state.stream.lock().unwrap().is_none());
+
+        // Idempotent: a second end (e.g. `/cancel` racing the ENet Disconnect) is a no-op.
+        assert!(!state.end_session("test again"));
+    }
+}
+
 #[cfg(all(test, unix))]
 mod tests {
     use std::os::unix::fs::PermissionsExt;

crates/punktfunk-host/src/gamestream/nvhttp.rs

@@ -250,14 +250,9 @@ async fn h_cancel(
         tracing::warn!("cancel rejected — caller does not own the session");
         return xml(error_xml());
     }
-    *st.launch.lock().unwrap() = None;
-    // Quit semantics: stop the running media threads (they observe these flags) so the session
-    // actually ends — the virtual output/gamescope teardown follows via the capturer's RAII.
-    st.streaming
-        .store(false, std::sync::atomic::Ordering::SeqCst);
-    st.audio_streaming
-        .store(false, std::sync::atomic::Ordering::SeqCst);
-    tracing::info!("cancel — launch session cleared, streams stopping");
+    // Quit semantics: the shared full teardown (launch cleared + both media threads stop on
+    // their flags) — the virtual output/gamescope teardown follows via the capturer's RAII.
+    st.end_session("client /cancel");
     xml("<?xml version=\"1.0\" encoding=\"utf-8\"?>\n<root status_code=\"200\"><cancel>1</cancel></root>\n".to_string())
 }
 

crates/punktfunk-host/src/gamestream/rtsp.rs

@@ -190,7 +190,9 @@ fn authorized_launch(state: &AppState, peer: Option<SocketAddr>) -> Option<Launc
     }
 }
 
-fn handle_request(req: &Request, state: &AppState, peer: Option<SocketAddr>) -> String {
+// `&Arc<AppState>` (not `&AppState`): PLAY hands the media threads a `'static` session-lost
+// callback, which needs an owned clone of the state.
+fn handle_request(req: &Request, state: &Arc<AppState>, peer: Option<SocketAddr>) -> String {
     match req.method.as_str() {
         "OPTIONS" => response(
             &req.cseq,
@@ -254,6 +256,15 @@ fn handle_request(req: &Request, state: &AppState, peer: Option<SocketAddr>) ->
                 return response_status("401 Unauthorized", &req.cseq, &[], None);
             };
             let cfg = *state.stream.lock().unwrap();
+            // Client-unreachable teardown for the media threads: ends the WHOLE session (both
+            // planes + launch state), so one plane detecting the dead client can't leave the
+            // other streaming at it — or leave a stale launch to wedge the next connect.
+            let on_lost: super::OnSessionLost = {
+                let st = state.clone();
+                Arc::new(move || {
+                    st.end_session("client unreachable");
+                })
+            };
             match cfg {
                 Some(cfg) if !state.streaming.swap(true, Ordering::SeqCst) => {
                     // Resolve the launched catalog entry (session recipe) for the stream.
@@ -267,6 +278,7 @@ fn handle_request(req: &Request, state: &AppState, peer: Option<SocketAddr>) ->
                         state.rfi_range.clone(),
                         state.video_cap.clone(),
                         state.stats.clone(),
+                        on_lost.clone(),
                     );
                 }
                 Some(_) => tracing::info!("RTSP PLAY — stream already running"),
@@ -283,6 +295,7 @@ fn handle_request(req: &Request, state: &AppState, peer: Option<SocketAddr>) ->
                     ls.rikeyid,
                     *state.audio_params.lock().unwrap(),
                     state.audio_cap.clone(),
+                    on_lost,
                 );
             }
             response(&req.cseq, &[("Session", "DEADBEEFCAFE;timeout = 90")], None)

crates/punktfunk-host/src/gamestream/stream.rs

@@ -45,6 +45,7 @@ pub type RfiSlot = Arc<std::sync::Mutex<Option<(i64, i64)>>>;
 /// Spawn the video stream thread (idempotent via `running`). Stops when `running` clears.
 /// `force_idr` is set by the control stream on a client recovery request; `video_cap` holds
 /// the persistent capturer the thread borrows for the stream's duration.
+#[allow(clippy::too_many_arguments)]
 pub fn start(
     cfg: StreamConfig,
     app: Option<super::apps::AppEntry>,
@@ -53,6 +54,7 @@ pub fn start(
     rfi_range: RfiSlot,
     video_cap: CapturerSlot,
     stats: Arc<crate::stats_recorder::StatsRecorder>,
+    on_lost: super::OnSessionLost,
 ) {
     let _ = std::thread::Builder::new()
         .name("punktfunk-video".into())
@@ -103,6 +105,7 @@ pub fn start(
                 &rfi_range,
                 &video_cap,
                 &stats,
+                &on_lost,
             );
             // A clean return is a stop (RTSP teardown / cancel / client unreachable) → `quit`;
             // an error return is `error`. The compat plane can't tell a user stop from an idle
@@ -137,6 +140,8 @@ fn run(
     // Shared stats recorder for the web-console capture/graph. Threaded into `stream_body` (the
     // encode loop); per-frame sample emission is wired by a later pass.
     stats: &Arc<crate::stats_recorder::StatsRecorder>,
+    // Whole-session teardown for the send thread's client-unreachable detection.
+    on_lost: &super::OnSessionLost,
 ) -> Result<()> {
     // GameStream capture/encode thread: apply Windows session tuning (no-op off Windows).
     pf_frame::session_tuning::on_hot_thread();
@@ -252,6 +257,7 @@ fn run(
             rfi_range,
             stats,
             &client_label,
+            on_lost,
         );
     }
 
@@ -299,6 +305,7 @@ fn run(
         rfi_range,
         stats,
         &client_label,
+        on_lost,
     );
     capturer.set_active(false);
     *video_cap.lock().unwrap() = Some((capturer, cfg.hdr));
@@ -572,12 +579,15 @@ fn spawn_packetizer(
 /// shared [`send_pacing`](crate::send_pacing) policy at the GameStream parameterization: no
 /// microburst stage, a BOUNDED step count (≤ 12, chunk ≥ 16, see the policy's docs for the
 /// "send queue full" history that bound guards), each step ending in a sleep toward its slice
-/// of the fixed budget. On send failure (client gone) it clears `running`.
+/// of the fixed budget. On send failure (client gone) it ends the whole session via `on_lost` —
+/// not just this thread: audio would otherwise keep streaming at the dead endpoint and the stale
+/// launch state would wedge the next connect (see `AppState::end_session`).
 fn spawn_sender(
     sock: UdpSocket,
     rx: std::sync::mpsc::Receiver<PacketBatch>,
     frame_interval: Duration,
     running: Arc<AtomicBool>,
+    on_lost: super::OnSessionLost,
 ) -> Result<()> {
     std::thread::Builder::new()
         .name("punktfunk-send".into())
@@ -613,8 +623,9 @@ fn spawn_sender(
                     },
                 );
                 if let Err(e) = r {
-                    tracing::info!(error = %e, sent, "video: client unreachable — stopping stream");
+                    tracing::info!(error = %e, sent, "video: client unreachable — ending session");
                     running.store(false, Ordering::SeqCst);
+                    on_lost();
                     return;
                 }
             }
@@ -645,6 +656,8 @@ fn stream_body(
     stats: &Arc<crate::stats_recorder::StatsRecorder>,
     // Short client label (peer IP) seeded into the capture meta on the first armed registration.
     client_label: &str,
+    // Whole-session teardown, handed to the send thread's client-unreachable detection.
+    on_lost: &super::OnSessionLost,
 ) -> Result<()> {
     // The first frame establishes the authoritative size/format for the encoder.
     let mut frame = capturer.next_frame().context("capture first frame")?;
@@ -708,6 +721,7 @@ fn stream_body(
         batch_rx,
         Duration::from_secs_f64(1.0 / target_fps as f64),
         running.clone(),
+        on_lost.clone(),
     )?;
     let (raw_tx, raw_rx) = std::sync::mpsc::sync_channel::<RawFrame>(2);
     spawn_packetizer(raw_rx, batch_tx, pk, goodput.clone())?;
@@ -1075,6 +1089,7 @@ mod tests {
             rx,
             Duration::from_millis(8), // ~120fps frame interval
             running.clone(),
+            Arc::new(|| {}),
         )
         .unwrap();
 

crates/punktfunk-host/src/mgmt/session.rs

@@ -19,11 +19,8 @@ use std::sync::atomic::Ordering;
     )
 )]
 pub(crate) async fn stop_session(State(st): State<Arc<MgmtState>>) -> StatusCode {
-    let was_streaming = st.app.streaming.swap(false, Ordering::SeqCst);
-    st.app.audio_streaming.store(false, Ordering::SeqCst);
-    *st.app.launch.lock().unwrap_or_else(|e| e.into_inner()) = None;
-    *st.app.stream.lock().unwrap_or_else(|e| e.into_inner()) = None;
-    // Native plane: the GameStream flags above don't reach it (it runs its own loops off the shared
+    let was_streaming = st.app.end_session("management API stop");
+    // Native plane: the GameStream teardown above doesn't reach it (it runs its own loops off the shared
     // session registry), so signal every live native session to tear down too.
     let native = crate::session_status::count();
     crate::session_status::stop_all();