From 9c34ce0cd4177ccc1e635b922c4758e2ef796471 Mon Sep 17 00:00:00 2001 From: Nathan Herald Date: Sun, 6 Sep 2026 00:51:17 +0200 Subject: [PATCH 1/2] Add seam hooks and failing tests for the rename split A local rename can end with both the old and the new path present on every peer. The sync pass materializes with a disk view it captured before the peer step. A file that an inbound session materialized during the peer step is missing from that view, so a rename that lands between the post-peer scan and its materialization looks like a file this machine never had, and the old path is written back. This adds test-only seams between each scan and the materialization that follows it, and five tests that pin events to those seams: - a rename at the post-peer seam splits the file (fails today), - a file created and removed inside one inbound session comes back (fails today), - a file created and removed around the pre-peer scan comes back (fails today), - the same rename one seam earlier is recorded by the scan (passes), - a remote-only path adopted during the peer step is still written (passes). No behavior changes. Refs #175. --- src/sync/engine.rs | 378 +++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 378 insertions(+) diff --git a/src/sync/engine.rs b/src/sync/engine.rs index 7f15451..fca4e27 100644 --- a/src/sync/engine.rs +++ b/src/sync/engine.rs @@ -783,6 +783,27 @@ impl PreparedInbound { } } +/// The points between a scan and the materialization that follows it. +/// +/// A rename or delete that lands in one of these windows is invisible to the +/// scan that just ran and visible to the materialization that follows. Tests +/// pin an event to a seam so the window is exercised on demand rather than by +/// scheduler luck. Production builds carry the enum and an empty call. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) enum PhaseSeam { + /// `sync_once`, after the pre-peer scan and before its materialization. + AfterPrePeerScan, + /// `sync_once`, after the peer step, before the post-peer scan. + BeforePostPeerScan, + /// `sync_once`, after the post-peer scan and before its materialization. + AfterPostPeerScan, + /// `complete_inbound`, after its scan and before its materialization. + AfterInboundScan, +} + +#[cfg(test)] +type SeamHook = Arc; + impl std::fmt::Debug for SyncEngine { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { f.debug_struct("SyncEngine").finish_non_exhaustive() @@ -800,6 +821,9 @@ pub struct SyncEngine { /// for newly added entries. watching: StdMutex>, cancel: CancellationToken, + /// Test-only seam between a scan and the materialization that follows it. + #[cfg(test)] + seam_hook: StdMutex>, } impl SyncEngine { @@ -824,6 +848,8 @@ impl SyncEngine { entries: RwLock::new(HashMap::new()), watching: StdMutex::new(HashSet::new()), cancel, + #[cfg(test)] + seam_hook: StdMutex::new(None), }); engine.load_from_config().await?; Ok(engine) @@ -831,6 +857,23 @@ impl SyncEngine { /// (Re)load entries from `syncs.toml`, keeping existing nodes for entries /// that are unchanged and dropping entries no longer configured. + #[cfg(test)] + pub(crate) fn set_seam_hook(&self, hook: impl Fn(PhaseSeam) + Send + Sync + 'static) { + *self.seam_hook.lock().unwrap() = Some(Arc::new(hook)); + } + + #[cfg(test)] + fn at_seam(&self, seam: PhaseSeam) { + let hook = self.seam_hook.lock().unwrap().clone(); + if let Some(hook) = hook { + hook(seam); + } + } + + #[cfg(not(test))] + #[inline(always)] + fn at_seam(&self, _seam: PhaseSeam) {} + pub async fn load_from_config(&self) -> Result<()> { let book = SyncBook::load_path(self.paths.config_path())?; self.load_book(book).await @@ -1077,6 +1120,7 @@ impl SyncEngine { // It is cheap now because scan_folder reuses recorded hashes for files // whose size and mtime are unchanged. self.scan_entry(&entry).await?; + self.at_seam(PhaseSeam::AfterInboundScan); self.materialize_entry_state(&entry, &baseline).await?; let final_manifest_revision = entry.node.lock().await.manifest_revision(); let final_receipt_revision = entry.disk.lock().unwrap().receipt_revision; @@ -1197,6 +1241,7 @@ impl SyncEngine { let phase = Instant::now(); let scan_changed = self.scan_entry(&entry).await?; EntryWork::add_phase(&entry.work.scan_micros, phase); + self.at_seam(PhaseSeam::AfterPrePeerScan); let phase = Instant::now(); self.materialize_entry_state(&entry, &protected).await?; EntryWork::add_phase(&entry.work.materialize_micros, phase); @@ -1352,9 +1397,11 @@ impl SyncEngine { EntryWork::add_phase(&entry.work.reconcile_micros, reconcile_phase); let _operation = entry.operation.lock().await; + self.at_seam(PhaseSeam::BeforePostPeerScan); let phase = Instant::now(); self.scan_entry(&entry).await?; EntryWork::add_phase(&entry.work.scan_micros, phase); + self.at_seam(PhaseSeam::AfterPostPeerScan); let phase = Instant::now(); self.materialize_entry_state(&entry, &baseline).await?; EntryWork::add_phase(&entry.work.materialize_micros, phase); @@ -7959,6 +8006,7 @@ mod tests { entries: RwLock::new(HashMap::new()), watching: StdMutex::new(HashSet::new()), cancel: CancellationToken::new(), + seam_hook: StdMutex::new(None), }; let state = engine.read_durable_state("bus").unwrap().unwrap(); // The test hook pauses after the durable rename and before log removal. @@ -10718,4 +10766,334 @@ mod tests { ); } } + + // ---- issue 175: a rename between a scan and the materialization after it ---- + + /// A peer that behaves like the client half of a real exchange and, once, + /// lands a guarded inbound session on this engine during its unguarded + /// peer step. That inbound session is what first materializes the peer's + /// file on local disk, after the outbound pass captured its pre-peer view + /// and before its post-peer scan. + struct SeamRaceTransport { + engine: StdMutex>>>, + remote: Arc>, + inbound_during_peer_step: bool, + inbound_done: std::sync::atomic::AtomicBool, + } + + impl SeamRaceTransport { + fn new(remote: Arc>, inbound_during_peer_step: bool) -> Arc { + Arc::new(Self { + engine: StdMutex::new(None), + remote, + inbound_during_peer_step, + inbound_done: std::sync::atomic::AtomicBool::new(false), + }) + } + + fn attach(&self, engine: &Arc>) { + *self.engine.lock().unwrap() = Some(Arc::downgrade(engine)); + } + } + + impl SyncTransport for SeamRaceTransport { + async fn peers_for(&self, _peers: &SyncPeers) -> ResolvedPeers { + ResolvedPeers::all(vec![PeerRef { + key: "peer-a".to_string(), + id: "peer-a".to_string(), + roaming: false, + }]) + } + + async fn reconcile( + &self, + _peer: PeerRef, + name: String, + node: Arc>, + ) -> Result { + // The client half of a real exchange: adopt the peer's winning + // entries and pull the bytes behind them. + let (remote_manifest, remote_content) = { + let remote = self.remote.lock().await; + let manifest = remote.manifest().clone(); + let hashes: Vec = + manifest.present_paths().map(|(_, meta)| meta.hash).collect(); + (manifest, remote.gather_content(&hashes)) + }; + let pulled = { + let mut node = node.lock().await; + let pulled = node.adopt_from_peer(&remote_manifest); + for (_, bytes) in remote_content { + node.put_content(bytes); + } + pulled + }; + // Once, while this pass holds no operation guard: the same peer's + // own pass arrives as a guarded inbound session. It scans, adopts + // nothing new, and materializes the file adopted above. + if self.inbound_during_peer_step + && !self.inbound_done.swap(true, Ordering::AcqRel) + { + let engine = self + .engine + .lock() + .unwrap() + .clone() + .and_then(|weak| weak.upgrade()) + .expect("the transport is attached to its engine"); + let prepared = engine + .prepare_inbound(&name) + .await? + .expect("the entry exists"); + engine.complete_inbound(prepared).await?; + } + Ok(Reconciled { + pulled, + ..Reconciled::default() + }) + } + } + + async fn rename_race_fixture( + inbound_during_peer_step: bool, + ) -> ( + tempfile::TempDir, + PathBuf, + Arc>, + Arc>, + ) { + let dir = tempfile::tempdir().unwrap(); + let root = dir.path().join("resources"); + std::fs::create_dir_all(root.join("inbox")).unwrap(); + write_bus_sync(dir.path(), &root); + let mut remote = SyncNode::new(Author([2; 32])); + remote.local_write("inbox/job.md", b"concurrent update", 0, 0); + let remote = Arc::new(Mutex::new(remote)); + let transport = SeamRaceTransport::new(remote.clone(), inbound_during_peer_step); + let engine = SyncEngine::new( + FabricHome::new(dir.path()), + Author([1; 32]), + transport.clone(), + CancellationToken::new(), + ) + .await + .unwrap(); + transport.attach(&engine); + (dir, root, engine, remote) + } + + /// Rename `inbox/job.md` to `archive/job.md` the first time `seam` fires, + /// and record what the seam saw each time it fired: whether the source was + /// on disk, and whether the manifest held it Present. + fn rename_at_seam( + engine: &SyncEngine, + root: &Path, + node: &Arc>, + seam: PhaseSeam, + ) -> Arc)>>> { + let evidence = Arc::new(StdMutex::new(Vec::new())); + let hook_root = root.to_path_buf(); + let hook_node = node.clone(); + let hook_evidence = evidence.clone(); + engine.set_seam_hook(move |fired| { + if fired != seam { + return; + } + let mut evidence = hook_evidence.lock().unwrap(); + let on_disk = hook_root.join("inbox/job.md").is_file(); + let present = hook_node.try_lock().ok().map(|node| { + node.manifest() + .get("inbox/job.md") + .is_some_and(Entry::is_present) + }); + evidence.push((on_disk, present)); + if evidence.len() == 1 { + std::fs::create_dir_all(hook_root.join("archive")).unwrap(); + std::fs::rename( + hook_root.join("inbox/job.md"), + hook_root.join("archive/job.md"), + ) + .unwrap(); + } + }); + evidence + } + + async fn assert_rename_outcome( + node: &Arc>, + remote: &Arc>, + root: &Path, + ) { + let node = node.lock().await; + assert!( + matches!(node.manifest().get("inbox/job.md"), Some(Entry::Tombstone(_))), + "the rename source must be a tombstone, got {:?}", + node.manifest().get("inbox/job.md") + ); + assert!( + matches!(node.manifest().get("archive/job.md"), Some(Entry::Present(_))), + "the rename destination must be present, got {:?}", + node.manifest().get("archive/job.md") + ); + assert!( + !root.join("inbox/job.md").exists(), + "the rename source was written back to disk" + ); + assert_eq!( + std::fs::read(root.join("archive/job.md")).unwrap(), + b"concurrent update" + ); + // The peer converges on the same answer from this node's state alone. + let mut remote = remote.lock().await; + remote.adopt_from_peer(node.manifest()); + assert!( + matches!(remote.manifest().get("inbox/job.md"), Some(Entry::Tombstone(_))), + "the peer must adopt the source tombstone, got {:?}", + remote.manifest().get("inbox/job.md") + ); + assert!( + matches!(remote.manifest().get("archive/job.md"), Some(Entry::Present(_))), + "the peer must adopt the destination" + ); + } + + /// Issue 175. The outbound pass carries its pre-peer disk view across the + /// peer step and uses it as the protected view for the post-peer + /// materialization. A file that an inbound session materialized during the + /// peer step is absent from that view, so a rename landing between the + /// post-peer scan and its materialization reads as a never-seen remote + /// file, and the source is written back. Both paths end Present. + #[tokio::test] + async fn a_rename_between_the_post_peer_scan_and_its_materialization_does_not_split_the_file() + { + let (_dir, root, engine, remote) = rename_race_fixture(true).await; + let node = engine.node_for("bus").await.unwrap(); + let evidence = rename_at_seam(&engine, &root, &node, PhaseSeam::AfterPostPeerScan); + + engine.sync_once("bus").await.unwrap(); + engine.sync_once("bus").await.unwrap(); + + // The fixture must have reached the window: when the seam first fired, + // the peer's file was on disk and Present in the manifest. + let evidence = evidence.lock().unwrap().clone(); + assert_eq!( + evidence.first(), + Some(&(true, Some(true))), + "the seam did not see the peer's file on disk and Present: {evidence:?}" + ); + assert_rename_outcome(&node, &remote, &root).await; + } + + /// Control for the test above: the same rename one seam earlier, before + /// the post-peer scan, is recorded by that scan. This passes with or + /// without the fix, so a failure of the test above is the window itself. + #[tokio::test] + async fn a_rename_before_the_post_peer_scan_is_recorded_by_that_scan() { + let (_dir, root, engine, remote) = rename_race_fixture(true).await; + let node = engine.node_for("bus").await.unwrap(); + let evidence = rename_at_seam(&engine, &root, &node, PhaseSeam::BeforePostPeerScan); + + engine.sync_once("bus").await.unwrap(); + engine.sync_once("bus").await.unwrap(); + + let evidence = evidence.lock().unwrap().clone(); + assert_eq!(evidence.first(), Some(&(true, Some(true))), "{evidence:?}"); + assert_rename_outcome(&node, &remote, &root).await; + } + + /// Control against over-correction: a path adopted during the peer step + /// that never existed on local disk must still be materialized by the + /// post-peer materialization. Protecting every path in the manifest would + /// turn every remote-only file into a local delete. + #[tokio::test] + async fn a_path_adopted_during_the_peer_step_and_never_seen_locally_is_still_materialized() { + let (_dir, root, engine, _remote) = rename_race_fixture(false).await; + engine.sync_once("bus").await.unwrap(); + + assert_eq!( + std::fs::read(root.join("inbox/job.md")).unwrap(), + b"concurrent update" + ); + let node = engine.node_for("bus").await.unwrap(); + let node = node.lock().await; + assert!(matches!( + node.manifest().get("inbox/job.md"), + Some(Entry::Present(_)) + )); + } + + /// The inbound half of the same window. `complete_inbound` scans, then + /// materializes with the view captured before the wire session. A file + /// created during the session and removed between that scan and its + /// materialization is absent from the older view, so it is written back. + #[tokio::test] + async fn a_file_created_and_removed_inside_one_inbound_session_stays_removed() { + let (_dir, root, engine, _remote) = rename_race_fixture(false).await; + engine.sync_once("bus").await.unwrap(); + let node = engine.node_for("bus").await.unwrap(); + let fired = Arc::new(AtomicUsize::new(0)); + let hook_root = root.clone(); + let hook_fired = fired.clone(); + engine.set_seam_hook(move |seam| { + if seam == PhaseSeam::AfterInboundScan + && hook_fired.fetch_add(1, Ordering::AcqRel) == 0 + { + std::fs::remove_file(hook_root.join("inbox/draft.md")).unwrap(); + } + }); + + let prepared = engine.prepare_inbound("bus").await.unwrap().unwrap(); + // A file appears during the wire session. The completion scan records + // it, the seam removes it, and the materialization must not restore it. + std::fs::write(root.join("inbox/draft.md"), b"draft").unwrap(); + engine.complete_inbound(prepared).await.unwrap(); + + assert_eq!(fired.load(Ordering::Relaxed), 1, "the inbound seam did not fire exactly once"); + let node = node.lock().await; + assert!( + matches!(node.manifest().get("inbox/draft.md"), Some(Entry::Tombstone(_))), + "a file removed after the completion scan must be a tombstone, got {:?}", + node.manifest().get("inbox/draft.md") + ); + assert!( + !root.join("inbox/draft.md").exists(), + "the removed file was written back to disk" + ); + } + + /// The pre-peer half of the same window. `sync_once` captures its + /// protected view before the pre-peer scan, so a file created before that + /// scan and removed between the scan and its materialization is absent + /// from the protected view and is written back. + #[tokio::test] + async fn a_file_created_and_removed_around_the_pre_peer_scan_stays_removed() { + let (_dir, root, engine, _remote) = rename_race_fixture(false).await; + engine.sync_once("bus").await.unwrap(); + let node = engine.node_for("bus").await.unwrap(); + let fired = Arc::new(AtomicUsize::new(0)); + let hook_root = root.clone(); + let hook_fired = fired.clone(); + engine.set_seam_hook(move |seam| { + if seam == PhaseSeam::AfterPrePeerScan + && hook_fired.fetch_add(1, Ordering::AcqRel) == 0 + { + std::fs::remove_file(hook_root.join("inbox/draft.md")).unwrap(); + } + }); + + std::fs::write(root.join("inbox/draft.md"), b"draft").unwrap(); + engine.sync_once("bus").await.unwrap(); + + assert_eq!(fired.load(Ordering::Relaxed), 1, "the pre-peer seam did not fire exactly once"); + let node = node.lock().await; + assert!( + matches!(node.manifest().get("inbox/draft.md"), Some(Entry::Tombstone(_))), + "a file removed after the pre-peer scan must be a tombstone, got {:?}", + node.manifest().get("inbox/draft.md") + ); + assert!( + !root.join("inbox/draft.md").exists(), + "the removed file was written back to disk" + ); + } } From aa94381531a97623dfaf0e1fc945c245aaf01bb4 Mon Sep 17 00:00:00 2001 From: Nathan Herald Date: Sun, 6 Sep 2026 01:01:55 +0200 Subject: [PATCH 2/2] Protect exactly what the last scan saw when materializing A local rename could leave both the old and the new path present on every peer. The sync pass captured a disk view before its peer step and materialized with that view after it. An inbound session that ran during the peer step had already written the peer's file to disk, so the view lacked it. A rename landing between the post-peer scan and its materialization then read as a file this machine never had, and the old path was written back. The delete was never recorded. Every materialization now takes its protected view for itself, under the caller's operation guard, at the start of each attempt. That view is what the scan immediately before it published. The function no longer accepts a caller's view, so no caller can hand it a stale one. The same rule stops a file created and removed inside one pass or one inbound session from coming back. The seam hooks and the hook slot are compiled only in test builds. A non-test `cargo check --lib` passes, and the release binary holds no seam symbol while it holds 61 symbols for the surrounding functions. Fixes #175. --- CHANGELOG.md | 11 +++ src/sync/engine.rs | 162 +++++++++++++++++++++++++++------------------ 2 files changed, 109 insertions(+), 64 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 57ab4d1..924dc59 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -26,6 +26,17 @@ EXPERIMENTAL, so on-disk formats and the CLI may change without notice. ### Fixed +- **A local rename can no longer leave both the old and the new path present + on every peer.** A sync pass materialized with a disk view it captured before + its peer step. An inbound session that ran during that step had already + written the peer's file to disk, so the view lacked it, and a rename landing + between the post-peer scan and its materialization read as a file this + machine never had. The old path was written back and the delete was never + recorded. Every materialization now protects exactly what the scan before it + saw, and the function that materializes no longer accepts a caller's view. + The same rule stops a file created and removed inside one pass or one + inbound session from coming back. Issue #175. + - **`fabric doctor` no longer calls an unverified peer build current.** An unreachable peer now makes the build check `unknown`, including a roaming peer. The closing summary counts that unknown result as needing attention. diff --git a/src/sync/engine.rs b/src/sync/engine.rs index fca4e27..ee2efd3 100644 --- a/src/sync/engine.rs +++ b/src/sync/engine.rs @@ -788,7 +788,13 @@ impl PreparedInbound { /// A rename or delete that lands in one of these windows is invisible to the /// scan that just ran and visible to the materialization that follows. Tests /// pin an event to a seam so the window is exercised on demand rather than by -/// scheduler luck. Production builds carry the enum and an empty call. +/// scheduler luck. +/// +/// TEST BUILDS ONLY. This enum, the hook slot, `set_seam_hook`, `at_seam`, and +/// every call to `at_seam` are under `cfg(test)`. A production build never +/// compiles them, so a production reference to any of them fails to compile +/// rather than shipping a hook in the sync path. +#[cfg(test)] #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub(crate) enum PhaseSeam { /// `sync_once`, after the pre-peer scan and before its materialization. @@ -870,10 +876,6 @@ impl SyncEngine { } } - #[cfg(not(test))] - #[inline(always)] - fn at_seam(&self, _seam: PhaseSeam) {} - pub async fn load_from_config(&self) -> Result<()> { let book = SyncBook::load_path(self.paths.config_path())?; self.load_book(book).await @@ -1099,9 +1101,11 @@ impl SyncEngine { } /// Complete an inbound transaction while its entry operation guard is still - /// held. Disk changes that landed during the wire session are compared to - /// the pre-merge baseline: a vanished baseline Present is a local delete, - /// while a remote-only Present is materialized instead of tombstoned. + /// held. The completion scan records disk changes that landed during the + /// wire session, and the materialization that follows protects exactly what + /// that scan saw: a Present the scan observed and that is gone now is a + /// local delete, while a Present no scan observed is remote-only and is + /// materialized. The pre-merge baseline decides only whether to persist. pub(crate) async fn complete_inbound(&self, prepared: PreparedInbound) -> Result<()> { let PreparedInbound { entry, mode } = prepared; let PreparedInboundMode::Guarded { @@ -1120,8 +1124,9 @@ impl SyncEngine { // It is cheap now because scan_folder reuses recorded hashes for files // whose size and mtime are unchanged. self.scan_entry(&entry).await?; + #[cfg(test)] self.at_seam(PhaseSeam::AfterInboundScan); - self.materialize_entry_state(&entry, &baseline).await?; + self.materialize_entry_state(&entry).await?; let final_manifest_revision = entry.node.lock().await.manifest_revision(); let final_receipt_revision = entry.disk.lock().unwrap().receipt_revision; if final_manifest_revision != manifest_revision @@ -1232,18 +1237,23 @@ impl SyncEngine { entry.work.sync_passes.fetch_add(1, Ordering::Relaxed); // Never hold the local operation guard across a peer dial. If A and B // initiate together, retaining A while awaiting B's inbound guard (and - // vice versa) is a distributed lock inversion. Carry a pre-merge - // baseline across the unlocked network step instead. + // vice versa) is a distributed lock inversion. The baseline carried + // across the unlocked network step decides only whether the post-peer + // half has anything to persist. It is NOT the protected view for the + // post-peer materialization: an inbound session can materialize a file + // during the network step, and a view from before that step would + // read the file as never having been here. Issue #175. let (baseline, manifest_revision) = { let _operation = entry.operation.lock().await; - let protected = entry.disk.lock().unwrap().view(); + let before_scan = entry.disk.lock().unwrap().view(); let generation = entry.work.mutation_generation.load(Ordering::Acquire); let phase = Instant::now(); let scan_changed = self.scan_entry(&entry).await?; EntryWork::add_phase(&entry.work.scan_micros, phase); + #[cfg(test)] self.at_seam(PhaseSeam::AfterPrePeerScan); let phase = Instant::now(); - self.materialize_entry_state(&entry, &protected).await?; + self.materialize_entry_state(&entry).await?; EntryWork::add_phase(&entry.work.materialize_micros, phase); // A pass that changed nothing has nothing to record. This call was // unguarded while `prepare_inbound_entry` guards the identical one, @@ -1259,7 +1269,7 @@ impl SyncEngine { // - a legacy entry may have no state.json yet, so a first pass must // write even when it changed nothing. let observed_changed = - { entry.disk.lock().unwrap().receipt_revision != protected.receipt_revision }; + { entry.disk.lock().unwrap().receipt_revision != before_scan.receipt_revision }; if scan_changed || observed_changed || entry.work.durable_generation.load(Ordering::Acquire) != generation @@ -1397,13 +1407,15 @@ impl SyncEngine { EntryWork::add_phase(&entry.work.reconcile_micros, reconcile_phase); let _operation = entry.operation.lock().await; + #[cfg(test)] self.at_seam(PhaseSeam::BeforePostPeerScan); let phase = Instant::now(); self.scan_entry(&entry).await?; EntryWork::add_phase(&entry.work.scan_micros, phase); + #[cfg(test)] self.at_seam(PhaseSeam::AfterPostPeerScan); let phase = Instant::now(); - self.materialize_entry_state(&entry, &baseline).await?; + self.materialize_entry_state(&entry).await?; EntryWork::add_phase(&entry.work.materialize_micros, phase); self.sweep_entry_tombstones(&entry, &peers).await; let manifest_changed = entry.node.lock().await.manifest_revision() != manifest_revision; @@ -1423,8 +1435,7 @@ impl SyncEngine { return Ok(()); }; let _operation = entry.operation.lock().await; - let protected = entry.disk.lock().unwrap().view(); - self.materialize_entry_state(&entry, &protected).await?; + self.materialize_entry_state(&entry).await?; self.persist_entry(&entry).await } @@ -1498,11 +1509,24 @@ impl SyncEngine { anyhow::bail!("sync state changed during three scan attempts") } - async fn materialize_entry_state( - &self, - entry: &Arc, - protected: &DiskView, - ) -> Result<()> { + /// Materialize the entry's manifest onto disk. + /// + /// THE PROTECTED VIEW IS THE ONE THIS CALL TAKES FOR ITSELF, under the + /// caller's operation guard, and never one the caller carried in. Under + /// the guard that view is exactly what the scan immediately before this + /// call published, so a Present path that the scan saw on disk and that + /// is absent now is a local delete, and a Present path no scan saw is a + /// remote-only file to write. + /// + /// This used to take the view as a parameter, and `sync_once` passed the + /// view it captured BEFORE its peer step into the materialization AFTER + /// it. An inbound session that ran during the peer step had materialized + /// the peer's file by then, so that view lacked the file. A rename landing + /// between the post-peer scan and its materialization then read as a + /// never-seen remote file and the source was written back, leaving both + /// paths present on every peer. Issue #175. Removing the parameter is the + /// fix: no caller can hand this function a stale view again. + async fn materialize_entry_state(&self, entry: &Arc) -> Result<()> { let root = entry.config.folder.clone(); let policy = entry.policy; let cfg = entry.config.clone(); @@ -1518,18 +1542,20 @@ impl SyncEngine { }; let root = root.clone(); let cfg = cfg.clone(); - let protected = protected.clone(); let work = entry.work.clone(); let phase = tokio::task::spawn_blocking(move || { let mut node = staged_node; let mut disk = staged_disk; + // One view serves both roles. It is what the preceding scan + // published, taken again on every retry so a re-staged attempt + // cannot inherit an older one. let current = disk.view(); let _ = materialize_tracked_disk( &mut node, &root, &cfg, policy, - &protected, + ¤t, ¤t, &mut disk, Some((&work, generation)), @@ -1720,13 +1746,7 @@ impl SyncEngine { scan_cache: ScanCacheMapRef(&disk), peer_acks: &peer_acks_for_write, })?; - write_state_bytes_at( - &name, - &state_path, - &log_path, - &manifest_path, - &raw, - )?; + write_state_bytes_at(&name, &state_path, &log_path, &manifest_path, &raw)?; } else { let records: Vec = changed .iter() @@ -5736,11 +5756,7 @@ mod tests { .await .local_write("remote.md", b"daemon bytes", 0, 0); let generation = entry.work.mutation_generation.load(Ordering::Acquire); - let protected = DiskState::from_persisted(HashMap::new(), HashMap::new()).view(); - engine - .materialize_entry_state(&entry, &protected) - .await - .unwrap(); + engine.materialize_entry_state(&entry).await.unwrap(); engine.persist_entry(&entry).await.unwrap(); entry.work.mark_generation_durable(generation); entry.work.full_scans.store(0, Ordering::Relaxed); @@ -7994,10 +8010,8 @@ mod tests { // The parent deliberately owns the real state lease while this crash // helper stops at a write boundary. Use a test-only lease root so the // helper can exercise that boundary without becoming a second engine. - let owner_paths = SyncPaths::new( - home.syncs_path(), - home.root().join("sync-crash-test-owner"), - ); + let owner_paths = + SyncPaths::new(home.syncs_path(), home.root().join("sync-crash-test-owner")); let engine = SyncEngine { paths: SyncPaths::new(home.syncs_path(), home.root().join("sync")), _owner: SyncOwnerLease::acquire(&owner_paths).unwrap(), @@ -10498,7 +10512,7 @@ mod tests { // materialization must version this local write above the remote edit. let PreparedInbound { entry, mode } = prepared; let PreparedInboundMode::Guarded { - baseline, + baseline: _, manifest_revision: _, _waiter, _operation, @@ -10508,10 +10522,7 @@ mod tests { }; engine.scan_entry(&entry).await.unwrap(); std::fs::write(root.join("shared.md"), b"local after scan").unwrap(); - engine - .materialize_entry_state(&entry, &baseline) - .await - .unwrap(); + engine.materialize_entry_state(&entry).await.unwrap(); engine.persist_entry(&entry).await.unwrap(); assert_eq!( @@ -10816,8 +10827,10 @@ mod tests { let (remote_manifest, remote_content) = { let remote = self.remote.lock().await; let manifest = remote.manifest().clone(); - let hashes: Vec = - manifest.present_paths().map(|(_, meta)| meta.hash).collect(); + let hashes: Vec = manifest + .present_paths() + .map(|(_, meta)| meta.hash) + .collect(); (manifest, remote.gather_content(&hashes)) }; let pulled = { @@ -10831,9 +10844,7 @@ mod tests { // Once, while this pass holds no operation guard: the same peer's // own pass arrives as a guarded inbound session. It scans, adopts // nothing new, and materializes the file adopted above. - if self.inbound_during_peer_step - && !self.inbound_done.swap(true, Ordering::AcqRel) - { + if self.inbound_during_peer_step && !self.inbound_done.swap(true, Ordering::AcqRel) { let engine = self .engine .lock() @@ -10926,12 +10937,18 @@ mod tests { ) { let node = node.lock().await; assert!( - matches!(node.manifest().get("inbox/job.md"), Some(Entry::Tombstone(_))), + matches!( + node.manifest().get("inbox/job.md"), + Some(Entry::Tombstone(_)) + ), "the rename source must be a tombstone, got {:?}", node.manifest().get("inbox/job.md") ); assert!( - matches!(node.manifest().get("archive/job.md"), Some(Entry::Present(_))), + matches!( + node.manifest().get("archive/job.md"), + Some(Entry::Present(_)) + ), "the rename destination must be present, got {:?}", node.manifest().get("archive/job.md") ); @@ -10947,12 +10964,18 @@ mod tests { let mut remote = remote.lock().await; remote.adopt_from_peer(node.manifest()); assert!( - matches!(remote.manifest().get("inbox/job.md"), Some(Entry::Tombstone(_))), + matches!( + remote.manifest().get("inbox/job.md"), + Some(Entry::Tombstone(_)) + ), "the peer must adopt the source tombstone, got {:?}", remote.manifest().get("inbox/job.md") ); assert!( - matches!(remote.manifest().get("archive/job.md"), Some(Entry::Present(_))), + matches!( + remote.manifest().get("archive/job.md"), + Some(Entry::Present(_)) + ), "the peer must adopt the destination" ); } @@ -10964,8 +10987,7 @@ mod tests { /// post-peer scan and its materialization reads as a never-seen remote /// file, and the source is written back. Both paths end Present. #[tokio::test] - async fn a_rename_between_the_post_peer_scan_and_its_materialization_does_not_split_the_file() - { + async fn a_rename_between_the_post_peer_scan_and_its_materialization_does_not_split_the_file() { let (_dir, root, engine, remote) = rename_race_fixture(true).await; let node = engine.node_for("bus").await.unwrap(); let evidence = rename_at_seam(&engine, &root, &node, PhaseSeam::AfterPostPeerScan); @@ -11035,8 +11057,7 @@ mod tests { let hook_root = root.clone(); let hook_fired = fired.clone(); engine.set_seam_hook(move |seam| { - if seam == PhaseSeam::AfterInboundScan - && hook_fired.fetch_add(1, Ordering::AcqRel) == 0 + if seam == PhaseSeam::AfterInboundScan && hook_fired.fetch_add(1, Ordering::AcqRel) == 0 { std::fs::remove_file(hook_root.join("inbox/draft.md")).unwrap(); } @@ -11048,10 +11069,17 @@ mod tests { std::fs::write(root.join("inbox/draft.md"), b"draft").unwrap(); engine.complete_inbound(prepared).await.unwrap(); - assert_eq!(fired.load(Ordering::Relaxed), 1, "the inbound seam did not fire exactly once"); + assert_eq!( + fired.load(Ordering::Relaxed), + 1, + "the inbound seam did not fire exactly once" + ); let node = node.lock().await; assert!( - matches!(node.manifest().get("inbox/draft.md"), Some(Entry::Tombstone(_))), + matches!( + node.manifest().get("inbox/draft.md"), + Some(Entry::Tombstone(_)) + ), "a file removed after the completion scan must be a tombstone, got {:?}", node.manifest().get("inbox/draft.md") ); @@ -11074,8 +11102,7 @@ mod tests { let hook_root = root.clone(); let hook_fired = fired.clone(); engine.set_seam_hook(move |seam| { - if seam == PhaseSeam::AfterPrePeerScan - && hook_fired.fetch_add(1, Ordering::AcqRel) == 0 + if seam == PhaseSeam::AfterPrePeerScan && hook_fired.fetch_add(1, Ordering::AcqRel) == 0 { std::fs::remove_file(hook_root.join("inbox/draft.md")).unwrap(); } @@ -11084,10 +11111,17 @@ mod tests { std::fs::write(root.join("inbox/draft.md"), b"draft").unwrap(); engine.sync_once("bus").await.unwrap(); - assert_eq!(fired.load(Ordering::Relaxed), 1, "the pre-peer seam did not fire exactly once"); + assert_eq!( + fired.load(Ordering::Relaxed), + 1, + "the pre-peer seam did not fire exactly once" + ); let node = node.lock().await; assert!( - matches!(node.manifest().get("inbox/draft.md"), Some(Entry::Tombstone(_))), + matches!( + node.manifest().get("inbox/draft.md"), + Some(Entry::Tombstone(_)) + ), "a file removed after the pre-peer scan must be a tombstone, got {:?}", node.manifest().get("inbox/draft.md") );