hakowork/rojo
3
0
Fork 0
mirror of https://github.com/rojo-rbx/rojo.git synced 2026-10-23 05:51:39 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Implement comprehensive fixes for sync system issues

Browse source 
- Introduced a new plan to address eight identified issues in the sync system, including improvements to standalone script handling, property persistence, and file watcher behavior.
- Key changes include extracting standalone-to-directory conversion logic, enhancing property update persistence, and implementing a recovery mechanism for rapid delete+recreate scenarios.
- Added integration tests to ensure coverage for each fix and improve overall system reliability.
- Enhanced logging for VFS events to aid in debugging file watcher desynchronization issues.

This commit significantly improves the robustness of the sync system and addresses critical issues identified in the audit.
This commit is contained in:
Joe 2026-02-05 18:25:47 -06:00
parent 4a03d4bab5
commit 0968e74208
5 changed files with 1651 additions and 319 deletions
Download patch file Download diff file Expand all files Collapse all files

528
src/change_processor.rs
View file

@ -3,6 +3,7 @@ use jod_thread::JoinHandle;
use memofs::{IoResultExt, Vfs, VfsEvent};
use rbx_dom_weak::types::{Ref, Variant};
use std::path::PathBuf;
use std::time::{Duration, Instant};
use std::{
fs,
sync::{Arc, Mutex},
@ -51,6 +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>>>,
) -> Self {
let (shutdown_sender, shutdown_receiver) = crossbeam_channel::bounded(1);
let vfs_receiver = vfs.event_receiver();
@ -58,6 +60,8 @@ impl ChangeProcessor {
tree,
vfs,
message_queue,
pending_recovery: Mutex::new(Vec::new()),
suppressed_paths,
};
let job_thread = jod_thread::Builder::new()
@ -69,6 +73,7 @@ impl ChangeProcessor {
select! {
recv(vfs_receiver) -> event => {
task.handle_vfs_event(event?);
task.process_pending_recoveries();
},
recv(tree_mutation_receiver) -> patch_set => {
task.handle_tree_event(patch_set?);
@ -77,6 +82,11 @@ impl ChangeProcessor {
log::trace!("ChangeProcessor shutdown signal received...");
return Ok(());
},
default(Duration::from_millis(500)) => {
// Periodic sweep even when no events arrive —
// catches files recreated after their Remove was processed.
task.process_pending_recoveries();
},
}
}
})
@ -111,6 +121,16 @@ struct JobThreadContext {
/// Whenever changes are applied to the DOM, we should push those changes
/// into this message queue to inform any connected clients.
message_queue: Arc<MessageQueue<AppliedPatchSet>>,
/// Paths recently removed from the tree that should be re-checked after a
/// delay. On Windows, rapid delete+recreate (e.g., editor undo) can cause
/// the Remove event to arrive but the Create event to be lost. We record
/// removed paths here and periodically verify they are still gone.
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>>>,
}
impl JobThreadContext {
@ -133,23 +153,41 @@ impl JobThreadContext {
let affected_ids = loop {
let ids = tree.get_ids_at_path(current_path);
log::trace!("Path {} affects IDs {:?}", current_path.display(), ids);
log::info!(
"apply_patches: path {} affects IDs {:?}",
current_path.display(),
ids
);
if !ids.is_empty() {
break ids.to_vec();
}
log::trace!("Trying parent path...");
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(),
}
};
if affected_ids.is_empty() {
log::info!(
"apply_patches: no affected instances found for path {}",
path.display()
);
}
for id in affected_ids {
if let Some(patch) = compute_and_apply_changes(&mut tree, &self.vfs, id) {
if !patch.is_empty() {
applied_patches.push(patch);
if let Some(result) = compute_and_apply_changes(&mut tree, &self.vfs, id) {
// If an instance was removed, schedule a recovery check
// in case the path is recreated momentarily.
if let Some(removed_path) = result.removed_path {
let mut pending = self.pending_recovery.lock().unwrap();
pending.push((removed_path, Instant::now()));
}
if !result.applied.is_empty() {
applied_patches.push(result.applied);
}
}
}
@ -158,12 +196,31 @@ impl JobThreadContext {
}
fn handle_vfs_event(&self, event: VfsEvent) {
// Log VFS events at debug level (not too verbose, but visible when needed)
// Log EVERY VFS event at INFO level for diagnostics.
// This is intentionally verbose — it is critical for debugging
// file watcher desync issues (e.g., rapid delete+recreate).
match &event {
VfsEvent::Create(path) => log::debug!("VFS: File created: {}", path.display()),
VfsEvent::Write(path) => log::debug!("VFS: File modified: {}", path.display()),
VfsEvent::Remove(path) => log::debug!("VFS: File removed: {}", path.display()),
_ => log::trace!("VFS: Other event: {:?}", event),
VfsEvent::Create(path) => log::info!("VFS event: CREATE {}", path.display()),
VfsEvent::Write(path) => log::info!("VFS event: WRITE {}", path.display()),
VfsEvent::Remove(path) => log::info!("VFS event: REMOVE {}", path.display()),
_ => log::info!("VFS event: OTHER {:?}", event),
}
// Check if this event should be suppressed (one-shot, from API syncback).
let event_path = match &event {
VfsEvent::Create(p) | VfsEvent::Write(p) | VfsEvent::Remove(p) => Some(p.clone()),
_ => None,
};
if let Some(ref path) = event_path {
let mut suppressed = self.suppressed_paths.lock().unwrap();
if suppressed.remove(path) {
// 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;
}
}
// Update the VFS immediately with the event.
@ -177,15 +234,43 @@ impl JobThreadContext {
VfsEvent::Create(path) | VfsEvent::Write(path) => {
// The path might have been deleted before we could process this event
// (e.g., during syncback when directories are being removed).
// In that case, just skip this event.
// If canonicalize fails, retry once after a brief delay — on Windows,
// the file may still be locked by the editor process.
match self.vfs.canonicalize(&path) {
Ok(canonical_path) => self.apply_patches(canonical_path),
Err(_) => {
log::trace!(
"Skipping create/write event for {:?} - path no longer exists",
path
Ok(canonical_path) => {
log::info!(
"VFS: canonicalize OK for {} -> {}",
path.display(),
canonical_path.display()
);
Vec::new()
self.apply_patches(canonical_path)
}
Err(err) => {
log::info!(
"VFS: canonicalize FAILED for {} ({}), retrying after 50ms...",
path.display(),
err
);
std::thread::sleep(Duration::from_millis(50));
match self.vfs.canonicalize(&path) {
Ok(canonical_path) => {
log::info!(
"VFS: retry canonicalize OK for {} -> {}",
path.display(),
canonical_path.display()
);
self.apply_patches(canonical_path)
}
Err(err2) => {
log::warn!(
"VFS: Create/Write event DROPPED for {} after retry — \
canonicalize failed: {}. File may be desynchronized!",
path.display(),
err2
);
Vec::new()
}
}
}
}
}
@ -197,12 +282,19 @@ impl JobThreadContext {
let parent = path.parent().unwrap();
let file_name = path.file_name().unwrap();
match self.vfs.canonicalize(parent) {
Ok(parent_normalized) => self.apply_patches(parent_normalized.join(file_name)),
Err(_) => {
// Parent directory was also deleted, skip this event
log::trace!(
"Skipping remove event for {:?} - parent no longer exists",
path
Ok(parent_normalized) => {
let resolved = parent_normalized.join(file_name);
log::info!(
"VFS: Remove resolved to {}",
resolved.display()
);
self.apply_patches(resolved)
}
Err(err) => {
log::info!(
"VFS: Skipping remove event for {} — parent no longer exists: {}",
path.display(),
err
);
Vec::new()
}
@ -214,19 +306,21 @@ impl JobThreadContext {
}
};
// Log patch application summary at debug level
// Log patch application summary at INFO level
if !applied_patches.is_empty() {
let total_added: usize = applied_patches.iter().map(|p| p.added.len()).sum();
let total_removed: usize = applied_patches.iter().map(|p| p.removed.len()).sum();
let total_updated: usize = applied_patches.iter().map(|p| p.updated.len()).sum();
if total_added > 0 || total_removed > 0 || total_updated > 0 {
log::debug!(
log::info!(
"VFS event applied: {} added, {} removed, {} updated",
total_added,
total_removed,
total_updated
);
}
} else {
log::info!("VFS event applied: no changes");
}
// Notify anyone listening to the message queue about the changes we
@ -234,6 +328,57 @@ impl JobThreadContext {
self.message_queue.push_messages(&applied_patches);
}
/// Processes any pending recovery checks for paths that were recently
/// removed. If a path has reappeared on the real filesystem after the
/// recovery delay, we trigger a re-snapshot to bring the tree back in sync.
fn process_pending_recoveries(&self) {
const RECOVERY_DELAY: Duration = Duration::from_millis(200);
let ready: Vec<PathBuf> = {
let mut pending = self.pending_recovery.lock().unwrap();
let now = Instant::now();
// Drain entries that are old enough to check
let mut ready = Vec::new();
pending.retain(|(path, recorded_at)| {
if now.duration_since(*recorded_at) >= RECOVERY_DELAY {
ready.push(path.clone());
false // remove from pending
} else {
true // keep in pending
}
});
ready
};
for path in ready {
if std::fs::metadata(&path).is_ok() {
log::info!(
"VFS recovery: path {} was removed but has reappeared on disk. Re-snapshotting.",
path.display()
);
let patches = self.apply_patches(path);
if !patches.is_empty() {
let total_added: usize = patches.iter().map(|p| p.added.len()).sum();
let total_removed: usize = patches.iter().map(|p| p.removed.len()).sum();
let total_updated: usize = patches.iter().map(|p| p.updated.len()).sum();
log::info!(
"VFS recovery applied: {} added, {} removed, {} updated",
total_added,
total_removed,
total_updated
);
self.message_queue.push_messages(&patches);
}
} else {
log::info!(
"VFS recovery: path {} confirmed removed from disk.",
path.display()
);
}
}
}
fn handle_tree_event(&self, patch_set: PatchSet) {
// Log incoming patch summary at debug level
log::debug!(
@ -251,13 +396,40 @@ impl JobThreadContext {
if let Some(instigating_source) = &instance.metadata().instigating_source {
match instigating_source {
InstigatingSource::Path(path) => {
log::info!("Two-way sync: Removing file {}", path.display());
if let Err(err) = fs::remove_file(path) {
log::error!(
"Failed to remove file {:?} for instance {:?}: {}",
path,
id,
err
// Guard: file may already be deleted by the API's
// syncback_removed_instance before this PatchSet arrives.
if path.exists() {
if path.is_dir() {
log::info!(
"Two-way sync: Removing directory {}",
path.display()
);
if let Err(err) = fs::remove_dir_all(path) {
log::error!(
"Failed to remove directory {:?} for instance {:?}: {}",
path,
id,
err
);
}
} else {
log::info!(
"Two-way sync: Removing file {}",
path.display()
);
if let Err(err) = fs::remove_file(path) {
log::error!(
"Failed to remove file {:?} for instance {:?}: {}",
path,
id,
err
);
}
}
} else {
log::info!(
"Two-way sync: File already removed: {}",
path.display()
);
}
}
@ -269,7 +441,6 @@ impl JobThreadContext {
}
}
} else {
// TODO
log::warn!(
"Cannot remove instance {:?}, it is not an instigating source.",
id
@ -284,12 +455,147 @@ impl JobThreadContext {
let id = update.id;
if let Some(instance) = tree.get_instance(id) {
if update.changed_name.is_some() {
log::warn!("Cannot rename instances yet.");
// Handle instance rename on disk
if let Some(ref new_name) = update.changed_name {
if let Some(instigating_source) = &instance.metadata().instigating_source {
match instigating_source {
InstigatingSource::Path(path) => {
if path.exists() {
let old_name = instance.name();
// Compute new path by replacing the old name
// in the filename/directory name
if let Some(parent) = path.parent() {
let old_file_name = path
.file_name()
.and_then(|f| f.to_str())
.unwrap_or("");
let new_file_name =
old_file_name.replacen(old_name, new_name, 1);
let new_path = parent.join(&new_file_name);
if new_path != *path {
log::info!(
"Two-way sync: Renaming {} -> {}",
path.display(),
new_path.display()
);
if let Err(err) = fs::rename(path, &new_path) {
log::error!(
"Failed to rename {:?} to {:?}: {}",
path,
new_path,
err
);
} else {
// Also rename adjacent meta file if it exists
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 _ = fs::rename(&old_meta, &new_meta);
}
}
}
}
}
}
InstigatingSource::ProjectNode { .. } => {
log::warn!(
"Cannot rename instance {:?} — defined in project file",
id
);
}
}
} else {
log::warn!(
"Cannot rename instance {:?} — no instigating source",
id
);
}
}
if update.changed_class_name.is_some() {
log::warn!("Cannot change ClassName yet.");
// Handle ClassName changes (script class transitions)
if let Some(ref new_class) = update.changed_class_name {
if let Some(instigating_source) = &instance.metadata().instigating_source {
match instigating_source {
InstigatingSource::Path(path) => {
let old_class = instance.class_name();
let old_is_script = matches!(
old_class.as_str(),
"ModuleScript" | "Script" | "LocalScript"
);
let new_is_script = matches!(
new_class.as_str(),
"ModuleScript" | "Script" | "LocalScript"
);
if old_is_script && new_is_script && path.exists() {
// Script-to-script transition: rename the file extension
if let Some(parent) = path.parent() {
let name = instance.name();
let new_suffix = match new_class.as_str() {
"ModuleScript" => "",
"Script" => ".server",
"LocalScript" => ".local",
_ => "",
};
let is_init = path
.file_name()
.and_then(|f| f.to_str())
.map(|f| f.starts_with("init."))
.unwrap_or(false);
let new_file_name = if is_init {
if new_suffix.is_empty() {
"init.luau".to_string()
} else {
format!("init{}.luau", new_suffix)
}
} else if new_suffix.is_empty() {
format!("{}.luau", name)
} else {
format!("{}{}.luau", name, new_suffix)
};
let new_path = parent.join(&new_file_name);
if new_path != *path {
log::info!(
"Two-way sync: Changing class {} -> {}, \
renaming {} -> {}",
old_class,
new_class,
path.display(),
new_path.display()
);
if let Err(err) = fs::rename(path, &new_path) {
log::error!(
"Failed to rename {:?} to {:?}: {}",
path,
new_path,
err
);
}
}
}
} else if old_is_script != new_is_script {
log::warn!(
"Cannot change ClassName from {} to {} — \
cross-category changes (script <-> non-script) \
are not yet supported",
old_class,
new_class
);
}
}
InstigatingSource::ProjectNode { .. } => {
log::warn!(
"Cannot change ClassName for {:?} — defined in project file",
id
);
}
}
}
}
if update.changed_metadata.is_some() {
@ -351,7 +657,16 @@ impl JobThreadContext {
}
}
fn compute_and_apply_changes(tree: &mut RojoTree, vfs: &Vfs, id: Ref) -> Option<AppliedPatchSet> {
/// Result of computing and applying changes to a single instance.
/// Includes the path that was removed (if any) for pending recovery tracking.
struct ComputeResult {
applied: AppliedPatchSet,
/// If the instance was removed because its path no longer exists,
/// record the path here so the caller can schedule a recovery check.
removed_path: Option<PathBuf>,
}
fn compute_and_apply_changes(tree: &mut RojoTree, vfs: &Vfs, id: Ref) -> Option<ComputeResult> {
let metadata = tree
.get_metadata(id)
.expect("metadata missing for instance present in tree");
@ -370,41 +685,99 @@ fn compute_and_apply_changes(tree: &mut RojoTree, vfs: &Vfs, id: Ref) -> Option<
// How we process a file change event depends on what created this
// file/folder in the first place.
let applied_patch_set = match instigating_source {
InstigatingSource::Path(path) => match vfs.metadata(path).with_not_found() {
Ok(Some(_)) => {
// Our instance was previously created from a path and that
// path still exists. We can generate a snapshot starting at
// that path and use it as the source for our patch.
match instigating_source {
InstigatingSource::Path(path) => {
log::info!(
"compute_and_apply_changes: checking path {} for instance {:?}",
path.display(),
id
);
let snapshot = match snapshot_from_vfs(&metadata.context, vfs, path) {
Ok(snapshot) => snapshot,
Err(err) => {
log::error!("Snapshot error: {:?}", err);
return None;
match vfs.metadata(path).with_not_found() {
Ok(Some(_)) => {
// Our instance was previously created from a path and that
// path still exists. We can generate a snapshot starting at
// that path and use it as the source for our patch.
log::info!(
"compute_and_apply_changes: path EXISTS via VFS, re-snapshotting {}",
path.display()
);
let snapshot = match snapshot_from_vfs(&metadata.context, vfs, path) {
Ok(snapshot) => snapshot,
Err(err) => {
log::error!("Snapshot error: {:?}", err);
return None;
}
};
let patch_set = compute_patch_set(snapshot, tree, id);
let applied = apply_patch_set(tree, patch_set);
Some(ComputeResult {
applied,
removed_path: None,
})
}
Ok(None) => {
// Path not found via VFS. Before removing, verify it's truly
// gone from the REAL filesystem. On Windows, rapid
// delete+recreate (e.g., editor undo) can cause Remove events
// where the file has already been recreated by the time we
// process the event.
if std::fs::metadata(path).is_ok() {
log::info!(
"compute_and_apply_changes: VFS says path removed but REAL \
filesystem confirms it EXISTS: {}. Re-snapshotting instead \
of removing.",
path.display()
);
let snapshot = match snapshot_from_vfs(&metadata.context, vfs, path) {
Ok(snapshot) => snapshot,
Err(err) => {
log::error!("Recovery snapshot error for {}: {:?}", path.display(), err);
return None;
}
};
let patch_set = compute_patch_set(snapshot, tree, id);
let applied = apply_patch_set(tree, patch_set);
Some(ComputeResult {
applied,
removed_path: None,
})
} else {
// Path is genuinely gone from both VFS and real filesystem.
// Remove the instance, but record the path for recovery
// checking — the file might be recreated momentarily.
log::info!(
"compute_and_apply_changes: path NOT FOUND on disk, removing \
instance {:?} for {}. Scheduling recovery check.",
id,
path.display()
);
let removed_path = path.to_path_buf();
let mut patch_set = PatchSet::new();
patch_set.removed_instances.push(id);
let applied = apply_patch_set(tree, patch_set);
Some(ComputeResult {
applied,
removed_path: Some(removed_path),
})
}
};
let patch_set = compute_patch_set(snapshot, tree, id);
apply_patch_set(tree, patch_set)
}
Err(err) => {
log::error!(
"Error processing filesystem change for {}: {:?}",
path.display(),
err
);
None
}
}
Ok(None) => {
// Our instance was previously created from a path, but that
// path no longer exists.
//
// We associate deleting the instigating file for an
// instance with deleting that instance.
let mut patch_set = PatchSet::new();
patch_set.removed_instances.push(id);
apply_patch_set(tree, patch_set)
}
Err(err) => {
log::error!("Error processing filesystem change: {:?}", err);
return None;
}
},
}
InstigatingSource::ProjectNode {
path,
@ -415,6 +788,11 @@ fn compute_and_apply_changes(tree: &mut RojoTree, vfs: &Vfs, id: Ref) -> Option<
// This instance is the direct subject of a project node. Since
// there might be information associated with our instance from
// the project file, we snapshot the entire project node again.
log::info!(
"compute_and_apply_changes: re-snapshotting project node '{}' at {}",
name,
path.display()
);
let snapshot_result = snapshot_project_node(
&metadata.context,
@ -434,9 +812,11 @@ fn compute_and_apply_changes(tree: &mut RojoTree, vfs: &Vfs, id: Ref) -> Option<
};
let patch_set = compute_patch_set(snapshot, tree, id);
apply_patch_set(tree, patch_set)
let applied = apply_patch_set(tree, patch_set);
Some(ComputeResult {
applied,
removed_path: None,
})
}
};
Some(applied_patch_set)
}
}