From 6ef323c7da9a90694dd93e721c602d4f7d2b6f97 Mon Sep 17 00:00:00 2001 From: Aly Raffauf Date: Mon, 3 Aug 2026 00:26:55 -0400 Subject: [PATCH] Harden filesystem change handling --- src/app/run.rs | 19 +++++++------- src/filesystem.rs | 65 ++++++++++++++++++++++++++++++++++------------- 2 files changed, 57 insertions(+), 27 deletions(-) diff --git a/src/app/run.rs b/src/app/run.rs index 8d3b552..06e6fb6 100644 --- a/src/app/run.rs +++ b/src/app/run.rs @@ -3,7 +3,7 @@ use std::{collections::BTreeMap, path::Path, time::Duration}; use crate::{ app::{AppResult, AppaService}, domain::FolderId, - filesystem::{drain_events, lock_folder, watch_folder}, + filesystem::{FolderChangeWatcher, lock_app_process, watch_folder}, iroh::NodeHost, storage::FolderConfig, }; @@ -22,7 +22,7 @@ struct RetryState { struct FolderWatcher { folder: FolderConfig, - events: (notify::RecommendedWatcher, std::sync::mpsc::Receiver<()>), + events: FolderChangeWatcher, next_local_scan: tokio::time::Instant, next_peer_poll: tokio::time::Instant, next_safety_scan: tokio::time::Instant, @@ -31,7 +31,7 @@ struct FolderWatcher { impl AppaService { pub async fn run_forever(&self, folder_path: Option<&Path>) -> AppResult<()> { let folders = self.folders_to_run(folder_path)?; - let _lock = lock_folder(&self.paths.data_directory.join("locks"), "device")?; + let _lock = lock_app_process(&self.paths.data_directory.join("locks"))?; let node = self.load_node().await?; node.wait_until_online().await?; let mut watchers = folders @@ -64,7 +64,7 @@ impl AppaService { watcher.schedule_remote_sync(); tracing::info!(folder = %watcher.folder.name, "Peer announced folder changes; checking now"); } - if drain_events(&watcher.events) { + if watcher.events.has_pending_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"); @@ -72,7 +72,7 @@ impl AppaService { if retry_is_pending(retries, watcher.folder.id) { continue; } - if !watcher.is_sync_due() { + if !watcher.is_sync_due(tokio::time::Instant::now()) { continue; } self.sync_watched_folder(watcher, node, retries).await?; @@ -86,7 +86,7 @@ impl AppaService { node: &NodeHost, retries: &mut BTreeMap, ) -> AppResult<()> { - let should_scan_local_files = watcher.is_local_scan_due(); + let should_scan_local_files = watcher.is_local_scan_due(tokio::time::Instant::now()); match self .sync_with_node_using_local_scan(&watcher.folder, node, should_scan_local_files) .await @@ -122,12 +122,11 @@ impl FolderWatcher { }) } - fn is_sync_due(&self) -> bool { - self.is_local_scan_due() || tokio::time::Instant::now() >= self.next_peer_poll + fn is_sync_due(&self, now: tokio::time::Instant) -> bool { + self.is_local_scan_due(now) || now >= self.next_peer_poll } - fn is_local_scan_due(&self) -> bool { - let now = tokio::time::Instant::now(); + fn is_local_scan_due(&self, now: tokio::time::Instant) -> bool { now >= self.next_local_scan || now >= self.next_safety_scan } diff --git a/src/filesystem.rs b/src/filesystem.rs index 2c444f1..919d669 100644 --- a/src/filesystem.rs +++ b/src/filesystem.rs @@ -79,15 +79,21 @@ pub fn safe_destination(folder_path: &Path, relative_path: &str) -> anyhow::Resu } pub fn remove_path(path: &Path) -> anyhow::Result<()> { - if path.is_dir() { - fs::remove_dir_all(path)?; - } else if path.exists() { - fs::remove_file(path)?; + match fs::symlink_metadata(path) { + Ok(metadata) if metadata.file_type().is_dir() => fs::remove_dir_all(path)?, + Ok(_) => fs::remove_file(path)?, + Err(error) if error.kind() == std::io::ErrorKind::NotFound => {} + Err(error) => return Err(error.into()), } Ok(()) } -pub fn watch_folder(path: &Path) -> anyhow::Result<(notify::RecommendedWatcher, Receiver<()>)> { +pub struct FolderChangeWatcher { + _watcher: notify::RecommendedWatcher, + events: Receiver<()>, +} + +pub fn watch_folder(path: &Path) -> anyhow::Result { let (sender, receiver) = mpsc::channel(); let mut watcher = notify::recommended_watcher(move |result: notify::Result| { if matches!( @@ -98,28 +104,32 @@ pub fn watch_folder(path: &Path) -> anyhow::Result<(notify::RecommendedWatcher, } })?; watcher.watch(path, RecursiveMode::Recursive)?; - Ok((watcher, receiver)) + Ok(FolderChangeWatcher { + _watcher: watcher, + events: receiver, + }) } -pub fn drain_events(events: &(notify::RecommendedWatcher, Receiver<()>)) -> bool { - let mut changed = false; - while events.1.try_recv().is_ok() { - changed = true; +impl FolderChangeWatcher { + pub fn has_pending_events(&self) -> bool { + let mut has_pending_events = false; + while self.events.try_recv().is_ok() { + has_pending_events = true; + } + has_pending_events } - changed } pub struct FolderLock { _file: File, } -pub fn lock_folder(lock_directory: &Path, folder_id: &str) -> anyhow::Result { +pub fn lock_app_process(lock_directory: &Path) -> anyhow::Result { fs::create_dir_all(lock_directory)?; - let lock_path = lock_directory.join(format!("{folder_id}.lock")); + let lock_path = lock_directory.join("appa.lock"); let file = File::create(lock_path)?; - file.try_lock().map_err(|error| { - anyhow::anyhow!("another Appa process is already running this folder: {error}") - })?; + file.try_lock() + .map_err(|error| anyhow::anyhow!("another Appa process is already running: {error}"))?; Ok(FolderLock { _file: file }) } @@ -144,6 +154,8 @@ fn collect_entries_into( collect_entries_into(&path, ignore_rules, entries)?; } else if metadata.is_file() { entries.files.push(path); + } else { + tracing::warn!(path = %path.display(), "Skipping unsupported filesystem entry in synced folder"); } } Ok(()) @@ -172,7 +184,7 @@ mod tests { use tempfile::TempDir; - use super::{IgnoreRules, collect_entries, safe_destination}; + use super::{IgnoreRules, collect_entries, remove_path, safe_destination}; #[test] fn excludes_matching_paths_and_the_local_ignore_file() -> anyhow::Result<()> { @@ -209,4 +221,23 @@ mod tests { assert!(safe_destination(folder.path(), "outside/secret.txt").is_err()); Ok(()) } + + #[cfg(unix)] + #[test] + fn removes_a_symlink_without_following_its_target() -> anyhow::Result<()> { + use std::os::unix::fs::symlink; + + let directory = TempDir::new()?; + let target = directory.path().join("target"); + let link = directory.path().join("link"); + fs::create_dir(&target)?; + fs::write(target.join("preserved.txt"), "contents")?; + symlink(&target, &link)?; + + remove_path(&link)?; + + assert!(!link.exists()); + assert!(target.join("preserved.txt").exists()); + Ok(()) + } } -- 2.51.2