diff --git a/crates/lanspread-peer/src/services/local_monitor.rs b/crates/lanspread-peer/src/services/local_monitor.rs index e4e0c53..c6c8746 100644 --- a/crates/lanspread-peer/src/services/local_monitor.rs +++ b/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>>, } +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, +} + +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, + 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 { // 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 { +fn snapshot_game_directory( + game_dir: &Path, + catalog: Option<&GameCatalog>, +) -> io::Result { + 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 { 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 { 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 { } }; 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 { }) } -fn snapshot_game_root(game_root: &Path) -> io::Result> { +fn snapshot_game_root( + game_root: &Path, + budget: &mut PollBudget, +) -> io::Result> { 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> 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!["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!["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;