Refactor suppression mechanism for file watcher events

- Updated the suppression logic for file watcher events to use a HashMap instead of a HashSet, allowing for count-based suppression. Each API write increments the count, while each suppressed VFS event decrements it.
- Introduced a new method to register paths in the suppression map, improving the handling of redundant re-snapshots and WebSocket messages.
- Enhanced logging for suppressed events to provide better context during debugging.
- Adjusted related structures and methods across ChangeProcessor, ServeSession, and ApiService to accommodate the new suppression mechanism.
This commit is contained in:
Joe 2026-02-05 18:46:36 -06:00
parent 0968e74208
commit f4f07744c7
4 changed files with 139 additions and 46 deletions

View file

@ -52,7 +52,7 @@ impl ChangeProcessor {
vfs: Arc<Vfs>,
message_queue: Arc<MessageQueue<AppliedPatchSet>>,
tree_mutation_receiver: Receiver<PatchSet>,
suppressed_paths: Arc<Mutex<std::collections::HashSet<PathBuf>>>,
suppressed_paths: Arc<Mutex<std::collections::HashMap<PathBuf, usize>>>,
) -> Self {
let (shutdown_sender, shutdown_receiver) = crossbeam_channel::bounded(1);
let vfs_receiver = vfs.event_receiver();
@ -129,8 +129,8 @@ struct JobThreadContext {
pending_recovery: Mutex<Vec<(PathBuf, Instant)>>,
/// Paths recently written by the API's syncback. Events for these paths
/// are suppressed (one-shot) to avoid redundant re-snapshots.
suppressed_paths: Arc<Mutex<std::collections::HashSet<PathBuf>>>,
/// are suppressed to avoid redundant re-snapshots. Values are counts.
suppressed_paths: Arc<Mutex<std::collections::HashMap<PathBuf, usize>>>,
}
impl JobThreadContext {
@ -163,7 +163,10 @@ impl JobThreadContext {
break ids.to_vec();
}
log::info!("apply_patches: no IDs at {}, trying parent...", current_path.display());
log::info!(
"apply_patches: no IDs at {}, trying parent...",
current_path.display()
);
match current_path.parent() {
Some(parent) => current_path = parent,
None => break Vec::new(),
@ -213,12 +216,20 @@ impl JobThreadContext {
};
if let Some(ref path) = event_path {
let mut suppressed = self.suppressed_paths.lock().unwrap();
if suppressed.remove(path) {
if let Some(count) = suppressed.get_mut(path) {
*count = count.saturating_sub(1);
if *count == 0 {
suppressed.remove(path);
}
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());
log::info!(
"VFS event SUPPRESSED (API syncback echo): {}",
path.display()
);
return;
}
}
@ -284,10 +295,7 @@ impl JobThreadContext {
match self.vfs.canonicalize(parent) {
Ok(parent_normalized) => {
let resolved = parent_normalized.join(file_name);
log::info!(
"VFS: Remove resolved to {}",
resolved.display()
);
log::info!("VFS: Remove resolved to {}", resolved.display());
self.apply_patches(resolved)
}
Err(err) => {
@ -488,11 +496,13 @@ impl JobThreadContext {
);
} else {
// Also rename adjacent meta file if it exists
let old_meta =
parent.join(format!("{}.meta.json5", old_name));
let old_meta = parent
.join(format!("{}.meta.json5", old_name));
if old_meta.exists() {
let new_meta = parent
.join(format!("{}.meta.json5", new_name));
let new_meta = parent.join(format!(
"{}.meta.json5",
new_name
));
let _ = fs::rename(&old_meta, &new_meta);
}
}
@ -508,10 +518,7 @@ impl JobThreadContext {
}
}
} else {
log::warn!(
"Cannot rename instance {:?} — no instigating source",
id
);
log::warn!("Cannot rename instance {:?} — no instigating source", id);
}
}
@ -735,7 +742,11 @@ fn compute_and_apply_changes(tree: &mut RojoTree, vfs: &Vfs, id: Ref) -> Option<
let snapshot = match snapshot_from_vfs(&metadata.context, vfs, path) {
Ok(snapshot) => snapshot,
Err(err) => {
log::error!("Recovery snapshot error for {}: {:?}", path.display(), err);
log::error!(
"Recovery snapshot error for {}: {:?}",
path.display(),
err
);
return None;
}
};

View file

@ -87,9 +87,11 @@ pub struct ServeSession {
tree_mutation_sender: Sender<PatchSet>,
/// Paths recently written by the API's syncback. The ChangeProcessor
/// checks this set and suppresses the file watcher echo for these paths
/// checks this map and suppresses the file watcher echo for these paths
/// to avoid redundant re-snapshots and WebSocket messages.
suppressed_paths: Arc<Mutex<HashSet<std::path::PathBuf>>>,
/// Values are counts — each API write increments, each suppressed VFS event decrements.
#[allow(dead_code)]
suppressed_paths: Arc<Mutex<std::collections::HashMap<std::path::PathBuf, usize>>>,
}
impl ServeSession {
@ -137,7 +139,7 @@ impl ServeSession {
let vfs = Arc::new(vfs);
let (tree_mutation_sender, tree_mutation_receiver) = crossbeam_channel::unbounded();
let suppressed_paths = Arc::new(Mutex::new(HashSet::new()));
let suppressed_paths = Arc::new(Mutex::new(std::collections::HashMap::new()));
log::trace!("Starting ChangeProcessor");
let change_processor = ChangeProcessor::start(
@ -173,9 +175,10 @@ impl ServeSession {
self.tree_mutation_sender.clone()
}
/// Returns a handle to the suppressed paths set, used to avoid
/// Returns a handle to the suppressed paths map, used to avoid
/// file watcher echo when the API writes files to disk.
pub fn suppressed_paths(&self) -> Arc<Mutex<HashSet<std::path::PathBuf>>> {
#[allow(dead_code)]
pub fn suppressed_paths(&self) -> Arc<Mutex<std::collections::HashMap<std::path::PathBuf, usize>>> {
Arc::clone(&self.suppressed_paths)
}

View file

@ -4,9 +4,9 @@
use std::{
collections::{HashMap, HashSet},
fs,
path::PathBuf,
path::{Path, PathBuf},
str::FromStr,
sync::Arc,
sync::{Arc, Mutex},
};
use futures::{sink::SinkExt, stream::StreamExt};
@ -156,11 +156,36 @@ pub async fn call(serve_session: Arc<ServeSession>, mut request: Request<Body>)
pub struct ApiService {
serve_session: Arc<ServeSession>,
suppressed_paths: Arc<Mutex<HashMap<PathBuf, usize>>>,
}
impl ApiService {
pub fn new(serve_session: Arc<ServeSession>) -> Self {
ApiService { serve_session }
let suppressed_paths = serve_session.suppressed_paths();
ApiService {
serve_session,
suppressed_paths,
}
}
/// 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.
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.
if let Ok(canonical) = std::fs::canonicalize(path) {
*suppressed.entry(canonical).or_insert(0) += 1;
} else if let Some(parent) = path.parent() {
if let Ok(canonical_parent) = std::fs::canonicalize(parent) {
if let Some(file_name) = path.file_name() {
*suppressed.entry(canonical_parent.join(file_name)).or_insert(0) += 1;
}
}
}
// Also increment the raw path as fallback.
*suppressed.entry(path.to_path_buf()).or_insert(0) += 1;
}
/// Get a summary of information about the server
@ -297,19 +322,25 @@ impl ApiService {
continue;
}
if is_dir {
self.suppress_path(&path);
if let Err(err) = fs::remove_dir_all(&path) {
log::warn!(
"Failed to remove directory {:?} for instance {:?}: {}",
path, id, err
path,
id,
err
);
} else {
log::info!("Syncback: Removed directory at {}", path.display());
}
} else {
self.suppress_path(&path);
if let Err(err) = fs::remove_file(&path) {
log::warn!(
"Failed to remove file {:?} for instance {:?}: {}",
path, id, err
path,
id,
err
);
} else {
log::info!("Syncback: Removed file at {}", path.display());
@ -324,6 +355,7 @@ impl ApiService {
let meta_path =
parent_dir.join(format!("{}.meta.json5", instance_name));
if meta_path.exists() {
self.suppress_path(&meta_path);
let _ = fs::remove_file(&meta_path);
log::info!(
"Syncback: Removed adjacent meta file at {}",
@ -751,6 +783,7 @@ impl ApiService {
existing_path.to_path_buf()
};
self.suppress_path(&file_path);
fs::write(&file_path, source.as_bytes())
.with_context(|| format!("Failed to write file: {}", file_path.display()))?;
@ -773,6 +806,7 @@ impl ApiService {
let new_dir = parent_dir.join(script_name);
// Create the directory
self.suppress_path(&new_dir);
fs::create_dir_all(&new_dir).with_context(|| {
format!(
"Failed to create directory for script with children: {}",
@ -790,12 +824,14 @@ impl ApiService {
// Move the script content to init file
let init_path = new_dir.join(init_name);
self.suppress_path(&init_path);
fs::write(&init_path, source.as_bytes()).with_context(|| {
format!("Failed to write init file: {}", init_path.display())
})?;
// Remove the old standalone file
if existing_path.exists() && existing_path != init_path {
self.suppress_path(existing_path);
fs::remove_file(existing_path).with_context(|| {
format!(
"Failed to remove old standalone script: {}",
@ -852,6 +888,7 @@ impl ApiService {
} else {
// It's a standalone file (e.g., .model.json5)
let content = self.serialize_instance_to_model_json(added)?;
self.suppress_path(existing_path);
fs::write(existing_path, &content).with_context(|| {
format!("Failed to write file: {}", existing_path.display())
})?;
@ -891,6 +928,7 @@ impl ApiService {
// Create the directory
let new_dir = containing_dir.join(script_name);
self.suppress_path(&new_dir);
fs::create_dir_all(&new_dir).with_context(|| {
format!(
"Failed to create directory for script conversion: {}",
@ -908,12 +946,13 @@ impl ApiService {
// Write source content to the init file
let init_path = new_dir.join(init_name);
fs::write(&init_path, source.as_bytes()).with_context(|| {
format!("Failed to write init file: {}", init_path.display())
})?;
self.suppress_path(&init_path);
fs::write(&init_path, source.as_bytes())
.with_context(|| format!("Failed to write init file: {}", init_path.display()))?;
// Remove the old standalone file
if standalone_path.exists() && standalone_path != init_path {
self.suppress_path(standalone_path);
fs::remove_file(standalone_path).with_context(|| {
format!(
"Failed to remove old standalone script: {}",
@ -964,6 +1003,7 @@ impl ApiService {
use anyhow::Context;
let new_dir = containing_dir.join(instance_name);
self.suppress_path(&new_dir);
fs::create_dir_all(&new_dir).with_context(|| {
format!(
"Failed to create directory for instance conversion: {}",
@ -989,9 +1029,9 @@ impl ApiService {
if standalone_path.exists() {
let content = fs::read(standalone_path).unwrap_or_default();
let init_meta_path = new_dir.join("init.meta.json5");
fs::write(&init_meta_path, &content).with_context(|| {
format!("Failed to write {}", init_meta_path.display())
})?;
self.suppress_path(&init_meta_path);
fs::write(&init_meta_path, &content)
.with_context(|| format!("Failed to write {}", init_meta_path.display()))?;
}
} else if file_ext == "txt" {
// StringValue .txt → init.meta.json5 with className and Value property
@ -1009,17 +1049,17 @@ impl ApiService {
let init_meta_path = new_dir.join("init.meta.json5");
let content = crate::json::to_vec_pretty_sorted(&meta)
.context("Failed to serialize init.meta.json5")?;
fs::write(&init_meta_path, &content).with_context(|| {
format!("Failed to write {}", init_meta_path.display())
})?;
self.suppress_path(&init_meta_path);
fs::write(&init_meta_path, &content)
.with_context(|| format!("Failed to write {}", init_meta_path.display()))?;
} else if file_ext == "csv" {
// LocalizationTable .csv → init.csv
if standalone_path.exists() {
let content = fs::read(standalone_path).unwrap_or_default();
let init_csv_path = new_dir.join("init.csv");
fs::write(&init_csv_path, &content).with_context(|| {
format!("Failed to write {}", init_csv_path.display())
})?;
self.suppress_path(&init_csv_path);
fs::write(&init_csv_path, &content)
.with_context(|| format!("Failed to write {}", init_csv_path.display()))?;
}
} else {
// Generic fallback: create init.meta.json5 with className
@ -1029,13 +1069,14 @@ impl ApiService {
let init_meta_path = new_dir.join("init.meta.json5");
let content = crate::json::to_vec_pretty_sorted(&meta)
.context("Failed to serialize init.meta.json5")?;
fs::write(&init_meta_path, &content).with_context(|| {
format!("Failed to write {}", init_meta_path.display())
})?;
self.suppress_path(&init_meta_path);
fs::write(&init_meta_path, &content)
.with_context(|| format!("Failed to write {}", init_meta_path.display()))?;
}
// Remove the old standalone file
if standalone_path.exists() {
self.suppress_path(standalone_path);
fs::remove_file(standalone_path).with_context(|| {
format!(
"Failed to remove old standalone file: {}",
@ -1111,11 +1152,13 @@ impl ApiService {
// Delete the file or directory
if instance_path.is_dir() {
self.suppress_path(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);
fs::remove_file(instance_path)
.with_context(|| format!("Failed to remove file: {}", instance_path.display()))?;
log::info!("Syncback: Removed file at {}", instance_path.display());
@ -1340,19 +1383,23 @@ impl ApiService {
added.name
);
if old_path.exists() {
self.suppress_path(old_path);
let _ = fs::remove_file(old_path);
}
let meta_path = parent_dir.join(format!("{}.meta.json5", added.name));
if meta_path.exists() {
self.suppress_path(&meta_path);
let _ = fs::remove_file(&meta_path);
}
}
let dir_path = parent_dir.join(&added.name);
self.suppress_path(&dir_path);
fs::create_dir_all(&dir_path).with_context(|| {
format!("Failed to create directory: {}", dir_path.display())
})?;
let init_path = dir_path.join("init.luau");
self.suppress_path(&init_path);
fs::write(&init_path, source.as_bytes()).with_context(|| {
format!("Failed to write file: {}", init_path.display())
})?;
@ -1363,6 +1410,7 @@ impl ApiService {
}
} else {
let file_path = parent_dir.join(format!("{}.luau", added.name));
self.suppress_path(&file_path);
fs::write(&file_path, source.as_bytes()).with_context(|| {
format!("Failed to write file: {}", file_path.display())
})?;
@ -1387,19 +1435,23 @@ impl ApiService {
added.name
);
if old_path.exists() {
self.suppress_path(old_path);
let _ = fs::remove_file(old_path);
}
let meta_path = parent_dir.join(format!("{}.meta.json5", added.name));
if meta_path.exists() {
self.suppress_path(&meta_path);
let _ = fs::remove_file(&meta_path);
}
}
let dir_path = parent_dir.join(&added.name);
self.suppress_path(&dir_path);
fs::create_dir_all(&dir_path).with_context(|| {
format!("Failed to create directory: {}", dir_path.display())
})?;
let init_path = dir_path.join(format!("init.{}.luau", script_suffix));
self.suppress_path(&init_path);
fs::write(&init_path, source.as_bytes()).with_context(|| {
format!("Failed to write file: {}", init_path.display())
})?;
@ -1411,6 +1463,7 @@ impl ApiService {
} else {
let file_path =
parent_dir.join(format!("{}.{}.luau", added.name, script_suffix));
self.suppress_path(&file_path);
fs::write(&file_path, source.as_bytes()).with_context(|| {
format!("Failed to write file: {}", file_path.display())
})?;
@ -1434,19 +1487,23 @@ impl ApiService {
added.name
);
if old_path.exists() {
self.suppress_path(old_path);
let _ = fs::remove_file(old_path);
}
let meta_path = parent_dir.join(format!("{}.meta.json5", added.name));
if meta_path.exists() {
self.suppress_path(&meta_path);
let _ = fs::remove_file(&meta_path);
}
}
let dir_path = parent_dir.join(&added.name);
self.suppress_path(&dir_path);
fs::create_dir_all(&dir_path).with_context(|| {
format!("Failed to create directory: {}", dir_path.display())
})?;
let init_path = dir_path.join("init.local.luau");
self.suppress_path(&init_path);
fs::write(&init_path, source.as_bytes()).with_context(|| {
format!("Failed to write file: {}", init_path.display())
})?;
@ -1457,6 +1514,7 @@ impl ApiService {
}
} else {
let file_path = parent_dir.join(format!("{}.local.luau", added.name));
self.suppress_path(&file_path);
fs::write(&file_path, source.as_bytes()).with_context(|| {
format!("Failed to write file: {}", file_path.display())
})?;
@ -1469,6 +1527,7 @@ impl ApiService {
"Folder" | "Configuration" | "Tool" | "ScreenGui" | "SurfaceGui" | "BillboardGui"
| "AdGui" => {
let dir_path = parent_dir.join(&added.name);
self.suppress_path(&dir_path);
fs::create_dir_all(&dir_path).with_context(|| {
format!("Failed to create directory: {}", dir_path.display())
})?;
@ -1484,7 +1543,9 @@ impl ApiService {
// Add .gitkeep for empty directories with no metadata (matches dedicated syncback)
// Uses !has_children which accounts for filtered duplicate children
if !has_children && !has_metadata {
fs::write(dir_path.join(".gitkeep"), b"")
let gitkeep = dir_path.join(".gitkeep");
self.suppress_path(&gitkeep);
fs::write(gitkeep, b"")
.with_context(|| "Failed to write .gitkeep")?;
}
@ -1505,6 +1566,7 @@ impl ApiService {
if has_children {
// Must become directory - store StringValue data in init.meta.json5
let dir_path = parent_dir.join(&added.name);
self.suppress_path(&dir_path);
fs::create_dir_all(&dir_path).with_context(|| {
format!("Failed to create directory: {}", dir_path.display())
})?;
@ -1526,6 +1588,7 @@ impl ApiService {
})
.unwrap_or_default();
let file_path = parent_dir.join(format!("{}.txt", added.name));
self.suppress_path(&file_path);
fs::write(&file_path, value.as_bytes()).with_context(|| {
format!("Failed to write file: {}", file_path.display())
})?;
@ -1541,10 +1604,12 @@ impl ApiService {
if has_children {
// Must become directory with init.csv
let dir_path = parent_dir.join(&added.name);
self.suppress_path(&dir_path);
fs::create_dir_all(&dir_path).with_context(|| {
format!("Failed to create directory: {}", dir_path.display())
})?;
let init_path = dir_path.join("init.csv");
self.suppress_path(&init_path);
fs::write(&init_path, content.as_bytes()).with_context(|| {
format!("Failed to write file: {}", init_path.display())
})?;
@ -1557,6 +1622,7 @@ impl ApiService {
}
} else {
let file_path = parent_dir.join(format!("{}.csv", added.name));
self.suppress_path(&file_path);
fs::write(&file_path, content.as_bytes()).with_context(|| {
format!("Failed to write file: {}", file_path.display())
})?;
@ -1584,11 +1650,13 @@ impl ApiService {
added.name
);
if old_path.exists() {
self.suppress_path(old_path);
let _ = fs::remove_file(old_path);
}
}
let dir_path = parent_dir.join(&added.name);
self.suppress_path(&dir_path);
fs::create_dir_all(&dir_path).with_context(|| {
format!("Failed to create directory: {}", dir_path.display())
})?;
@ -1614,6 +1682,7 @@ impl ApiService {
ExistingFileFormat::Standalone(p) => p.clone(),
_ => parent_dir.join(format!("{}.model.json5", added.name)),
};
self.suppress_path(&file_path);
fs::write(&file_path, &content).with_context(|| {
format!("Failed to write file: {}", file_path.display())
})?;
@ -1651,6 +1720,7 @@ impl ApiService {
let meta_path = dir_path.join("init.meta.json5");
let content = crate::json::to_vec_pretty_sorted(&meta)
.context("Failed to serialize init.meta.json5")?;
self.suppress_path(&meta_path);
fs::write(&meta_path, &content)
.with_context(|| format!("Failed to write meta file: {}", meta_path.display()))?;
log::info!(
@ -1682,6 +1752,7 @@ impl ApiService {
let meta_path = dir_path.join("init.meta.json5");
let content = crate::json::to_vec_pretty_sorted(&meta)
.context("Failed to serialize init.meta.json5")?;
self.suppress_path(&meta_path);
fs::write(&meta_path, &content)
.with_context(|| format!("Failed to write meta file: {}", meta_path.display()))?;
log::info!(
@ -1709,6 +1780,7 @@ impl ApiService {
let meta_path = dir_path.join("init.meta.json5");
let content = crate::json::to_vec_pretty_sorted(&meta)
.context("Failed to serialize init.meta.json5")?;
self.suppress_path(&meta_path);
fs::write(&meta_path, &content)
.with_context(|| format!("Failed to write meta file: {}", meta_path.display()))?;
log::info!(
@ -1913,6 +1985,7 @@ impl ApiService {
let meta_path = parent_dir.join(format!("{}.meta.json5", name));
let content =
crate::json::to_vec_pretty_sorted(&meta).context("Failed to serialize meta.json5")?;
self.suppress_path(&meta_path);
fs::write(&meta_path, &content)
.with_context(|| format!("Failed to write meta file: {}", meta_path.display()))?;
log::info!(
@ -2024,6 +2097,7 @@ impl ApiService {
let meta = self.merge_or_build_meta(&meta_path, None, properties, attributes)?;
let content = crate::json::to_vec_pretty_sorted(&meta)
.context("Failed to serialize init.meta.json5")?;
self.suppress_path(&meta_path);
fs::write(&meta_path, &content)
.with_context(|| format!("Failed to write {}", meta_path.display()))?;
@ -2040,6 +2114,7 @@ impl ApiService {
let meta = self.merge_or_build_meta(&meta_path, None, properties, attributes)?;
let content = crate::json::to_vec_pretty_sorted(&meta)
.context("Failed to serialize meta.json5")?;
self.suppress_path(&meta_path);
fs::write(&meta_path, &content)
.with_context(|| format!("Failed to write {}", meta_path.display()))?;
@ -2057,6 +2132,7 @@ impl ApiService {
)?;
let content = crate::json::to_vec_pretty_sorted(&meta)
.context("Failed to serialize model file")?;
self.suppress_path(inst_path);
fs::write(inst_path, &content)
.with_context(|| format!("Failed to write {}", inst_path.display()))?;