diff --git a/src/change_processor.rs b/src/change_processor.rs index 8755a0fe..6bc48122 100644 --- a/src/change_processor.rs +++ b/src/change_processor.rs @@ -53,7 +53,7 @@ impl ChangeProcessor { vfs: Arc, message_queue: Arc>, tree_mutation_receiver: Receiver, - suppressed_paths: Arc>>, + suppressed_paths: Arc>>, ) -> Self { let (shutdown_sender, shutdown_receiver) = crossbeam_channel::bounded(1); let vfs_receiver = vfs.event_receiver(); @@ -130,8 +130,8 @@ struct JobThreadContext { pending_recovery: Mutex>, /// Paths recently written by the API's syncback. Events for these paths - /// are suppressed to avoid redundant re-snapshots. Values are counts. - suppressed_paths: Arc>>, + /// are suppressed to avoid redundant re-snapshots. Values are `(remove_count, create_write_count)`. + suppressed_paths: Arc>>, } impl JobThreadContext { @@ -160,12 +160,9 @@ impl JobThreadContext { None } - /// Suppress a single VFS event for the given path. Prevents the - /// ChangeProcessor from re-processing a file-watcher event that we - /// triggered ourselves (renames, writes). - fn suppress_path(&self, path: &Path) { - let mut suppressed = self.suppressed_paths.lock().unwrap(); - let key = if let Ok(canonical) = std::fs::canonicalize(path) { + /// Canonicalize a path for use as a suppression map key. + fn suppression_key(path: &Path) -> PathBuf { + if let Ok(canonical) = std::fs::canonicalize(path) { canonical } else if let Some(parent) = path.parent() { if let Ok(canonical_parent) = std::fs::canonicalize(parent) { @@ -179,33 +176,45 @@ impl JobThreadContext { } } else { path.to_path_buf() - }; - *suppressed.entry(key).or_insert(0) += 1; + } } - /// Remove a suppression previously added by [`suppress_path`]. Called when - /// the filesystem operation that warranted the suppression failed, so that - /// future VFS events for that path are not incorrectly swallowed. + /// Suppress the next Create/Write VFS event for the given path. + /// Prevents re-processing a file-watcher event we triggered ourselves. + fn suppress_path(&self, path: &Path) { + let mut suppressed = self.suppressed_paths.lock().unwrap(); + let key = Self::suppression_key(path); + suppressed.entry(key).or_insert((0, 0)).1 += 1; + } + + /// Suppress the next Remove VFS event for the given path. + fn suppress_path_remove(&self, path: &Path) { + let mut suppressed = self.suppressed_paths.lock().unwrap(); + let key = Self::suppression_key(path); + suppressed.entry(key).or_insert((0, 0)).0 += 1; + } + + /// Remove a Create/Write suppression previously added by [`suppress_path`]. + /// Called when the filesystem operation failed, so that future VFS events + /// for that path are not incorrectly swallowed. fn unsuppress_path(&self, path: &Path) { let mut suppressed = self.suppressed_paths.lock().unwrap(); - let key = if let Ok(canonical) = std::fs::canonicalize(path) { - canonical - } else if let Some(parent) = path.parent() { - if let Ok(canonical_parent) = std::fs::canonicalize(parent) { - if let Some(file_name) = path.file_name() { - canonical_parent.join(file_name) - } else { - path.to_path_buf() - } - } else { - path.to_path_buf() + let key = Self::suppression_key(path); + if let Some(counts) = suppressed.get_mut(&key) { + counts.1 = counts.1.saturating_sub(1); + if counts.0 == 0 && counts.1 == 0 { + suppressed.remove(&key); } - } else { - path.to_path_buf() - }; - if let Some(count) = suppressed.get_mut(&key) { - *count -= 1; - if *count == 0 { + } + } + + /// Remove a Remove suppression previously added by [`suppress_path_remove`]. + fn unsuppress_path_remove(&self, path: &Path) { + let mut suppressed = self.suppressed_paths.lock().unwrap(); + let key = Self::suppression_key(path); + if let Some(counts) = suppressed.get_mut(&key) { + counts.0 = counts.0.saturating_sub(1); + if counts.0 == 0 && counts.1 == 0 { suppressed.remove(&key); } } @@ -287,8 +296,10 @@ impl JobThreadContext { } // Check if this event should be suppressed (one-shot, from API syncback). - // Try both the raw event path and its canonical form since suppress_path - // stores whichever form it could resolve at insertion time. + // Suppressions are event-type-aware: a Remove suppression only matches + // Remove events, and a Create/Write suppression only matches Create/Write + // events. This prevents a Remove suppression from consuming a Create event + // (which can happen on macOS when FSEvents coalesces delete+recreate). let event_path = match &event { VfsEvent::Create(p) | VfsEvent::Write(p) | VfsEvent::Remove(p) => Some(p.clone()), _ => None, @@ -310,22 +321,34 @@ impl JobThreadContext { } }); if let Some(key) = matched_key { - if let Some(count) = suppressed.get_mut(&key) { - *count = count.saturating_sub(1); - if *count == 0 { + if let Some(counts) = suppressed.get_mut(&key) { + let should_suppress = match &event { + VfsEvent::Remove(_) if counts.0 > 0 => { + counts.0 -= 1; + true + } + VfsEvent::Create(_) | VfsEvent::Write(_) if counts.1 > 0 => { + counts.1 -= 1; + true + } + _ => false, + }; + if counts.0 == 0 && counts.1 == 0 { suppressed.remove(&key); } + if should_suppress { + drop(suppressed); + // Still commit so VFS stays consistent, but skip patching. + self.vfs + .commit_event(&event) + .expect("Error applying VFS change"); + log::info!( + "VFS event SUPPRESSED (API syncback echo): {}", + path.display() + ); + return; + } } - drop(suppressed); - // Still commit so VFS stays consistent, but skip patching. - self.vfs - .commit_event(&event) - .expect("Error applying VFS change"); - log::info!( - "VFS event SUPPRESSED (API syncback echo): {}", - path.display() - ); - return; } } @@ -604,12 +627,12 @@ impl JobThreadContext { dir_path.display(), new_dir_path.display() ); - self.suppress_path(dir_path); + self.suppress_path_remove(dir_path); self.suppress_path(&new_dir_path); if let Err(err) = fs::rename(dir_path, &new_dir_path) { - self.unsuppress_path(dir_path); + self.unsuppress_path_remove(dir_path); self.unsuppress_path(&new_dir_path); log::error!( "Failed to rename directory {:?} to {:?}: {}", @@ -631,12 +654,14 @@ impl JobThreadContext { "{}.meta.json5", encoded_new_name )); - self.suppress_path(&old_meta); + self.suppress_path_remove(&old_meta); self.suppress_path(&new_meta); if fs::rename(&old_meta, &new_meta) .is_err() { - self.unsuppress_path(&old_meta); + self.unsuppress_path_remove( + &old_meta, + ); self.unsuppress_path(&new_meta); } } @@ -688,10 +713,10 @@ impl JobThreadContext { path.display(), new_path.display() ); - self.suppress_path(path); + self.suppress_path_remove(path); self.suppress_path(&new_path); if let Err(err) = fs::rename(path, &new_path) { - self.unsuppress_path(path); + self.unsuppress_path_remove(path); self.unsuppress_path(&new_path); log::error!( "Failed to rename {:?} to {:?}: {}", @@ -708,11 +733,11 @@ impl JobThreadContext { "{}.meta.json5", encoded_new_name )); - self.suppress_path(&old_meta); + self.suppress_path_remove(&old_meta); self.suppress_path(&new_meta); if fs::rename(&old_meta, &new_meta).is_err() { - self.unsuppress_path(&old_meta); + self.unsuppress_path_remove(&old_meta); self.unsuppress_path(&new_meta); } } @@ -827,12 +852,12 @@ impl JobThreadContext { actual_file.display(), new_path.display() ); - self.suppress_path(&actual_file); + self.suppress_path_remove(&actual_file); self.suppress_path(&new_path); if let Err(err) = fs::rename(&actual_file, &new_path) { - self.unsuppress_path(&actual_file); + self.unsuppress_path_remove(&actual_file); self.unsuppress_path(&new_path); log::error!( "Failed to rename {:?} to {:?}: {}", diff --git a/src/cli/studio.rs b/src/cli/studio.rs index f29e93e9..88e33ab2 100644 --- a/src/cli/studio.rs +++ b/src/cli/studio.rs @@ -1,4 +1,6 @@ use std::path::PathBuf; +#[cfg(windows)] +use std::process::Command; use anyhow::Context; use clap::Parser; diff --git a/src/serve_session.rs b/src/serve_session.rs index 60430d25..04a0520f 100644 --- a/src/serve_session.rs +++ b/src/serve_session.rs @@ -89,9 +89,10 @@ pub struct ServeSession { /// Paths recently written by the API's syncback. The ChangeProcessor /// checks this map and suppresses the file watcher echo for these paths /// to avoid redundant re-snapshots and WebSocket messages. - /// Values are counts — each API write increments, each suppressed VFS event decrements. + /// Values are `(remove_count, create_write_count)` — each API write increments + /// the appropriate counter, each suppressed VFS event decrements it. #[allow(dead_code)] - suppressed_paths: Arc>>, + suppressed_paths: Arc>>, } impl ServeSession { @@ -180,7 +181,7 @@ impl ServeSession { #[allow(dead_code)] pub fn suppressed_paths( &self, - ) -> Arc>> { + ) -> Arc>> { Arc::clone(&self.suppressed_paths) } diff --git a/src/syncback/fs_snapshot.rs b/src/syncback/fs_snapshot.rs index afc126ed..89e62a06 100644 --- a/src/syncback/fs_snapshot.rs +++ b/src/syncback/fs_snapshot.rs @@ -5,6 +5,9 @@ use std::{ sync::atomic::{AtomicUsize, Ordering}, }; +#[cfg(windows)] +use std::{thread, time::Duration}; + use memofs::Vfs; use rayon::prelude::*; diff --git a/src/web/api.rs b/src/web/api.rs index ca8b49c6..eda15afb 100644 --- a/src/web/api.rs +++ b/src/web/api.rs @@ -157,7 +157,7 @@ pub async fn call(serve_session: Arc, mut request: Request) pub struct ApiService { serve_session: Arc, - suppressed_paths: Arc>>, + suppressed_paths: Arc>>, } /// Derives the directory name from a standalone script's filesystem path. @@ -219,17 +219,9 @@ impl ApiService { } } - /// Register a path in the suppression map so that the ChangeProcessor - /// ignores the next VFS event for it (counter-based echo suppression). - /// Each call increments the counter; each suppressed VFS event decrements it. - /// - /// Inserts exactly ONE entry per call — canonical form preferred, raw as - /// fallback — to avoid leaking the unmatched variant. - fn suppress_path(&self, path: &Path) { - let mut suppressed = self.suppressed_paths.lock().unwrap(); - // Try to canonicalize (adds \\?\ prefix on Windows, resolves symlinks). - // For files that don't exist yet, canonicalize the parent and join. - let key = if let Ok(canonical) = std::fs::canonicalize(path) { + /// Canonicalize a path for use as a suppression map key. + fn suppression_key(path: &Path) -> PathBuf { + if let Ok(canonical) = std::fs::canonicalize(path) { canonical } else if let Some(parent) = path.parent() { if let Ok(canonical_parent) = std::fs::canonicalize(parent) { @@ -243,8 +235,21 @@ impl ApiService { } } else { path.to_path_buf() - }; - *suppressed.entry(key).or_insert(0) += 1; + } + } + + /// Suppress the next Create/Write VFS event for the given path. + fn suppress_path(&self, path: &Path) { + let mut suppressed = self.suppressed_paths.lock().unwrap(); + let key = Self::suppression_key(path); + suppressed.entry(key).or_insert((0, 0)).1 += 1; + } + + /// Suppress the next Remove VFS event for the given path. + fn suppress_path_remove(&self, path: &Path) { + let mut suppressed = self.suppressed_paths.lock().unwrap(); + let key = Self::suppression_key(path); + suppressed.entry(key).or_insert((0, 0)).0 += 1; } /// Get a summary of information about the server @@ -386,7 +391,7 @@ impl ApiService { continue; } if is_dir { - self.suppress_path(&path); + self.suppress_path_remove(&path); if let Err(err) = fs::remove_dir_all(&path) { log::warn!( "Failed to remove directory {:?} for instance {:?}: {}", @@ -399,7 +404,7 @@ impl ApiService { actually_removed.push(id); } } else { - self.suppress_path(&path); + self.suppress_path_remove(&path); if let Err(err) = fs::remove_file(&path) { log::warn!( "Failed to remove file {:?} for instance {:?}: {}", @@ -427,7 +432,7 @@ impl ApiService { let meta_path = parent_dir.join(format!("{}.meta.json5", base_name)); if meta_path.exists() { - self.suppress_path(&meta_path); + self.suppress_path_remove(&meta_path); let _ = fs::remove_file(&meta_path); log::info!( "Syncback: Removed adjacent meta file at {}", @@ -915,7 +920,7 @@ impl ApiService { // Remove the old standalone file if existing_path.exists() && existing_path != init_path { - self.suppress_path(existing_path); + self.suppress_path_remove(existing_path); fs::remove_file(existing_path).with_context(|| { format!( "Failed to remove old standalone script: {}", @@ -1038,7 +1043,7 @@ impl ApiService { // Remove the old standalone file if standalone_path.exists() && standalone_path != init_path { - self.suppress_path(standalone_path); + self.suppress_path_remove(standalone_path); fs::remove_file(standalone_path).with_context(|| { format!( "Failed to remove old standalone script: {}", @@ -1051,7 +1056,7 @@ impl ApiService { let meta_path = containing_dir.join(format!("{}.meta.json5", dir_name)); if meta_path.exists() { let init_meta_path = new_dir.join("init.meta.json5"); - self.suppress_path(&meta_path); + self.suppress_path_remove(&meta_path); self.suppress_path(&init_meta_path); fs::rename(&meta_path, &init_meta_path).with_context(|| { format!( @@ -1227,7 +1232,7 @@ impl ApiService { // Remove the old standalone file if standalone_path.exists() { - self.suppress_path(standalone_path); + self.suppress_path_remove(standalone_path); fs::remove_file(standalone_path).with_context(|| { format!( "Failed to remove old standalone file: {}", @@ -1303,13 +1308,13 @@ impl ApiService { // Delete the file or directory if instance_path.is_dir() { - self.suppress_path(instance_path); + self.suppress_path_remove(instance_path); fs::remove_dir_all(instance_path).with_context(|| { format!("Failed to remove directory: {}", instance_path.display()) })?; log::info!("Syncback: Removed directory at {}", instance_path.display()); } else if instance_path.is_file() { - self.suppress_path(instance_path); + self.suppress_path_remove(instance_path); fs::remove_file(instance_path) .with_context(|| format!("Failed to remove file: {}", instance_path.display()))?; log::info!("Syncback: Removed file at {}", instance_path.display()); @@ -1538,12 +1543,12 @@ impl ApiService { added.name ); if old_path.exists() { - self.suppress_path(old_path); + self.suppress_path_remove(old_path); let _ = fs::remove_file(old_path); } let meta_path = parent_dir.join(format!("{}.meta.json5", encoded_name)); if meta_path.exists() { - self.suppress_path(&meta_path); + self.suppress_path_remove(&meta_path); let _ = fs::remove_file(&meta_path); } } @@ -1590,12 +1595,12 @@ impl ApiService { added.name ); if old_path.exists() { - self.suppress_path(old_path); + self.suppress_path_remove(old_path); let _ = fs::remove_file(old_path); } let meta_path = parent_dir.join(format!("{}.meta.json5", encoded_name)); if meta_path.exists() { - self.suppress_path(&meta_path); + self.suppress_path_remove(&meta_path); let _ = fs::remove_file(&meta_path); } } @@ -1642,12 +1647,12 @@ impl ApiService { added.name ); if old_path.exists() { - self.suppress_path(old_path); + self.suppress_path_remove(old_path); let _ = fs::remove_file(old_path); } let meta_path = parent_dir.join(format!("{}.meta.json5", encoded_name)); if meta_path.exists() { - self.suppress_path(&meta_path); + self.suppress_path_remove(&meta_path); let _ = fs::remove_file(&meta_path); } } @@ -1804,7 +1809,7 @@ impl ApiService { added.name ); if old_path.exists() { - self.suppress_path(old_path); + self.suppress_path_remove(old_path); let _ = fs::remove_file(old_path); } } diff --git a/tests/rojo_test/serve_util.rs b/tests/rojo_test/serve_util.rs index 74857ee1..8a672780 100644 --- a/tests/rojo_test/serve_util.rs +++ b/tests/rojo_test/serve_util.rs @@ -38,6 +38,14 @@ pub fn run_serve_test(test_name: &str, callback: impl FnOnce(TestServeSession, R let mut session = TestServeSession::new(test_name); let info = session.wait_to_come_online(); + // On macOS, the FSEvents-backed file watcher (via notify-debouncer-full) + // may not have fully populated its internal cache by the time the HTTP + // server reports as online. Without a warm cache, the debouncer cannot + // correctly classify events (e.g., it may miss a Remove if the file was + // not yet in the cache). Give it time to complete its initial scan. + #[cfg(target_os = "macos")] + thread::sleep(Duration::from_secs(1)); + redactions.intern(info.session_id); redactions.intern(info.root_instance_id); @@ -91,11 +99,9 @@ impl TestServeSession { // This is an ugly workaround for FSEvents sometimes reporting events // for the above copy operations, similar to this Stack Overflow question: // https://stackoverflow.com/questions/47679298/howto-avoid-receiving-old-events-in-fseventstream-callback-fsevents-framework-o - // We'll hope that 100ms is enough for FSEvents to get whatever it is - // out of its system. - // TODO: find a better way to avoid processing these spurious events. + // 500ms gives FSEvents time to flush any stale events from the copy. #[cfg(target_os = "macos")] - std::thread::sleep(Duration::from_millis(100)); + std::thread::sleep(Duration::from_millis(500)); let port = get_port_number(); let port_string = port.to_string(); @@ -324,8 +330,8 @@ fn deserialize_msgpack<'a, T: Deserialize<'a>>( /// reads back the assigned port, then drops the listener. The brief /// TOCTOU window before Rojo rebinds is negligible on localhost. fn get_port_number() -> usize { - let listener = std::net::TcpListener::bind("127.0.0.1:0") - .expect("Failed to bind ephemeral port for test"); + let listener = + std::net::TcpListener::bind("127.0.0.1:0").expect("Failed to bind ephemeral port for test"); let port = listener.local_addr().unwrap().port() as usize; drop(listener); port