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 7f15451..ee2efd3 100644 --- a/src/sync/engine.rs +++ b/src/sync/engine.rs @@ -783,6 +783,33 @@ 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. +/// +/// 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. + 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 +827,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 +854,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 +863,19 @@ 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); + } + } + pub async fn load_from_config(&self) -> Result<()> { let book = SyncBook::load_path(self.paths.config_path())?; self.load_book(book).await @@ -1056,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 { @@ -1077,7 +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?; - self.materialize_entry_state(&entry, &baseline).await?; + #[cfg(test)] + self.at_seam(PhaseSeam::AfterInboundScan); + 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 @@ -1188,17 +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, @@ -1214,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 @@ -1352,11 +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; @@ -1376,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 } @@ -1451,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(); @@ -1471,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)), @@ -1673,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() @@ -5689,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); @@ -7947,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(), @@ -7959,6 +8020,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. @@ -10450,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, @@ -10460,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!( @@ -10718,4 +10777,357 @@ 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" + ); + } }