From e8d1ef81385ecaa8e3d5046ba89cdfd0aa2d8cfb Mon Sep 17 00:00:00 2001 From: Aly Raffauf Date: Sun, 2 Aug 2026 21:50:50 -0400 Subject: [PATCH] Gate local scans on filesystem changes --- src/app/manifest.rs | 19 +++++++++++++-- src/app/run.rs | 58 ++++++++++++++++++++++++++++++++++++++------- src/app/sync.rs | 12 +++++++++- 3 files changed, 77 insertions(+), 12 deletions(-) diff --git a/src/app/manifest.rs b/src/app/manifest.rs index a63049a..7b56ebe 100644 --- a/src/app/manifest.rs +++ b/src/app/manifest.rs @@ -1,6 +1,7 @@ use std::{ collections::{BTreeMap, BTreeSet}, fs, + time::Instant, }; use time::OffsetDateTime; @@ -22,12 +23,15 @@ pub(super) async fn build_manifest( state_store: &StateStore, node: &NodeHost, ) -> anyhow::Result { + let scan_started_at = Instant::now(); let previous = state_store.load_manifest(folder.id)?; let ignore_rules = IgnoreRules::load(&folder.path)?; let device_id = node.endpoint_address().id.to_string(); let mut entries = BTreeMap::new(); let mut observed_paths = BTreeSet::new(); + let mut file_count = 0; for file_path in collect_files(&folder.path, &ignore_rules)? { + file_count += 1; let path = relative_path(&folder.path, &file_path)?; observed_paths.insert(path.clone()); let metadata = fs::metadata(&file_path)?; @@ -52,7 +56,9 @@ pub(super) async fn build_manifest( }; entries.insert(path, entry); } + let mut directory_count = 0; for directory_path in collect_directories(&folder.path, &ignore_rules)? { + directory_count += 1; let path = relative_path(&folder.path, &directory_path)?; observed_paths.insert(path.clone()); let entry = match previous.entries.get(&path) { @@ -87,13 +93,22 @@ pub(super) async fn build_manifest( .filter(|(path, entry)| previous.entries.get(*path) != Some(*entry)) .map(|(path, _)| path.clone()) .collect(); - Ok(ManifestBuild { + let manifest_build = ManifestBuild { manifest: Manifest { folder_id: folder.id, entries, }, changed_paths, - }) + }; + tracing::info!( + folder = %folder.name, + file_count, + directory_count, + changed_path_count = manifest_build.changed_paths.len(), + elapsed = ?scan_started_at.elapsed(), + "Scanned local folder" + ); + Ok(manifest_build) } pub(super) fn report_local_changes(changed_paths: &[String]) { diff --git a/src/app/run.rs b/src/app/run.rs index 594139b..7b49dd1 100644 --- a/src/app/run.rs +++ b/src/app/run.rs @@ -8,7 +8,10 @@ use crate::{ storage::FolderConfig, }; -const SYNC_INTERVAL: Duration = Duration::from_secs(2); +const WATCHER_POLL_INTERVAL: Duration = Duration::from_millis(250); +const LOCAL_CHANGE_DEBOUNCE: Duration = Duration::from_millis(750); +const PEER_POLL_INTERVAL: Duration = Duration::from_secs(30); +const SAFETY_SCAN_INTERVAL: Duration = Duration::from_secs(10 * 60); const MAX_RETRY_DELAY: Duration = Duration::from_secs(60); #[derive(Clone, Debug)] @@ -20,6 +23,9 @@ struct RetryState { struct FolderWatcher { folder: FolderConfig, events: (notify::RecommendedWatcher, std::sync::mpsc::Receiver<()>), + next_local_scan: tokio::time::Instant, + next_peer_poll: tokio::time::Instant, + next_safety_scan: tokio::time::Instant, } impl AppaService { @@ -28,18 +34,18 @@ impl AppaService { let _lock = lock_folder(&self.paths.data_directory.join("locks"), "device")?; let node = self.load_node().await?; node.wait_until_online().await; - let watchers = folders + let mut watchers = folders .into_iter() .map(FolderWatcher::new) .collect::>>()?; - let mut interval = tokio::time::interval(SYNC_INTERVAL); + let mut interval = tokio::time::interval(WATCHER_POLL_INTERVAL); let mut retries = BTreeMap::new(); tracing::info!(folder_count = watchers.len(), endpoint = %node.endpoint_address().id, "Appa is watching for changes"); loop { tokio::select! { _ = tokio::signal::ctrl_c() => break, - _ = interval.tick() => self.sync_watched_folders(&watchers, &node, &mut retries).await?, + _ = interval.tick() => self.sync_watched_folders(&mut watchers, &node, &mut retries).await?, } } node.shutdown().await?; @@ -48,18 +54,22 @@ impl AppaService { async fn sync_watched_folders( &self, - watchers: &[FolderWatcher], + watchers: &mut [FolderWatcher], node: &NodeHost, retries: &mut BTreeMap, ) -> AppResult<()> { for watcher in watchers { if drain_events(&watcher.events) { retries.remove(&watcher.folder.id); + watcher.next_local_scan = tokio::time::Instant::now() + LOCAL_CHANGE_DEBOUNCE; tracing::info!(folder = %watcher.folder.name, "Filesystem change detected; checking the folder"); } if retry_is_pending(retries, watcher.folder.id) { continue; } + if !watcher.is_sync_due() { + continue; + } self.sync_watched_folder(watcher, node, retries).await?; } Ok(()) @@ -67,13 +77,18 @@ impl AppaService { async fn sync_watched_folder( &self, - watcher: &FolderWatcher, + watcher: &mut FolderWatcher, node: &NodeHost, retries: &mut BTreeMap, ) -> AppResult<()> { - match self.sync_with_node(&watcher.folder, node).await { + let should_scan_local_files = watcher.is_local_scan_due(); + match self + .sync_with_node_using_local_scan(&watcher.folder, node, should_scan_local_files) + .await + { Ok(change_count) => { retries.remove(&watcher.folder.id); + watcher.record_successful_sync(should_scan_local_files); if change_count > 0 { tracing::info!(folder = %watcher.folder.name, change_count, "Applied remote changes"); } @@ -92,7 +107,32 @@ impl AppaService { impl FolderWatcher { fn new(folder: FolderConfig) -> AppResult { let events = watch_folder(&folder.path)?; - Ok(Self { folder, events }) + let now = tokio::time::Instant::now(); + Ok(Self { + folder, + events, + next_local_scan: now, + next_peer_poll: now, + next_safety_scan: now + SAFETY_SCAN_INTERVAL, + }) + } + + fn is_sync_due(&self) -> bool { + self.is_local_scan_due() || tokio::time::Instant::now() >= self.next_peer_poll + } + + fn is_local_scan_due(&self) -> bool { + let now = tokio::time::Instant::now(); + now >= self.next_local_scan || now >= self.next_safety_scan + } + + fn record_successful_sync(&mut self, scanned_local_files: bool) { + let now = tokio::time::Instant::now(); + self.next_peer_poll = now + PEER_POLL_INTERVAL; + if scanned_local_files { + self.next_local_scan = now + SAFETY_SCAN_INTERVAL; + self.next_safety_scan = now + SAFETY_SCAN_INTERVAL; + } } } @@ -115,6 +155,6 @@ fn schedule_retry(retries: &mut BTreeMap, folder_id: Folde pub(super) fn retry_delay(failure_count: u32) -> Duration { let exponent = failure_count.saturating_sub(1).min(5); - let seconds = SYNC_INTERVAL.as_secs() * (1_u64 << exponent); + let seconds = 2_u64 * (1_u64 << exponent); Duration::from_secs(seconds).min(MAX_RETRY_DELAY) } diff --git a/src/app/sync.rs b/src/app/sync.rs index 2095f4b..16d9b59 100644 --- a/src/app/sync.rs +++ b/src/app/sync.rs @@ -14,10 +14,20 @@ impl AppaService { &self, folder: &FolderConfig, node: &NodeHost, + ) -> AppResult { + self.sync_with_node_using_local_scan(folder, node, true) + .await + } + + pub(super) async fn sync_with_node_using_local_scan( + &self, + folder: &FolderConfig, + node: &NodeHost, + should_scan_local_files: bool, ) -> AppResult { self.save_lan_discovered_peers(folder, node)?; self.save_discovered_peers(folder, node)?; - let manifest_build = if folder.mode.can_send() { + let manifest_build = if should_scan_local_files && folder.mode.can_send() { build_manifest(folder, &self.state_store, node).await? } else { ManifestBuild { -- knotmirror 0.0.1