diff --git a/crates/punktfunk-host/src/gamelease.rs b/crates/punktfunk-host/src/gamelease.rs index 975198a8..d40bcfd1 100644 --- a/crates/punktfunk-host/src/gamelease.rs +++ b/crates/punktfunk-host/src/gamelease.rs @@ -254,7 +254,14 @@ pub struct LeaseRequest { /// Seconds-since-boot from **before** the launch ([`launch_clock`]): the floor for adopting a /// process, which is what keeps a copy of the game the player already had open from being /// mistaken for this session's. `None` disables the filter (no readable uptime clock). + /// + /// A *reconnecting* session inherits this from [`crate::launchreg`] rather than minting its own, + /// which is the only way its lease can see a game the previous session started. pub launch_stamp: Option, + /// Where the watcher publishes the processes it adopts, so the host's launch record can still + /// answer "is our launch up?" after this lease and its watcher are gone + /// ([`crate::launchreg::LiveProcs`]). `None` for a launch that isn't recorded. + pub procs: Option, } /// The reference instant for adopting this launch's processes, in seconds since boot. Call it @@ -284,6 +291,7 @@ pub fn open(req: LeaseRequest, on_exit: OnExit) -> GameLease { nested, child, launch_stamp, + procs, } = req; let kind = if nested { @@ -333,7 +341,7 @@ pub fn open(req: LeaseRequest, on_exit: OnExit) -> GameLease { ); } - let watcher = spawn_watcher(shared.clone(), child, on_exit); + let watcher = spawn_watcher(shared.clone(), child, procs, on_exit); if watcher.is_none() { // Nothing is polling this lease (no signals to poll, or a platform without a matcher yet), so // its state will never advance on its own. Report it as running rather than leaving the @@ -349,6 +357,7 @@ pub fn open(req: LeaseRequest, on_exit: OnExit) -> GameLease { fn spawn_watcher( shared: Arc, child: Option, + procs: Option, on_exit: OnExit, ) -> Option> { // An untracked lease has nothing to observe (it still exposes state for the status surface). @@ -369,23 +378,45 @@ fn spawn_watcher( // surface, but nothing polls it. #[cfg(not(any(target_os = "linux", windows)))] { - let _ = (child, on_exit); + let _ = (child, procs, on_exit); return None; } #[cfg(any(target_os = "linux", windows))] { std::thread::Builder::new() .name("pf1-gamelease".into()) - .spawn(move || watch(shared, child, on_exit)) + .spawn(move || watch(shared, child, procs, on_exit)) .ok() } } /// The watch loop: wait for the game to appear, then for it to go away. #[cfg(any(target_os = "linux", windows))] -fn watch(shared: Arc, mut child: Option, on_exit: OnExit) { +fn watch( + shared: Arc, + mut child: Option, + procs: Option, + on_exit: OnExit, +) { let scanner = crate::procscan::Scanner::system(); let cancelled = || shared.cancel.load(Ordering::Relaxed); + // Publish what this lease adopted to the host's launch record, so a LATER session can tell "this + // host's launch is still up" from "nothing of ours is running" — which is what lets it inherit + // this launch instead of starting a second copy (`crate::launchreg`). + // + // Only ever the CONCRETE processes, and only ever a non-empty set. Never the spec: a later re-scan + // by spec would find a copy the player started for themselves since, and adopting that is exactly + // what procscan's rule 1 forbids. And never cleared on exit: the last set the watcher saw is what + // makes the record answer `Gone` (every recorded pid re-verified dead) rather than "no opinion", + // which is how a game the player quit becomes relaunchable at once. + let publish = |live: &[crate::procscan::ProcRef]| { + if live.is_empty() { + return; + } + if let Some(slot) = procs.as_ref() { + *slot.lock().unwrap_or_else(|e| e.into_inner()) = live.to_vec(); + } + }; let spawned_at = Instant::now(); let mut kind = shared.kind.clone(); @@ -481,6 +512,7 @@ fn watch(shared: Arc, mut child: Option, on_ex let live = scanner.find(&shared.spec, shared.launch_stamp); if !live.is_empty() || child_alive { known = live.clone(); + publish(&live); shared.was_running.store(true, Ordering::Relaxed); shared.last_seen_ms.store(now_ms(), Ordering::Relaxed); shared.set_state(GameState::Running); @@ -536,6 +568,7 @@ fn watch(shared: Arc, mut child: Option, on_ex } }; if !live.is_empty() || child_alive { + publish(&live); known = live; gone_since = None; vetoed = false; @@ -790,8 +823,17 @@ fn windows_term_ladder(shared: &LeaseShared) { // The grace registry: leases whose session is gone but whose game is on probation // --------------------------------------------------------------------------------------------- -/// A lease waiting out its reconnect window. If the client comes back before the deadline the lease -/// is handed to the new session and nothing is ended; if it doesn't, the game ends. +/// A lease waiting out its reconnect window. If the client comes back before the deadline the +/// pending termination is dropped and the game keeps running; if it doesn't, the game ends. +/// +/// The lease object itself is **not** handed to the new session, and cannot be: by the time an entry +/// lands here its [`GameLease`] has already been dropped (the guard's `Drop` runs [`on_session_end`] +/// and then drops the lease), which cancels its watcher — and its `on_exit` action closes a +/// connection that no longer exists. What the new session re-adopts is the *game*, through +/// [`crate::launchreg`], which is what carries the original launch's reference instant across +/// sessions so a fresh lease can see a game started before it. (This doc used to claim the lease was +/// handed over; nothing ever did that, and a reconnecting session was left with no game-exit +/// detection at all.) pub struct Pending { pub shared: Arc, pub deadline: Instant, @@ -826,10 +868,16 @@ pub fn arm_grace(shared: Arc, fingerprint: Option, grace: D } /// A reconnecting client takes its game back: drops any pending termination for `fingerprint` whose -/// title matches `app`. Returns the number of leases reprieved. -pub fn readopt(fingerprint: Option<&str>, app: Option<&str>) -> usize { +/// title matches `app`. +/// +/// Returns the reprieved leases, so a caller can name what it saved (and read the launch it came +/// from) rather than being handed a bare count. They are **corpses by design** — see [`Pending`]: +/// their watchers are cancelled and their exit actions point at a dead connection. The new session +/// opens its own lease; what it needs from the old launch (the reference instant to adopt against) +/// comes from [`crate::launchreg`], not from here. +pub fn readopt(fingerprint: Option<&str>, app: Option<&str>) -> Vec> { let mut reg = registry().lock().unwrap_or_else(|e| e.into_inner()); - let before = reg.len(); + let mut reprieved = Vec::new(); reg.retain(|p| { let same_client = match (&p.fingerprint, fingerprint) { (Some(a), Some(b)) => a == b, @@ -843,12 +891,13 @@ pub fn readopt(fingerprint: Option<&str>, app: Option<&str>) -> usize { title = %p.shared.game.title, "the client reconnected inside the window — the game keeps running" ); + reprieved.push(p.shared.clone()); false } else { true } }); - before - reg.len() + reprieved } /// Every lease currently on probation, with the time left, for the status surface. @@ -915,13 +964,40 @@ fn start_reaper() { /// What a session should do with its game when it ends. The policy lives in /// [`crate::session_settings`]; this is the one place that turns it into an action, so both planes /// behave identically. -pub fn on_session_end(lease: &GameLease, deliberate: bool, fingerprint: Option<&str>) { +/// +/// `launch` is this session's hold on the host's launch record ([`crate::launchreg`]), consulted for +/// one question only: has a newer session already taken this launch over? +pub fn on_session_end( + lease: &GameLease, + deliberate: bool, + fingerprint: Option<&str>, + launch: Option<&crate::launchreg::Claim>, +) { use crate::session_settings::GameOnSessionEnd; let settings = crate::session_settings::get(); let shared = lease.shared(); if !shared.is_trackable() || shared.state() == GameState::Exited { return; // nothing to end (or it already ended on its own) } + // A newer session has already claimed this launch: the game this lease tracks is the game that + // session is now streaming. Anything this policy would do to "our" game would be done to theirs, + // so it does nothing at all. + // + // **Which order this runs in relative to the new session's handshake does not matter, and that is + // the point.** The two are concurrent — the old session's stream loop exits (here) while the new + // one is already deciding its launch. If the teardown wins the race, `superseded` is false and the + // policy runs exactly as it always has; the new session then finds the record released and adopts + // it through the window/liveness arms. If the handshake wins, the new session's claim is already + // recorded when this runs, and this returns — which is the case that needed fixing: under + // `Always`, the handshake's `readopt` would have run BEFORE this `arm_grace` and so could not + // reprieve it, and the reaper would have ended the new session's game when the window closed. + if launch.is_some_and(|c| c.superseded()) { + tracing::info!( + title = %shared.game.title, + "this client already came back for this game — leaving it to the session that has it now" + ); + return; + } let end_now = |shared: Arc| { // A deliberate stop already forces this session's display down immediately (the `quit` flag // beats keep-alive linger), and for a nested launch that teardown *is* what ends the game — @@ -977,16 +1053,28 @@ pub struct SessionGuard { quit: Arc, /// Hex client fingerprint, so a reconnecting client can reclaim its own game and nothing else. fingerprint: Option, + /// This session's hold on the host's launch record. Held here because its lifetime is exactly the + /// session's: its drop is what opens the reconnect window a re-dial is matched against, and it + /// must not happen until after the policy above has read it. Rust drops fields **after** the + /// `Drop` body, so declaring it here is what orders those two. + launch: Option, } impl SessionGuard { /// Bind `lease` to the calling session's lifetime. `quit` is the session's deliberate-stop flag, - /// read at drop; `fingerprint` identifies the client allowed to reclaim the game on reconnect. - pub fn new(lease: GameLease, quit: Arc, fingerprint: Option) -> Self { + /// read at drop; `fingerprint` identifies the client allowed to reclaim the game on reconnect; + /// `launch` is this session's claim on the host's launch record ([`crate::launchreg::claim`]). + pub fn new( + lease: GameLease, + quit: Arc, + fingerprint: Option, + launch: Option, + ) -> Self { Self { lease, quit, fingerprint, + launch, } } @@ -1002,6 +1090,7 @@ impl Drop for SessionGuard { &self.lease, self.quit.load(Ordering::SeqCst), self.fingerprint.as_deref(), + self.launch.as_ref(), ); } } @@ -1027,6 +1116,8 @@ mod tests { child: None, // No start-time floor: these leases are never matched against real processes. launch_stamp: None, + // Not a recorded launch — nothing here spawns anything (`crate::launchreg`). + procs: None, } } @@ -1096,14 +1187,16 @@ mod tests { Duration::from_secs(3_600), ); // A different client, or a different title, does not reprieve it. - assert_eq!(readopt(Some("fp-other"), Some(id)), 0); - assert_eq!(readopt(Some("fp-130"), Some("steam:9999")), 0); + assert!(readopt(Some("fp-other"), Some(id)).is_empty()); + assert!(readopt(Some("fp-130"), Some("steam:9999")).is_empty()); // A missing fingerprint on either side must not reprieve anything either — otherwise any // unidentified reconnect could keep any game alive. - assert_eq!(readopt(None, Some(id)), 0); + assert!(readopt(None, Some(id)).is_empty()); assert!(is_pending(id), "none of those should have reprieved it"); - // The right client coming back for the right title does. - assert_eq!(readopt(Some("fp-130"), Some(id)), 1); + // The right client coming back for the right title does — and names what it saved. + let saved = readopt(Some("fp-130"), Some(id)); + assert_eq!(saved.len(), 1); + assert_eq!(saved[0].game.id.as_deref(), Some(id)); assert!(!is_pending(id)); } @@ -1118,7 +1211,7 @@ mod tests { .expect("armed lease is pending"); assert!(mine.1 > 290 && mine.1 <= 300, "remaining was {}", mine.1); // Leave the registry as we found it, so a sibling test's sweep can't see this entry. - assert_eq!(readopt(Some("fp-140"), Some(id)), 1); + assert_eq!(readopt(Some("fp-140"), Some(id)).len(), 1); } #[test] @@ -1144,7 +1237,7 @@ mod tests { assert!(!lb.shared().is_terminating()); // An id nobody is waiting on ends nothing. assert_eq!(end_pending(Some("steam:99999")), 0); - assert_eq!(readopt(Some("fp-151"), Some(b)), 1); + assert_eq!(readopt(Some("fp-151"), Some(b)).len(), 1); } /// A launcher that hands off and exits must never be mistaken for the game. @@ -1180,6 +1273,7 @@ mod tests { nested: false, child: Some((child, false)), launch_stamp: None, + procs: None, }, Box::new(|| { EXITS.fetch_add(1, Ordering::SeqCst); @@ -1238,6 +1332,7 @@ mod tests { nested: false, child: Some((child, true)), launch_stamp, + procs: None, }, Box::new(|| { EXITS.fetch_add(1, Ordering::SeqCst); diff --git a/crates/punktfunk-host/src/gamestream/stream.rs b/crates/punktfunk-host/src/gamestream/stream.rs index 37c7476f..2c4610f6 100644 --- a/crates/punktfunk-host/src/gamestream/stream.rs +++ b/crates/punktfunk-host/src/gamestream/stream.rs @@ -221,7 +221,7 @@ fn run( // steps, before the source (a bare-spawn gamescope nests the game inside it), before the // launch — because a reading taken later would reject the very process it is meant to find. // Erring early can only ever include more of our own launch, never a copy from before it. - let launch_stamp = crate::gamelease::launch_clock(); + let fresh_stamp = crate::gamelease::launch_clock(); // Everything the host knows about the title being launched, resolved in ONE library scan: // what to run, what to call it, and how to recognize it once it is up. let target = resolve_gs_app(app); @@ -231,14 +231,29 @@ fn run( if let Some(t) = target.as_ref() { let reprieved = crate::gamelease::readopt(life.fingerprint.as_deref(), t.game.id.as_deref()); - if reprieved > 0 { + if !reprieved.is_empty() { tracing::info!( - reprieved, + reprieved = reprieved.len(), title = %t.game.title, "gamestream: this client came back for its game — keeping it" ); } } + // ...and the other half of coming back for it: the host's own record of what it launched, for + // whom (`crate::launchreg`). Plane parity with the native path — same registry, same rule. + // A relaunch of a title this client's copy of which is still running neither starts a second + // copy nor mints a fresh reference instant the running game could never satisfy. A paired + // Moonlight client has a fingerprint; an anonymous one (or an operator-typed `apps.json` + // entry with no library id) is not recordable and behaves exactly as it always has. + let launch_claim = target.as_ref().map(|t| { + crate::launchreg::claim( + life.fingerprint.as_deref(), + t.game.id.as_deref(), + fresh_stamp, + ) + }); + let launch_stamp = launch_claim.as_ref().map_or(fresh_stamp, |c| c.stamp()); + let adopt_launch = launch_claim.as_ref().is_some_and(|c| !c.must_spawn()); // Per-app prep steps (RFC §6): the entry's own `prep` plus a custom library title's, // run synchronously BEFORE the virtual output opens or anything launches (an HDR // toggle / sink switch must land first — and gamescope's nested launch happens inside @@ -290,17 +305,34 @@ fn run( // store-qualified id — resolved against the host's OWN library (the client can only pick an // existing title, never inject a command). An apps.json entry instead carries an // operator-typed `cmd`. Library id wins when both are set. + // + // ...and once per LAUNCH, not once per `/launch` request: `adopt_launch` is the record's + // verdict that this client's copy of the title is already running (see above), and + // `spawned_now` is what actually happened — the record is settled from it below. + #[allow(unused_mut)] + let mut spawned_now = false; #[cfg(windows)] if let Some(t) = target.as_ref() { - // A library title launches by its store-qualified id (the interactive-session spawner - // resolves the store's own recipe); an operator-typed command runs as itself. - let launched = match (t.game.id.as_deref(), t.command.as_deref()) { - (Some(id), _) => crate::library::launch_gamestream_library(id), - (None, Some(cmd)) => crate::library::launch_gamestream_command(cmd), - (None, None) => Ok(()), - }; - if let Err(e) = launched { - tracing::warn!(title = %t.game.title, error = %e, "gamestream: could not launch app"); + if adopt_launch { + tracing::info!( + title = %t.game.title, + "gamestream: this client's copy of this title is already running — not starting \ + a second one" + ); + } else { + // A library title launches by its store-qualified id (the interactive-session spawner + // resolves the store's own recipe); an operator-typed command runs as itself. + let launched = match (t.game.id.as_deref(), t.command.as_deref()) { + (Some(id), _) => crate::library::launch_gamestream_library(id), + (None, Some(cmd)) => crate::library::launch_gamestream_command(cmd), + (None, None) => Ok(()), + }; + match launched { + Ok(()) => spawned_now = true, + Err(e) => { + tracing::warn!(title = %t.game.title, error = %e, "gamestream: could not launch app") + } + } } } // Linux keeps the spawned child rather than dropping it: it is the primary liveness signal @@ -309,11 +341,28 @@ fn run( // source open), so launching again would start it twice. #[cfg(target_os = "linux")] let spawned_launch = match target.as_ref().and_then(|t| t.command.as_deref()) { + // Already ours and still running: don't hand the player a second copy. The nested arm + // below reaches the same conclusion through the display registry, whose reuse key includes + // the launch command — a kept gamescope with the game inside it is re-attached, not + // respawned. + Some(cmd) if adopt_launch => { + tracing::info!( + command = %cmd, + "gamestream: this client's copy of this title is already running — not starting \ + a second one" + ); + None + } Some(_) if crate::vdisplay::launch_is_nested(compositor, gamescope_route.as_ref()) => { + // gamescope spawned it as its own nested child when the source opened above. + spawned_now = true; None } Some(cmd) => match crate::library::launch_session_command(compositor, cmd) { - Ok(spawned) => Some(spawned), + Ok(spawned) => { + spawned_now = true; + Some(spawned) + } Err(e) => { tracing::warn!(command = %cmd, error = %e, "gamestream: could not launch app"); None @@ -321,6 +370,16 @@ fn run( }, None => None, }; + #[cfg(not(any(target_os = "windows", target_os = "linux")))] + let _ = adopt_launch; + // Settle the record against what actually happened (see the native plane). + if let Some(c) = launch_claim.as_ref() { + if spawned_now { + c.launched(); + } else if c.must_spawn() { + c.abandon(); + } + } // The launched game's lifetime, in both directions (design/session-game-lifetime.md) — the // compat plane's half of what the native plane already does: @@ -371,6 +430,9 @@ fn run( nested, child, launch_stamp, + // For an adopted launch this is the ORIGINAL launch's slot, so the record keeps + // tracking the same processes across the handover. + procs: launch_claim.as_ref().and_then(|c| c.procs()), }, on_exit, ); @@ -383,6 +445,9 @@ fn run( lease, life.quit.clone(), life.fingerprint.clone(), + // The record's hold moves in here — its drop opens the reconnect window the next + // `/launch` of this title is matched against. + launch_claim, ), ) }); diff --git a/crates/punktfunk-host/src/launchreg.rs b/crates/punktfunk-host/src/launchreg.rs new file mode 100644 index 00000000..4dd5fb49 --- /dev/null +++ b/crates/punktfunk-host/src/launchreg.rs @@ -0,0 +1,652 @@ +//! What this host launched, for whom, and when — the record a *second* session needs in order not to +//! launch the same title twice (design/session-game-lifetime.md). +//! +//! A client that re-dials mid-session re-sends its `Hello::launch` **verbatim**. It cannot drop the +//! field: on Linux the per-session gamescope is re-adopted through pf-vdisplay's display registry, +//! whose reuse key includes the launch command, so a retry without it orphans the running game. The +//! host therefore has to be the one that notices, and two things go wrong when it doesn't: +//! +//! * **the title is launched twice.** `steam://rungameid`, Epic's launcher URI and an AUMID +//! activation all dedupe inside the launcher — it focuses the copy that is already up — but a +//! `gog:` or `custom:` target is a plain spawn and really does start a second copy of the game. +//! * **the reconnected session never notices the game exit.** A fresh session mints a fresh +//! [`crate::gamelease::launch_clock`] stamp, and [`crate::procscan`] refuses to adopt any process +//! that started more than [`crate::procscan::START_SLACK_SECS`] before it. The game was started by +//! the *original* session, minutes earlier, so it can never be adopted — and the reconnected +//! session has no game-exit detection for the rest of its life. +//! +//! ### Why a registry, and not "is this title already running?" +//! +//! Because that second question has a catastrophic answer. [`crate::procscan`]'s first rule is that a +//! process predating the launch is never adopted: a player may already have the game open when a +//! session starts, and treating that instance as "this session's game" would let a session end kill +//! something it never started — on Windows the host runs as SYSTEM and can signal anything. A +//! registry of the host's **own** launches preserves that rule *by construction*: a game the player +//! started for themselves was never recorded here, so it can never be reclaimed from here, and the +//! only reference instant a session can inherit is one this host took immediately before a spawn it +//! performed itself. +//! +//! The same care runs through the liveness probe: it only ever *re-verifies the processes a lease +//! actually adopted* ([`Liveness`]), never re-scans by [`crate::library::DetectSpec`]. A fresh scan +//! would find a copy the player started since, which is exactly the process rule 1 exists to keep out. +//! +//! ### Not the grace registry +//! +//! Deliberately separate from [`crate::gamelease::arm_grace`]. That one is **policy**: it exists only +//! under `GameOnSessionEnd::Always` and a non-deliberate end, so on the shipped default (`Keep`) it is +//! empty — which is precisely the configuration both defects above were reported on. This one is a +//! **record**: written whenever the host launches a title, whatever the operator's termination policy +//! says, and read at the next session's launch decision. Different owners, different lifetimes, and +//! no shared state between them. + +use std::sync::atomic::{AtomicU64, Ordering}; +use std::sync::{Arc, Mutex, OnceLock}; +use std::time::{Duration, Instant}; + +/// How long after its last session let go a launch is still treated as the *same* launch even though +/// nothing of it has ever been seen running. +/// +/// This window covers exactly one shape: a client that tears its session down and re-dials while the +/// launcher is still bringing the game up. The presenter's HEVC→H.264 codec fallback does that within +/// seconds; a client crash-and-restart within tens of them. Once the game *has* been seen, [`Liveness`] +/// answers the question exactly and this window stops mattering — and a launch whose processes are +/// confirmed gone is re-launchable immediately, whatever the window says. +/// +/// Kept short on purpose. The cost of it being too long is a title the player asked for and did not +/// get, which is a far worse failure than the second copy it exists to prevent. +const IN_FLIGHT_WINDOW: Duration = Duration::from_secs(90); + +/// How long an unheld record survives at all, so the registry can't grow without bound across a long +/// host uptime. Generous: a launch idle this long whose game is somehow *still* running gets started +/// again, which is exactly what the host did before this module existed. +const MAX_RECORD_AGE: Duration = Duration::from_secs(24 * 60 * 60); + +/// The processes a launch's watcher adopted, published as it sees them so the record can still answer +/// "is *our* launch up?" after the session — and therefore the watcher — is gone. +/// +/// Written by [`crate::gamelease`]'s watch loop, read here. Only ever re-verified through +/// [`crate::procscan::Scanner::alive`], which re-checks each process's start time and so cannot be +/// fooled by a recycled pid (rule 2), and never re-scanned by spec (rule 1 — see the module docs). +pub type LiveProcs = Arc>>; + +/// What became of the processes a recorded launch adopted. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum Liveness { + /// At least one process this launch adopted is still the same live process. + Running, + /// Every process it adopted is gone. The launch is over. + Gone, + /// No opinion — nothing was ever adopted (the game has not appeared yet, or the title has no + /// detect signals and its only liveness signal was a child handle that died with its session), or + /// this platform has no process matcher at all (macOS, which has no launch path either). + /// + /// The no-signals case is worth naming: a title the host can only track through the child it + /// spawned is [`crate::gamelease::LeaseKind::Child`], and adopting it hands the new session a + /// lease with no child and no signals — [`crate::gamelease::LeaseKind::Untracked`], for which + /// both lifetime behaviors were already inert. So inside [`IN_FLIGHT_WINDOW`] such a reconnect + /// trades game-exit detection for not handing the player a second copy of the game. That is the + /// right way round: the missing detection is an annoyance, a second running copy is not. + Unknown, +} + +/// What a starting session must do about the title it was asked to launch. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum Plan { + /// Start it, and adopt its processes against the freshly minted reference instant. + Spawn, + /// This host already launched this title for this client and that launch is still ours: do **not** + /// start a second copy, and adopt against the **original** launch's reference instant so the + /// lease can still see (and therefore notice the exit of) the game that is already running. + Adopt, +} + +/// Who a launch belongs to. Both halves are required — see [`key_for`]. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct Key { + fingerprint: String, + game_id: String, +} + +/// The identity half of the match rule, pure and total: which (client, title) pair may ever reclaim a +/// launch. `None` = this launch is not recordable at all, so it is started exactly as it always was. +/// +/// Both sides must name a client **and** a title: +/// +/// * no fingerprint — an anonymous client (TOFU / `--open`, and the whole GameStream compat plane) — +/// because otherwise any unidentified client could reclaim any other unidentified client's launch, +/// and on Linux the second client would then get its own empty display with the first one's game +/// nowhere on it. This is the same conservatism [`crate::gamelease::readopt`] already applies. +/// * no library id — an operator-typed `apps.json` command, which has no library entry behind it — because +/// every such launch would otherwise share the one `None` id and reclaim each other. +/// +/// Both exclusions are the safe direction: the affected launches simply keep the pre-existing +/// behavior (start it again), rather than reclaiming something that might not be theirs. +pub fn key_for(fingerprint: Option<&str>, game_id: Option<&str>) -> Option { + Some(Key { + fingerprint: fingerprint?.to_string(), + game_id: game_id?.to_string(), + }) +} + +impl Key { + /// The fingerprint prefix the rest of the host logs clients by (`client_label`), so a launch line + /// can be lined up with the session lines around it without printing a full cert hash. + fn short_client(&self) -> &str { + self.fingerprint.get(..12).unwrap_or(&self.fingerprint) + } +} + +/// One launch this host performed, for one client, for one title. +struct Record { + key: Key, + /// The reference instant taken immediately before that launch ([`crate::gamelease::launch_clock`]). + /// This is the value a reconnecting session inherits; `None` only ever means "this platform has no + /// process-start clock", never "we failed to inherit one" — a session that inherits nothing gets + /// [`Plan::Spawn`] and its own fresh stamp instead. + stamp: Option, + /// The processes the launch's lease adopted; see [`LiveProcs`]. + procs: LiveProcs, + /// Set once the launch *actually happened*. A record made by a session that then failed to spawn + /// (or never had a launch path at all) is never matched — nothing is running for it to reclaim. + launched: bool, + /// How many live sessions hold this record. A count, not a flag: an old session's teardown and a + /// new session's launch decision overlap, and the old one releasing must never zero out the new + /// one's hold. + holders: u32, + /// When the last holder let go. `None` while held. + released_at: Option, + /// The newest claim taken on this record. An older session compares its own claim against this to + /// find out that its game now belongs to somebody else ([`Claim::superseded`]). + claim: u64, +} + +impl Record { + fn new(key: Key, stamp: Option, claim: u64) -> Self { + Self { + key, + stamp, + // A fresh slot per launch: the previous launch's dead processes must never be inherited + // by the new one, and its lease may still be writing into the old handle. + procs: Arc::new(Mutex::new(Vec::new())), + launched: false, + holders: 1, + released_at: None, + claim, + } + } +} + +/// **The match rule.** Does `rec` cover a new session's request to launch the title it is keyed on? +/// +/// Pure and total: the caller supplies the liveness verdict, the clock and the window, so the rule is +/// unit-testable without a live session, a process table or real time. Identity is not checked here — +/// it is the record's key, decided once by [`key_for`]. +/// +/// Liveness is authoritative wherever it has an opinion. Only when it has none do the two tie-breakers +/// apply, and they are the two shapes a reconnect actually takes: another session is holding the +/// launch right now (the teardown and the re-dial overlapped), or the client came back promptly while +/// the launcher was still working (the [`IN_FLIGHT_WINDOW`]). +fn covers(rec: &Record, live: Liveness, now: Instant, window: Duration) -> bool { + // A launch that never happened has nothing running to reclaim, and inheriting its reference + // instant would hand the new lease a floor with no game above it. + if !rec.launched { + return false; + } + match live { + // Processes this very launch adopted are still alive. This IS the game — start a second copy + // and the player gets two. + Liveness::Running => true, + // Every process it adopted is dead. Whatever the client is asking for now, it is not this + // launch — so a title that crashed on startup, or that the player quit, launches again at once. + Liveness::Gone => false, + Liveness::Unknown => { + rec.holders > 0 + || rec + .released_at + .is_some_and(|t| now.saturating_duration_since(t) <= window) + } + } +} + +/// Re-verify the processes this launch adopted. Never a fresh scan — see the module docs. +fn liveness(rec: &Record) -> Liveness { + let procs = rec.procs.lock().unwrap_or_else(|e| e.into_inner()); + if procs.is_empty() { + return Liveness::Unknown; + } + match alive_count(&procs) { + Some(0) => Liveness::Gone, + Some(_) => Liveness::Running, + None => Liveness::Unknown, + } +} + +/// How many of `procs` are still the same live processes. `None` on a platform with no matcher +/// (macOS), which is "no opinion" — never "gone". +fn alive_count(procs: &[crate::procscan::ProcRef]) -> Option { + #[cfg(any(target_os = "linux", windows))] + { + Some(crate::procscan::Scanner::system().alive(procs).len()) + } + #[cfg(not(any(target_os = "linux", windows)))] + { + let _ = procs; + None + } +} + +/// Forget records nothing can reclaim: an unheld launch that never happened, and an unheld one idle +/// past [`MAX_RECORD_AGE`]. Deliberately free of any process scan — it runs under the registry lock, +/// on the one path that touches the registry at all (a session deciding its launch). +fn sweep(recs: &mut Vec, now: Instant) { + recs.retain(|r| { + if r.holders > 0 { + return true; + } + let idle = r + .released_at + .map_or(Duration::ZERO, |t| now.saturating_duration_since(t)); + r.launched && idle < MAX_RECORD_AGE + }); +} + +struct Reg { + records: Mutex>, + next_claim: AtomicU64, +} + +fn reg() -> &'static Reg { + static REG: OnceLock = OnceLock::new(); + REG.get_or_init(|| Reg { + records: Mutex::new(Vec::new()), + // 0 is reserved for an unrecorded claim, which must never look current. + next_claim: AtomicU64::new(1), + }) +} + +/// Decide what this session must do about its launch, and claim the answer. +/// +/// `fresh_stamp` is this session's own [`crate::gamelease::launch_clock`] reading, taken before +/// anything spawns; it is used when the answer is [`Plan::Spawn`], and discarded in favour of the +/// recorded one when it is [`Plan::Adopt`]. +/// +/// The returned [`Claim`] is an RAII guard: hold it for the whole session (see +/// [`crate::gamelease::SessionGuard`]), because its drop is what starts the reconnect window. +pub fn claim(fingerprint: Option<&str>, game_id: Option<&str>, fresh_stamp: Option) -> Claim { + let Some(key) = key_for(fingerprint, game_id) else { + // Nothing to key a record on. Launch exactly as this host always has. + return Claim { + key: None, + id: 0, + plan: Plan::Spawn, + stamp: fresh_stamp, + procs: None, + }; + }; + let reg = reg(); + let now = Instant::now(); + let mut recs = reg.records.lock().unwrap_or_else(|e| e.into_inner()); + // Allocated **under the lock**, so claim ids and record writes agree on their order. An id handed + // out before the lock could be stamped onto a record after a higher one had been, and then neither + // session would see itself superseded. + let id = reg.next_claim.fetch_add(1, Ordering::Relaxed); + sweep(&mut recs, now); + + let procs = if let Some(i) = recs.iter().position(|r| r.key == key) { + let live = liveness(&recs[i]); + let rec = &mut recs[i]; + if covers(rec, live, now, IN_FLIGHT_WINDOW) { + rec.holders += 1; + rec.released_at = None; + rec.claim = id; + let (stamp, procs) = (rec.stamp, rec.procs.clone()); + drop(recs); + tracing::info!( + app = %key.game_id, + client = %key.short_client(), + ?live, + "this client's own launch of this title is still this host's — adopting it instead \ + of starting a second copy" + ); + return Claim { + key: Some(key), + id, + plan: Plan::Adopt, + stamp, + procs: Some(procs), + }; + } + // The previous launch of this title by this client is over (or never happened). Re-stamp the + // record for the launch about to happen: a fresh reference instant and a fresh process slot, + // so the dead launch's processes can never be inherited by the new one. + // + // Reset in place rather than replaced, because `holders` must survive: an older session may + // still be holding this record, and its release has to decrement the count it incremented. + rec.stamp = fresh_stamp; + rec.procs = Arc::new(Mutex::new(Vec::new())); + rec.launched = false; + rec.holders += 1; + rec.released_at = None; + rec.claim = id; + rec.procs.clone() + } else { + let rec = Record::new(key.clone(), fresh_stamp, id); + let procs = rec.procs.clone(); + recs.push(rec); + procs + }; + drop(recs); + tracing::debug!( + app = %key.game_id, + client = %key.short_client(), + "recording this host's launch of the title" + ); + Claim { + key: Some(key), + id, + plan: Plan::Spawn, + stamp: fresh_stamp, + procs: Some(procs), + } +} + +/// A session's hold on its launch record. Its drop starts the reconnect window. +pub struct Claim { + /// `None` for an unrecordable launch ([`key_for`]) — every method is then inert. + key: Option, + id: u64, + plan: Plan, + stamp: Option, + procs: Option, +} + +impl Claim { + /// Must this session actually start the title? + pub fn must_spawn(&self) -> bool { + matches!(self.plan, Plan::Spawn) + } + + /// The reference instant this session's lease must adopt against + /// ([`crate::gamelease::LeaseRequest::launch_stamp`]) — freshly taken for a [`Plan::Spawn`], + /// inherited from the original launch for a [`Plan::Adopt`]. + /// + /// `None` here always means what it means everywhere else in [`crate::procscan`]: this platform + /// has no process-start clock, so there is no start-time filter. It never means "inheritance + /// failed" — a session that finds nothing to inherit is given [`Plan::Spawn`] and its own fresh + /// reading instead. + pub fn stamp(&self) -> Option { + self.stamp + } + + /// The slot this launch's lease publishes its adopted processes into + /// ([`crate::gamelease::LeaseRequest::procs`]). For a [`Plan::Adopt`] it is the *original* + /// launch's slot, so the record keeps tracking the same processes across the handover. + pub fn procs(&self) -> Option { + self.procs.clone() + } + + /// The launch happened. Only a confirmed record is ever matched by a later session. + /// + /// Ignored once a newer session has re-stamped the record: what it would be confirming is that + /// session's launch, not this one's, and that session confirms its own. + pub fn launched(&self) { + self.with_record(|r| { + if r.claim == self.id { + r.launched = true; + } + }); + } + + /// The launch did not happen — it failed, or this platform has no launch path. Forget the record + /// entirely, so a retry starts the title rather than inheriting a launch that never occurred. + pub fn abandon(&self) { + let Some(key) = self.key.as_ref() else { + return; + }; + let mut recs = reg().records.lock().unwrap_or_else(|e| e.into_inner()); + recs.retain(|r| &r.key != key || r.claim != self.id); + } + + /// Has a **newer** session claimed this launch? `false` when there is no record at all, so only a + /// positive signal ever changes a caller's behavior. + pub fn superseded(&self) -> bool { + let Some(key) = self.key.as_ref() else { + return false; + }; + let recs = reg().records.lock().unwrap_or_else(|e| e.into_inner()); + recs.iter().any(|r| &r.key == key && r.claim > self.id) + } + + /// Run `f` against this claim's record, if it still exists. Deliberately **not** claim-checked: + /// the release in [`Drop`] must decrement the very count it incremented, even after a newer + /// session re-stamped the record. Callers that need "only if it is still mine" check `claim` + /// themselves ([`Claim::launched`]). + fn with_record(&self, f: impl FnOnce(&mut Record)) { + let Some(key) = self.key.as_ref() else { + return; + }; + let mut recs = reg().records.lock().unwrap_or_else(|e| e.into_inner()); + if let Some(r) = recs.iter_mut().find(|r| &r.key == key) { + f(r); + } + } +} + +impl Drop for Claim { + fn drop(&mut self) { + self.with_record(|r| { + r.holders = r.holders.saturating_sub(1); + if r.holders == 0 { + r.released_at = Some(Instant::now()); + } + }); + } +} + +#[cfg(test)] +mod tests { + use super::*; + + /// A record shaped for the pure-rule table below. Never touches the global registry. + fn rec(launched: bool, holders: u32, released_at: Option) -> Record { + Record { + key: Key { + fingerprint: "fp".into(), + game_id: "steam:1".into(), + }, + stamp: Some(1.0), + procs: Arc::new(Mutex::new(Vec::new())), + launched, + holders, + released_at, + claim: 1, + } + } + + /// Identity is the record's key, and a launch that can't be keyed is never reclaimed. + #[test] + fn a_launch_is_keyed_by_both_the_client_and_the_title() { + assert!(key_for(Some("fp"), Some("steam:570")).is_some()); + // An anonymous client, or a title with no library entry, is not recordable — the launch + // behaves exactly as it did before this module existed. + assert!(key_for(None, Some("steam:570")).is_none()); + assert!(key_for(Some("fp"), None).is_none()); + assert!(key_for(None, None).is_none()); + // Different client, or different title, is a different launch. + assert_ne!( + key_for(Some("a"), Some("steam:570")), + key_for(Some("b"), Some("steam:570")) + ); + assert_ne!( + key_for(Some("a"), Some("steam:570")), + key_for(Some("a"), Some("gog:1")) + ); + } + + /// The match rule itself: liveness first, then the two tie-breakers. + #[test] + fn the_match_rule_puts_liveness_ahead_of_the_window() { + let t0 = Instant::now(); + let window = Duration::from_secs(90); + let inside = t0 + Duration::from_secs(30); + let outside = t0 + Duration::from_secs(600); + + // A launch that never happened is never reclaimed, however alive something looks. + let never = rec(false, 1, None); + assert!(!covers(&never, Liveness::Running, inside, window)); + + // Our own processes are still up: reclaim it, no matter how long ago the session let go. + let old = rec(true, 0, Some(t0)); + assert!(covers(&old, Liveness::Running, outside, window)); + + // Confirmed gone beats everything — including a live holder and a fresh release. This is what + // keeps a title that crashed on startup (or that the player quit) launchable at once. + let held = rec(true, 1, None); + assert!(!covers(&held, Liveness::Gone, inside, window)); + assert!(!covers(&old, Liveness::Gone, inside, window)); + + // Nothing seen yet: a live holder is itself the answer (the teardown and the re-dial + // overlapped), and a prompt return is the same launch. + assert!(covers(&held, Liveness::Unknown, inside, window)); + assert!(covers(&old, Liveness::Unknown, inside, window)); + // ...but a return long after the window, with nothing ever seen running, is a new launch. + assert!(!covers(&old, Liveness::Unknown, outside, window)); + } + + /// The sweep drops what nobody can reclaim and keeps what somebody can. + #[test] + fn the_sweep_keeps_only_reclaimable_records() { + let t0 = Instant::now(); + let mut recs = vec![ + rec(false, 0, Some(t0)), // never launched, nobody holding + rec(true, 0, Some(t0)), // launched, recently released + rec(false, 1, None), // never launched but HELD — its session is still deciding + ]; + sweep(&mut recs, t0 + Duration::from_secs(1)); + assert_eq!(recs.len(), 2); + // ...and an ancient one goes too. + let mut recs = vec![rec(true, 0, Some(t0))]; + sweep(&mut recs, t0 + MAX_RECORD_AGE + Duration::from_secs(1)); + assert!(recs.is_empty()); + } + + /// **Defect A.** A client that re-dials and re-sends `Hello::launch` verbatim must not get a + /// second copy of its game. + /// + /// Before this module the host launched unconditionally at + /// `native/stream.rs`'s launch site — i.e. the second decision here was always "spawn". + #[test] + fn a_reconnect_does_not_launch_the_title_a_second_time() { + let (fp, app) = (Some("fp-double"), Some("gog:double")); + let first = claim(fp, app, Some(100.0)); + assert!(first.must_spawn(), "the first session starts the title"); + first.launched(); + drop(first); // the session ends; the reconnect window opens + + let second = claim(fp, app, Some(900.0)); + assert!( + !second.must_spawn(), + "a reconnect inside the window must adopt the running launch, not start a second copy" + ); + second.abandon(); // leave the process-global registry as we found it + } + + /// **Defect B.** The reconnected session must adopt against the ORIGINAL launch's reference + /// instant, or [`crate::procscan`] rejects the game — started minutes before this session — and + /// the session has no game-exit detection for the rest of its life. + #[test] + fn a_reconnect_inherits_the_original_launchs_reference_instant() { + let (fp, app) = (Some("fp-stamp"), Some("steam:stamp")); + let first = claim(fp, app, Some(100.0)); + assert_eq!(first.stamp(), Some(100.0)); + first.launched(); + drop(first); + + // The new session mints its own (much later) reading and passes it in; the record's wins. + let second = claim(fp, app, Some(900.0)); + assert_eq!( + second.stamp(), + Some(100.0), + "the reconnect must adopt against the original launch, not against its own start" + ); + // Spelled out, because this is exactly what the host did before: a fresh reading here is a + // floor minutes above the running game's start time, and `procscan` rejects everything under + // it — the reconnected session then has no game-exit detection for the rest of its life. + assert_ne!(second.stamp(), Some(900.0)); + // Both sessions publish into the SAME slot, so the record keeps tracking the same processes + // across the handover. + assert!(second.procs().is_some()); + second.abandon(); + } + + /// Inheriting nothing must never turn into "adopt anything": wherever a launch has a reference + /// instant, every decision made from it has one too. + #[test] + fn a_decision_never_downgrades_a_reference_instant_to_no_filter() { + let (fp, app) = (Some("fp-filter"), Some("steam:filter")); + let first = claim(fp, app, Some(42.0)); + assert!(first.stamp().is_some()); + first.launched(); + drop(first); + let second = claim(fp, app, Some(99.0)); + assert!( + second.stamp().is_some(), + "a reconnect must never end up with the start-time filter disabled" + ); + second.abandon(); + // A launch that cannot be recorded still carries this session's own fresh reading through. + let anon = claim(None, app, Some(7.0)); + assert!(anon.must_spawn()); + assert_eq!(anon.stamp(), Some(7.0)); + assert!(anon.procs().is_none()); + } + + /// A launch that failed (or a platform with no launch path) leaves nothing behind: the next + /// attempt starts the title and gets its own reference instant. + #[test] + fn an_abandoned_launch_is_never_reclaimed() { + let (fp, app) = (Some("fp-fail"), Some("custom:fail")); + let first = claim(fp, app, Some(100.0)); + assert!(first.must_spawn()); + first.abandon(); // the spawn failed + drop(first); + + let second = claim(fp, app, Some(900.0)); + assert!(second.must_spawn(), "a failed launch must be retried"); + assert_eq!(second.stamp(), Some(900.0)); + second.abandon(); + } + + /// A confirmed launch that was never released is still adopted by an overlapping second session — + /// the order where the new session's handshake beats the old session's teardown. + #[test] + fn an_overlapping_session_adopts_a_still_held_launch() { + let (fp, app) = (Some("fp-overlap"), Some("steam:overlap")); + let old = claim(fp, app, Some(100.0)); + old.launched(); + // The old session has NOT torn down yet. + let new = claim(fp, app, Some(900.0)); + assert!(!new.must_spawn(), "a held launch is still ours"); + assert_eq!(new.stamp(), Some(100.0)); + // ...and the old session can see that its game now belongs to the new one, so its teardown + // policy leaves it alone. + assert!(old.superseded()); + assert!(!new.superseded()); + drop(old); + assert!( + !new.superseded(), + "the older session releasing must not look like a newer claim" + ); + new.abandon(); + } + + /// An unrecordable launch is inert in every direction. + #[test] + fn an_unrecordable_launch_never_supersedes_anything() { + let anon = claim(None, None, None); + assert!(anon.must_spawn()); + assert!(!anon.superseded()); + anon.launched(); // no-op + anon.abandon(); // no-op + } +} diff --git a/crates/punktfunk-host/src/main.rs b/crates/punktfunk-host/src/main.rs index 801a1f1e..9c6b8484 100644 --- a/crates/punktfunk-host/src/main.rs +++ b/crates/punktfunk-host/src/main.rs @@ -71,6 +71,10 @@ mod install; #[cfg(target_os = "windows")] #[path = "windows/interactive.rs"] mod interactive; +// What this host launched, for whom, and when — so a client that re-dials and re-sends its +// `Hello::launch` verbatim neither gets a second copy of its game nor loses sight of the one it has +// (design/session-game-lifetime.md). +mod launchreg; mod library; mod log_capture; mod mgmt; diff --git a/crates/punktfunk-host/src/native.rs b/crates/punktfunk-host/src/native.rs index e49b094d..e40077bd 100644 --- a/crates/punktfunk-host/src/native.rs +++ b/crates/punktfunk-host/src/native.rs @@ -1486,9 +1486,17 @@ async fn serve_session( // A client reconnecting inside its game's reconnect window takes the game back: nothing is ended, // and this session adopts it. Matched on (this client, this title) so it can only ever reclaim its // own game. + // + // Cancelling the pending termination is all this does — the *game* is re-adopted in the data plane + // through `crate::launchreg`, which is what carries the original launch's reference instant across + // sessions (a reprieved lease can't: its watcher is cancelled and its exit action closes a + // connection that is already gone). Both are needed, and neither subsumes the other: this one + // exists only under `GameOnSessionEnd::Always`, the record exists whatever the policy says. if let Some(target) = launch_target.as_ref() { let fp = punktfunk_core::quic::endpoint::peer_fingerprint(&conn).map(hex::encode); - crate::gamelease::readopt(fp.as_deref(), target.game.id.as_deref()); + // The reprieved leases are deliberately dropped: they are corpses (see `readopt`), and this + // plane has nothing to say about them that `readopt` has not already logged per lease. + let _reprieved = crate::gamelease::readopt(fp.as_deref(), target.game.id.as_deref()); } // Per-title prep steps (RFC §6) for a launched CUSTOM library title: run synchronously // before the data plane starts (so before the display opens and the title spawns); the diff --git a/crates/punktfunk-host/src/native/stream.rs b/crates/punktfunk-host/src/native/stream.rs index a26caee4..e6eaa227 100644 --- a/crates/punktfunk-host/src/native/stream.rs +++ b/crates/punktfunk-host/src/native/stream.rs @@ -1450,7 +1450,26 @@ pub(super) fn virtual_stream(ctx: SessionContext, prepared: Option { + tracing::info!( + command = %cmd, + "this client's copy of this title is already running from an earlier session — not \ + starting a second one" + ); + None + } Some(cmd) if crate::vdisplay::launch_is_nested(compositor, gamescope_route.as_ref()) => { tracing::info!(command = %cmd, "launch nested into the per-session gamescope"); + // gamescope spawns it as its own nested child, so the launch DID happen here. + spawned_now = true; None } Some(cmd) => match crate::library::launch_session_command(compositor, cmd) { - Ok(spawned) => Some(spawned), + Ok(spawned) => { + spawned_now = true; + Some(spawned) + } Err(e) => { tracing::warn!(command = %cmd, error = %e, "could not launch requested title into the session"); None @@ -1654,7 +1707,17 @@ pub(super) fn virtual_stream(ctx: SessionContext, prepared: Option None, }; #[cfg(not(any(target_os = "windows", target_os = "linux")))] - let _ = &launch; + let _ = (&launch, adopt_launch); + // Settle the record against what actually happened. A spawn that never ran — it failed, or this + // platform has no launch path — must leave nothing behind, or a retry would inherit a launch that + // never occurred and then decline to start the title at all. + if let Some(c) = launch_claim.as_ref() { + if spawned_now { + c.launched(); + } else if c.must_spawn() { + c.abandon(); + } + } // The launched game's lifetime, in both directions (design/session-game-lifetime.md): // @@ -1712,6 +1775,9 @@ pub(super) fn virtual_stream(ctx: SessionContext, prepared: Option