feat(game): cancellable detector service

Step 4 of game detection. One std-thread worker owns the SteamProbe cache + Debouncer across ticks, polls the OS adapters every 3s off the async runtime, and publishes the stable detected game on a tokio watch channel only when it changes. Manual override + process map are live-updatable via shared handles; a cancellable sleep honors stop promptly; drop stops it. The per-tick decision (match + resolve + debounce) is the pure poll_once, unit-tested with synthetic Steam/process inputs (debounce, process-only match, immediate manual override). +4 tests (397 lib). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-06-21 15:21:52 -04:00
parent e31d3db986
commit 7d44808a5e
2 changed files with 235 additions and 0 deletions
@@ -0,0 +1,234 @@
+//! The detector service (§5): one cancellable background worker that polls the OS
+//! adapters, runs the pure matcher + debouncer, and publishes the stable detected
+//! game on a watch channel — only when it changes, so a flapping detector can't
+//! spam `PeerState` re-announces.
+//!
+//! All the OS reads (Steam files / registry, the process scan) are blocking, so
+//! the worker is a dedicated `std::thread`, not a tokio task; it owns the
+//! [`SteamProbe`] cache and the [`Debouncer`] across ticks. The per-tick decision
+//! is factored into the pure [`poll_once`] so the wiring of resolve + match +
+//! debounce is unit-tested without any I/O.
+
+use super::{
+    builtin_denylist, match_processes, resolve, Debouncer, DetectedGame, ManualOverride,
+};
+use super::scan;
+use super::steam::SteamProbe;
+use std::collections::BTreeMap;
+use std::sync::atomic::{AtomicBool, Ordering};
+use std::sync::{Arc, Mutex};
+use std::time::Duration;
+use tokio::sync::watch;
+
+/// How often the detector samples Steam state + the process list.
+pub const POLL_INTERVAL: Duration = Duration::from_secs(3);
+/// Granularity of the cancellable sleep between polls, so a stop request is
+/// honored promptly instead of after a full [`POLL_INTERVAL`].
+const SLEEP_TICK: Duration = Duration::from_millis(200);
+
+/// Apply one poll's worth of inputs to the debouncer, returning the new published
+/// value **iff it changed** (the signal to re-announce presence / switch the
+/// background). Pure: the caller supplies the already-fetched Steam detection and
+/// process list, so resolve + match + debounce are testable with zero I/O.
+pub fn poll_once(
+    debouncer: &mut Debouncer,
+    override_: &ManualOverride,
+    steam: Option<DetectedGame>,
+    processes: &[String],
+    process_map: &BTreeMap<String, String>,
+    denylist: &std::collections::BTreeSet<&str>,
+) -> Option<Option<DetectedGame>> {
+    let matched = match_processes(processes, process_map, denylist);
+    let res = resolve(override_, steam, &matched);
+    if debouncer.observe(res.game, res.immediate) {
+        Some(debouncer.current().cloned())
+    } else {
+        None
+    }
+}
+
+/// Shared, live-updatable inputs to the detector, written by core (manual override
+/// changes, config edits to the process map) and read each poll by the worker.
+#[derive(Default)]
+pub struct DetectorInputs {
+    pub override_: Mutex<ManualOverride>,
+    pub process_map: Mutex<BTreeMap<String, String>>,
+}
+
+/// A running detector service. Holds the watch receiver for detected-game changes
+/// and the shared inputs; dropping it (or calling [`stop`](Self::stop)) ends the
+/// worker thread.
+pub struct GameDetector {
+    inputs: Arc<DetectorInputs>,
+    rx: watch::Receiver<Option<DetectedGame>>,
+    stop: Arc<AtomicBool>,
+}
+
+impl GameDetector {
+    /// Spawn the detector worker. `process_map` seeds the non-Steam mappings;
+    /// `override_` seeds the manual override (usually `Auto`). The worker runs
+    /// until [`stop`](Self::stop) or the returned `GameDetector` is dropped.
+    pub fn spawn(override_: ManualOverride, process_map: BTreeMap<String, String>) -> Self {
+        let inputs = Arc::new(DetectorInputs {
+            override_: Mutex::new(override_),
+            process_map: Mutex::new(process_map),
+        });
+        let (tx, rx) = watch::channel(None);
+        let stop = Arc::new(AtomicBool::new(false));
+
+        let worker_inputs = inputs.clone();
+        let worker_stop = stop.clone();
+        std::thread::Builder::new()
+            .name("game-detector".to_string())
+            .spawn(move || worker_loop(worker_inputs, tx, worker_stop))
+            .ok();
+
+        Self { inputs, rx, stop }
+    }
+
+    /// A clone of the watch receiver for detected-game changes. The current value
+    /// is `None` until the first non-empty detection is debounced in.
+    pub fn subscribe(&self) -> watch::Receiver<Option<DetectedGame>> {
+        self.rx.clone()
+    }
+
+    /// Replace the manual override (applied on the next poll, immediately,
+    /// bypassing debounce).
+    pub fn set_override(&self, override_: ManualOverride) {
+        *self.inputs.override_.lock().unwrap() = override_;
+    }
+
+    /// Replace the user process→name mappings (e.g. after a Settings edit).
+    pub fn set_process_map(&self, map: BTreeMap<String, String>) {
+        *self.inputs.process_map.lock().unwrap() = map;
+    }
+
+    /// Signal the worker to exit. Idempotent; also happens on drop.
+    pub fn stop(&self) {
+        self.stop.store(true, Ordering::Relaxed);
+    }
+}
+
+impl Drop for GameDetector {
+    fn drop(&mut self) {
+        self.stop();
+    }
+}
+
+/// The blocking worker loop: probe, decide, publish on change, sleep (cancellably).
+fn worker_loop(
+    inputs: Arc<DetectorInputs>,
+    tx: watch::Sender<Option<DetectedGame>>,
+    stop: Arc<AtomicBool>,
+) {
+    let denylist = builtin_denylist();
+    let mut steam = SteamProbe::new();
+    let mut debouncer = Debouncer::default();
+
+    while !stop.load(Ordering::Relaxed) {
+        let override_ = inputs.override_.lock().unwrap().clone();
+        let process_map = inputs.process_map.lock().unwrap().clone();
+
+        let steam_game = steam.detect();
+        let processes = scan::running_executables();
+
+        if let Some(new_current) =
+            poll_once(&mut debouncer, &override_, steam_game, &processes, &process_map, &denylist)
+        {
+            // A closed receiver means core shut down; stop quietly.
+            if tx.send(new_current).is_err() {
+                return;
+            }
+        }
+
+        // Cancellable sleep: wake promptly on a stop request.
+        let mut slept = Duration::ZERO;
+        while slept < POLL_INTERVAL && !stop.load(Ordering::Relaxed) {
+            std::thread::sleep(SLEEP_TICK);
+            slept += SLEEP_TICK;
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::super::GameSource;
+    use super::*;
+
+    fn game(id: &str, name: &str, source: GameSource) -> DetectedGame {
+        DetectedGame { id: id.into(), name: Some(name.into()), source }
+    }
+
+    fn map(pairs: &[(&str, &str)]) -> BTreeMap<String, String> {
+        pairs.iter().map(|(k, v)| (k.to_string(), v.to_string())).collect()
+    }
+
+    #[test]
+    fn poll_once_debounces_steam_detection() {
+        let deny = builtin_denylist();
+        let mut d = Debouncer::default();
+        let steam = game("steam:730", "CS2", GameSource::Steam);
+        let empty = BTreeMap::new();
+
+        // First poll: detected but not yet published (needs two hits).
+        assert_eq!(
+            poll_once(&mut d, &ManualOverride::Auto, Some(steam.clone()), &[], &empty, &deny),
+            None
+        );
+        // Second poll: published.
+        assert_eq!(
+            poll_once(&mut d, &ManualOverride::Auto, Some(steam.clone()), &[], &empty, &deny),
+            Some(Some(steam))
+        );
+        // Third identical poll: no change event.
+        assert_eq!(
+            poll_once(&mut d, &ManualOverride::Auto, Some(game("steam:730", "CS2", GameSource::Steam)), &[], &empty, &deny),
+            None
+        );
+    }
+
+    #[test]
+    fn poll_once_matches_process_when_no_steam() {
+        let deny = builtin_denylist();
+        let mut d = Debouncer::default();
+        let procs = vec!["/games/hl2_linux".to_string()];
+        let user = map(&[("hl2_linux", "Half-Life 2")]);
+
+        poll_once(&mut d, &ManualOverride::Auto, None, &procs, &user, &deny);
+        let change = poll_once(&mut d, &ManualOverride::Auto, None, &procs, &user, &deny);
+        let published = change.expect("should publish on second hit").expect("a game");
+        assert_eq!(published.id, "exe:hl2_linux");
+        assert_eq!(published.name.as_deref(), Some("Half-Life 2"));
+    }
+
+    #[test]
+    fn poll_once_manual_override_is_immediate() {
+        let deny = builtin_denylist();
+        let mut d = Debouncer::default();
+        let forced = game("steam:220", "HL2", GameSource::Steam);
+        // Even with a live Steam detection of something else, the override wins now.
+        let other = game("steam:730", "CS2", GameSource::Steam);
+        let change = poll_once(
+            &mut d,
+            &ManualOverride::Force(forced.clone()),
+            Some(other),
+            &[],
+            &BTreeMap::new(),
+            &deny,
+        );
+        assert_eq!(change, Some(Some(forced)));
+    }
+
+    #[test]
+    fn spawn_and_stop_is_clean() {
+        // Smoke test the lifecycle: spawning and stopping must not panic, and the
+        // initial published value is None.
+        let det = GameDetector::spawn(ManualOverride::Auto, BTreeMap::new());
+        assert_eq!(*det.subscribe().borrow(), None);
+        det.set_override(ManualOverride::ForceNone);
+        det.set_process_map(map(&[("x", "X")]));
+        det.stop();
+        // Dropping also stops; no hang/panic.
+        drop(det);
+    }
+}
@@ -12,6 +12,7 @@
 //! ([`scan`]) — feed already-parsed values into these pure functions, and the
 //! cancellable poll service ([`detector`]) wires them together.

+pub mod detector;
 pub mod scan;
 pub mod steam;
 pub mod vdf;