fix(peer): bound local monitor snapshots

Filter roots through the catalog before retaining them, cap entries, games, and poll time, and preserve the previous complete snapshot on failure. Repeated failures back off exponentially and a root change resets retry state.

Test Plan:
- just test
- just clippy
- focused catalog-filter, budget-preservation, and backoff tests
- git diff --check

crates/lanspread-peer/src/services/local_monitor.rs

@@ -13,6 +13,8 @@ use std::{
 };
 
 use futures::FutureExt;
+use lanspread_db::{content_manifest::MAX_CATALOG_ENTRIES, db::GameCatalog};
+use lanspread_proto::MAX_LIBRARY_GAMES;
 use tokio::{
     sync::RwLock,
     task::{JoinError, JoinSet},
@@ -66,11 +68,98 @@ struct RescanGate {
     pending: Arc<RwLock<HashSet<String>>>,
 }
 
+const MAX_POLL_SNAPSHOT_ENTRIES: usize = MAX_CATALOG_ENTRIES;
+const MAX_POLL_SNAPSHOT_DURATION: std::time::Duration = std::time::Duration::from_millis(500);
+const MAX_POLL_BACKOFF: std::time::Duration = std::time::Duration::from_mins(5);
+
+#[derive(Clone, Copy)]
+struct PollLimits {
+    entry_cap: usize,
+    game_ceiling: usize,
+    time_budget: std::time::Duration,
+}
+
+impl PollLimits {
+    const PRODUCTION: Self = Self {
+        entry_cap: MAX_POLL_SNAPSHOT_ENTRIES,
+        game_ceiling: MAX_LIBRARY_GAMES,
+        time_budget: MAX_POLL_SNAPSHOT_DURATION,
+    };
+}
+
+struct PollBudget {
+    limits: PollLimits,
+    entries: usize,
+    started: std::time::Instant,
+}
+
+impl PollBudget {
+    fn new(limits: PollLimits) -> Self {
+        Self {
+            limits,
+            entries: 0,
+            started: std::time::Instant::now(),
+        }
+    }
+
+    fn charge(&mut self) -> io::Result<()> {
+        self.entries = self.entries.checked_add(1).ok_or_else(|| {
+            io::Error::new(io::ErrorKind::InvalidData, "poll entry count overflow")
+        })?;
+        if self.entries > self.limits.entry_cap {
+            return Err(io::Error::new(
+                io::ErrorKind::InvalidData,
+                format!(
+                    "local game poll exceeds the {}-entry snapshot budget",
+                    self.limits.entry_cap
+                ),
+            ));
+        }
+        if self.started.elapsed() >= self.limits.time_budget {
+            return Err(io::Error::new(
+                io::ErrorKind::TimedOut,
+                format!(
+                    "local game poll exceeds the {:?} snapshot-time budget",
+                    self.limits.time_budget
+                ),
+            ));
+        }
+        Ok(())
+    }
+}
+
+#[derive(Default)]
+struct PollBackoff {
+    failures: u32,
+    retry_at: Option<Instant>,
+}
+
+impl PollBackoff {
+    fn should_attempt(&self, now: Instant) -> bool {
+        self.retry_at.is_none_or(|retry_at| now >= retry_at)
+    }
+
+    fn record_failure(&mut self, now: Instant) -> std::time::Duration {
+        self.failures = self.failures.saturating_add(1);
+        let exponent = self.failures.saturating_sub(1).min(9);
+        let seconds = 1_u64 << exponent;
+        let delay = std::time::Duration::from_secs(seconds).min(MAX_POLL_BACKOFF);
+        self.retry_at = Some(now + delay);
+        delay
+    }
+
+    fn record_success(&mut self) {
+        self.failures = 0;
+        self.retry_at = None;
+    }
+}
+
 /// Monitors the local game directory for changes.
 pub async fn run_local_game_monitor(tx_notify_ui: PeerEventSender, ctx: Ctx) -> eyre::Result<()> {
     log::info!("Starting polling-based local game directory monitor");
 
     let mut snapshot = initial_poll_snapshot(&ctx).await;
+    let mut poll_backoff = PollBackoff::default();
     let gate = RescanGate::default();
     let mut rescans = JoinSet::new();
     let now = Instant::now();
@@ -98,6 +187,7 @@ pub async fn run_local_game_monitor(tx_notify_ui: PeerEventSender, ctx: Ctx) ->
                         &gate,
                         &mut rescans,
                         &mut snapshot,
+                        &mut poll_backoff,
                     ).await;
                 }
                 _ = fallback_interval.tick() => {
@@ -172,11 +262,29 @@ async fn poll_local_game_changes(
     gate: &RescanGate,
     rescans: &mut JoinSet<()>,
     previous: &mut Option<PollSnapshot>,
+    backoff: &mut PollBackoff,
 ) {
+    let configured_game_dir = ctx.game_dir.read().await;
+    if previous
+        .as_ref()
+        .is_some_and(|snapshot| snapshot.game_dir != *configured_game_dir)
+    {
+        backoff.record_success();
+    }
+    drop(configured_game_dir);
+
+    let now = Instant::now();
+    if !backoff.should_attempt(now) {
+        return;
+    }
     let current = match capture_poll_snapshot(ctx).await {
-        Ok(snapshot) => snapshot,
+        Ok(snapshot) => {
+            backoff.record_success();
+            snapshot
+        }
         Err(error) => {
-            log::warn!("Failed to poll local game directory: {error}");
+            let delay = backoff.record_failure(Instant::now());
+            log::warn!("Failed to poll local game directory: {error}; retrying after {delay:?}");
             return;
         }
     };
@@ -191,11 +299,24 @@ async fn capture_poll_snapshot(ctx: &Ctx) -> io::Result<PollSnapshot> {
     // to this task and is complete before the admission guard is released.
     let _admission = ctx.operation_admission.lock().await;
     let game_dir = ctx.game_dir.read().await.clone();
-    scoped_blocking(|| snapshot_game_directory(&game_dir))
+    let catalog = ctx.catalog.catalog();
+    scoped_blocking(|| snapshot_game_directory(&game_dir, Some(catalog)))
 }
 
-fn snapshot_game_directory(game_dir: &Path) -> io::Result<PollSnapshot> {
+fn snapshot_game_directory(
+    game_dir: &Path,
+    catalog: Option<&GameCatalog>,
+) -> io::Result<PollSnapshot> {
+    snapshot_game_directory_with_limits(game_dir, catalog, PollLimits::PRODUCTION)
+}
+
+fn snapshot_game_directory_with_limits(
+    game_dir: &Path,
+    catalog: Option<&GameCatalog>,
+    limits: PollLimits,
+) -> io::Result<PollSnapshot> {
     let mut games = BTreeMap::new();
+    let mut budget = PollBudget::new(limits);
     let entries = match fs::read_dir(game_dir) {
         Ok(entries) => entries,
         Err(error) if error.kind() == io::ErrorKind::NotFound => {
@@ -209,17 +330,26 @@ fn snapshot_game_directory(game_dir: &Path) -> io::Result<PollSnapshot> {
 
     for entry in entries {
         let entry = entry?;
+        budget.charge()?;
         let name = entry.file_name();
         let Some(id) = name.to_str() else {
             continue;
         };
-        if is_ignored_games_root_name(id) {
+        if is_ignored_games_root_name(id) || catalog.is_some_and(|catalog| !catalog.contains(id)) {
             continue;
         }
 
-        let game_root = match snapshot_game_root(&entry.path()) {
+        let game_root = match snapshot_game_root(&entry.path(), &mut budget) {
             Ok(Some(snapshot)) => snapshot,
             Ok(None) => continue,
+            Err(error)
+                if matches!(
+                    error.kind(),
+                    io::ErrorKind::InvalidData | io::ErrorKind::TimedOut
+                ) =>
+            {
+                return Err(error);
+            }
             Err(error) => {
                 log::debug!(
                     "Could not snapshot local game root {}: {error}",
@@ -229,6 +359,15 @@ fn snapshot_game_directory(game_dir: &Path) -> io::Result<PollSnapshot> {
             }
         };
         games.insert(id.to_owned(), game_root);
+        if games.len() > limits.game_ceiling {
+            return Err(io::Error::new(
+                io::ErrorKind::InvalidData,
+                format!(
+                    "local game poll exceeds the {}-game snapshot budget",
+                    limits.game_ceiling
+                ),
+            ));
+        }
     }
 
     Ok(PollSnapshot {
@@ -237,7 +376,10 @@ fn snapshot_game_directory(game_dir: &Path) -> io::Result<PollSnapshot> {
     })
 }
 
-fn snapshot_game_root(game_root: &Path) -> io::Result<Option<GameRootSnapshot>> {
+fn snapshot_game_root(
+    game_root: &Path,
+    budget: &mut PollBudget,
+) -> io::Result<Option<GameRootSnapshot>> {
     let root_fingerprint = match fingerprint_entry(game_root) {
         Ok(fingerprint) => fingerprint,
         Err(error) if error.kind() == io::ErrorKind::NotFound => return Ok(None),
@@ -256,6 +398,7 @@ fn snapshot_game_root(game_root: &Path) -> io::Result<Option<GameRootSnapshot>>
 
     for entry in entries {
         let entry = entry?;
+        budget.charge()?;
         let name = entry.file_name();
         if name.to_str().is_some_and(is_download_protected_root_name) {
             continue;
@@ -497,7 +640,7 @@ mod tests {
     }
 
     fn snapshot(game_dir: &Path) -> PollSnapshot {
-        snapshot_game_directory(game_dir).expect("poll snapshot should succeed")
+        snapshot_game_directory(game_dir, None).expect("poll snapshot should succeed")
     }
 
     async fn injected_monitor_loop_panic() -> eyre::Result<()> {
@@ -532,6 +675,64 @@ mod tests {
         );
     }
 
+    #[test]
+    fn poll_budget_failure_keeps_previous_snapshot_and_filters_non_catalog_roots() {
+        let temp = TempDir::new("lanspread-local-monitor-budget");
+        write_file(&temp.path().join("game/version.ini"), b"20250101");
+        write_file(&temp.path().join("unknown/deep/payload.bin"), b"ignored");
+        let catalog = GameCatalog::from_ids(["game".to_string()]);
+        let initial = snapshot_game_directory(temp.path(), Some(&catalog))
+            .expect("catalog-filtered snapshot should succeed");
+        assert_eq!(
+            initial.games.keys().map(String::as_str).collect::<Vec<_>>(),
+            vec!["game"]
+        );
+
+        let previous = Some(initial);
+        let error = snapshot_game_directory_with_limits(
+            temp.path(),
+            Some(&catalog),
+            PollLimits {
+                entry_cap: 1,
+                game_ceiling: MAX_LIBRARY_GAMES,
+                time_budget: MAX_POLL_SNAPSHOT_DURATION,
+            },
+        )
+        .expect_err("one root plus its fingerprint must exceed a one-entry budget");
+
+        assert!(error.to_string().contains("entry snapshot budget"));
+        assert_eq!(
+            previous
+                .as_ref()
+                .expect("previous snapshot should remain available")
+                .games
+                .keys()
+                .map(String::as_str)
+                .collect::<Vec<_>>(),
+            vec!["game"]
+        );
+    }
+
+    #[test]
+    fn poll_failures_back_off_exponentially_and_success_resets() {
+        let now = Instant::now();
+        let mut backoff = PollBackoff::default();
+        assert!(backoff.should_attempt(now));
+
+        let first = backoff.record_failure(now);
+        assert_eq!(first, std::time::Duration::from_secs(1));
+        assert!(!backoff.should_attempt(now));
+        assert!(backoff.should_attempt(now + first));
+
+        let second = backoff.record_failure(now + first);
+        assert_eq!(second, std::time::Duration::from_secs(2));
+        assert!(!backoff.should_attempt(now + first));
+
+        backoff.record_success();
+        assert!(backoff.should_attempt(now));
+        assert_eq!(backoff.failures, 0);
+    }
+
     #[test]
     fn snapshot_diff_detects_game_change_and_disappearance() {
         let temp = TempDir::new("lanspread-local-monitor-change");
@@ -634,7 +835,15 @@ mod tests {
         );
 
         write_file(&temp.path().join("game/version.ini"), b"20250101");
-        poll_local_game_changes(&ctx, &tx, &rescan_gate, &mut rescans, &mut state).await;
+        poll_local_game_changes(
+            &ctx,
+            &tx,
+            &rescan_gate,
+            &mut rescans,
+            &mut state,
+            &mut PollBackoff::default(),
+        )
+        .await;
         drain_rescans(&mut rescans, &rescan_gate).await;
 
         let games = recv_local_update(&mut rx).await;