diff --git a/crates/punktfunk-host/src/gamestream/audio.rs b/crates/punktfunk-host/src/gamestream/audio.rs index f9379b58..96ad2535 100644 --- a/crates/punktfunk-host/src/gamestream/audio.rs +++ b/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>>, + 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(¶ms); let mut enc = SessionEncoder::new(layout)?; @@ -427,7 +434,8 @@ fn audio_body( .encrypt_padded_vec_mut::(&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(()); } } diff --git a/crates/punktfunk-host/src/gamestream/control.rs b/crates/punktfunk-host/src/gamestream/control.rs index 511ae27b..f64d4383 100644 --- a/crates/punktfunk-host/src/gamestream/control.rs +++ b/crates/punktfunk-host/src/gamestream/control.rs @@ -83,10 +83,34 @@ pub fn spawn(state: Arc) -> 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) -> 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, .. diff --git a/crates/punktfunk-host/src/gamestream/mod.rs b/crates/punktfunk-host/src/gamestream/mod.rs index 98b3ee99..4e371aa4 100644 --- a/crates/punktfunk-host/src/gamestream/mod.rs +++ b/crates/punktfunk-host/src/gamestream/mod.rs @@ -182,7 +182,46 @@ pub struct AppState { pub stats: Arc, } +/// 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`). +pub(crate) type OnSessionLost = Arc; + 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]) { } } +#[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; diff --git a/crates/punktfunk-host/src/gamestream/nvhttp.rs b/crates/punktfunk-host/src/gamestream/nvhttp.rs index c6d339f8..3a0bc694 100644 --- a/crates/punktfunk-host/src/gamestream/nvhttp.rs +++ b/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("\n1\n".to_string()) } diff --git a/crates/punktfunk-host/src/gamestream/rtsp.rs b/crates/punktfunk-host/src/gamestream/rtsp.rs index ecf25d77..5f155178 100644 --- a/crates/punktfunk-host/src/gamestream/rtsp.rs +++ b/crates/punktfunk-host/src/gamestream/rtsp.rs @@ -190,7 +190,9 @@ fn authorized_launch(state: &AppState, peer: Option) -> Option) -> String { +// `&Arc` (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, peer: Option) -> String { match req.method.as_str() { "OPTIONS" => response( &req.cseq, @@ -254,6 +256,15 @@ fn handle_request(req: &Request, state: &AppState, peer: Option) -> 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) -> 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) -> ls.rikeyid, *state.audio_params.lock().unwrap(), state.audio_cap.clone(), + on_lost, ); } response(&req.cseq, &[("Session", "DEADBEEFCAFE;timeout = 90")], None) diff --git a/crates/punktfunk-host/src/gamestream/stream.rs b/crates/punktfunk-host/src/gamestream/stream.rs index 150888fe..0ada5a9b 100644 --- a/crates/punktfunk-host/src/gamestream/stream.rs +++ b/crates/punktfunk-host/src/gamestream/stream.rs @@ -45,6 +45,7 @@ pub type RfiSlot = Arc>>; /// 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, @@ -53,6 +54,7 @@ pub fn start( rfi_range: RfiSlot, video_cap: CapturerSlot, stats: Arc, + 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, + // 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, frame_interval: Duration, running: Arc, + 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, // 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::(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(); diff --git a/crates/punktfunk-host/src/mgmt/session.rs b/crates/punktfunk-host/src/mgmt/session.rs index 165e5c15..c323939e 100644 --- a/crates/punktfunk-host/src/mgmt/session.rs +++ b/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>) -> 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();