From 4757b957d29cdc509c1daae39a6f73503e0e9924 Mon Sep 17 00:00:00 2001 From: Nils Pukropp Date: Sun, 4 Oct 2026 21:01:38 +0200 Subject: [PATCH] fix(fuse): end a mirror pass at the stop (B169) A stop that came while a mirror pass was uploading let the pass go on to the next level of folders until the stop gave up on it and the runtime's end cancelled the uploads under it. Once the daemon is stopping, a batch starts none of its waiting ops, lets the ones sending finish, and ends the pass; the remote walk and a new pass stop too. --- crates/pdfs-fuse/src/sim/daemon.rs | 71 ++++++++++++++++++++++++++++++ crates/pdfs-fuse/src/sync.rs | 55 ++++++++++++++++++----- crates/pdfs-fuse/src/sync/model.rs | 4 ++ docs/BUGS.md | 26 +++++++++++ docs/CHANGELOG.md | 2 + docs/MILESTONE-3.0.0-STATUS.md | 5 ++- 6 files changed, 149 insertions(+), 14 deletions(-) diff --git a/crates/pdfs-fuse/src/sim/daemon.rs b/crates/pdfs-fuse/src/sim/daemon.rs index 3124f79..28146cf 100644 --- a/crates/pdfs-fuse/src/sim/daemon.rs +++ b/crates/pdfs-fuse/src/sim/daemon.rs @@ -649,6 +649,77 @@ mod tests { let _ = std::fs::remove_dir_all(&dir); } + /// The database of a daemon just stopped, once the threads it left in a + /// request have let go of it, as [`Daemon::start`] waits for them. + fn reopen(dir: &Path) -> Db { + let mut db = None; + assert!(wait_until(LAST_RUN_DEADLINE, || { + db = Db::open(&dir.join("pdfs.db")).ok(); + db.is_some() + })); + db.unwrap() + } + + #[test] + #[ignore = "mounts FUSE: run with `cargo test -p pdfs-fuse sim:: -- --ignored`"] + fn a_stop_ends_a_mirror_pass_after_the_uploads_it_is_sending() { + // A stop that came while a mirror pass was uploading waited for it, + // and the pass went on to the next level of folders until the stop + // gave up on it and ended the runtime under it (B169). + let drive = FakeDrive::new(); + let dir = scratch("mirror-stop"); + let daemon = Daemon::start(&dir, drive.client(1, Faults::lan())).unwrap(); + std::fs::create_dir(daemon.mountpoint.join("mirrored")).unwrap(); + assert!(wait_until(Duration::from_secs(30), || { + daemon.pending().is_ok_and(|items| items.is_empty()) + && drive.lookup("mirrored").is_some() + })); + let remote = drive.lookup("mirrored").unwrap(); + assert!(matches!(daemon.stop(), Ok(MountOutcome::Shutdown))); + + let local = dir.join("local"); + std::fs::create_dir_all(local.join("deeper")).unwrap(); + std::fs::write(local.join("first"), b"abc").unwrap(); + std::fs::write(local.join("deeper/second"), b"def").unwrap(); + let folder = reopen(&dir) + .sync_folder_add(local.to_str().unwrap(), &remote.to_string(), "") + .unwrap(); + let client = drive.client(1, Faults::lan()); + let upload = client.hold_next_create(); + let daemon = Daemon::start(&dir, client).unwrap(); + assert!(wait_until(Duration::from_secs(30), || upload.reached())); + let socket = daemon.socket.clone(); + let stopping = std::thread::spawn(move || daemon.stop()); + assert!(wait_until(Duration::from_secs(30), || !socket.exists())); + drop(upload); + assert!(matches!( + stopping.join().unwrap(), + Ok(MountOutcome::Shutdown) + )); + + assert!(drive.lookup("mirrored/first").is_some()); + assert!(drive.lookup("mirrored/deeper").is_some()); + assert!(drive.lookup("mirrored/deeper/second").is_none()); + let logged = take_logged(); + assert!( + logged + .iter() + .any(|line| line.contains("pass cut short by the stop")) + ); + assert!( + !logged + .iter() + .any(|line| line.contains("did not finish in time")), + "{logged:#?}" + ); + let db = reopen(&dir); + assert_ne!( + db.sync_folder_get(folder).unwrap().unwrap().state, + "syncing" + ); + let _ = std::fs::remove_dir_all(&dir); + } + #[test] #[ignore = "mounts FUSE: run with `cargo test -p pdfs-fuse sim:: -- --ignored`"] fn a_file_removed_with_its_folder_while_uploading_stays_removed() { diff --git a/crates/pdfs-fuse/src/sync.rs b/crates/pdfs-fuse/src/sync.rs index 8425a51..6c7a7be 100644 --- a/crates/pdfs-fuse/src/sync.rs +++ b/crates/pdfs-fuse/src/sync.rs @@ -102,6 +102,9 @@ impl Core { /// One reconcile pass over a mirror folder, updating its `state` column. fn reconcile_pass(&self, folder: &StoredSyncFolder) { + if self.shutdown.is_stopping() { + return; + } // Hold the folder's lock for the whole pass so a mode switch can't evict the // local tree (and mount FUSE over it) while we walk and upload it. let lock = self.sync_lock(folder.id); @@ -148,8 +151,6 @@ impl Core { self.push_pass(folder.id, &local_root, &remote_root, &rules) } else { match self.do_reconcile(folder.id, &local_root, &remote_root, &rules) { - Ok(outcome) => Ok(outcome), - Err(PassAbort::Failed(e)) => Err(e), // The user asked for on-demand while this pass was running. Rather // than make them wait out a walk-and-download whose results are // about to be deleted, drop it here and do the only part that still @@ -164,6 +165,7 @@ impl Core { self.progress_begin(folder.id); self.push_pass(folder.id, &local_root, &remote_root, &rules) } + result => result, } }; self.progress_end(folder.id); @@ -216,7 +218,15 @@ impl Core { ); } } - Err(e) => { + // A push pass is never interrupted, so only the stop gets here: the + // folder keeps the state it had, which the next start's pass updates. + Err(PassAbort::Stopped | PassAbort::Interrupted) => { + info!(id = folder.id, "sync: pass cut short by the stop"); + let _ = self + .db + .sync_folder_set_state(folder.id, ¤t.state, current.last_sync); + } + Err(PassAbort::Failed(e)) => { warn!(id = folder.id, error = %e, "sync: reconcile failed"); let _ = self .db @@ -294,7 +304,7 @@ impl Core { local_root: &Path, remote_root: &NodeUid, rules: &IgnoreRules, - ) -> Result { + ) -> Result { let baseline = self .db .sync_entries(folder_id) @@ -400,7 +410,7 @@ impl Core { &mut remote_dirs, std::mem::take(&mut batch), &mut outcome, - ); + )?; batch_depth = depth; } let item = &local[rel]; @@ -498,7 +508,7 @@ impl Core { &mut remote_dirs, std::mem::take(&mut batch), &mut outcome, - ); + )?; // Paths the baseline knows that are no longer on disk: the user deleted them // locally, and the deletion has to reach Drive before the mount starts @@ -625,7 +635,7 @@ impl Core { &mut remote_dirs, std::mem::take(&mut batch), &mut outcome, - ); + )?; batch_depth = depth; // A depth boundary is the pass's natural checkpoint — everything // queued has been applied, so stopping here leaves nothing half done. @@ -813,7 +823,7 @@ impl Core { &mut remote_dirs, std::mem::take(&mut batch), &mut outcome, - ); + )?; // Deferred folder deletions, deepest first. delete_local_dirs.sort_by_key(|p| std::cmp::Reverse(p.matches('/').count())); @@ -941,6 +951,9 @@ impl Core { // is safe precisely because the walk has applied nothing — a *partial* // remote map, on the other hand, would be read as "the remote deleted // everything we haven't reached yet", so it must never reach the diff. + if self.shutdown.is_stopping() { + return Err(PassAbort::Stopped); + } if self.ondemand_queued(folder_id) { return Err(PassAbort::Interrupted); } @@ -1051,6 +1064,9 @@ impl Core { /// `remote_dirs` (so deeper paths resolve their parent), and what each op did /// is tallied into `outcome`. Driven by `block_on` from the (non-runtime) sync /// engine thread, spawning the tasks onto the shared runtime. + /// + /// Fails with [`PassAbort::Stopped`] once the daemon is stopping, after the + /// ops already sending have finished. fn flush_batch( &self, folder_id: i64, @@ -1058,9 +1074,9 @@ impl Core { remote_dirs: &mut HashMap, batch: Vec, outcome: &mut Outcome, - ) { + ) -> Result<(), PassAbort> { if batch.is_empty() { - return; + return Ok(()); } self.progress_queued(folder_id, batch.len()); let core = self.clone(); @@ -1074,14 +1090,22 @@ impl Core { let root = root.clone(); set.spawn(async move { let _permit = sem.acquire_owned().await.expect("semaphore not closed"); - core.apply_pending(folder_id, &root, op).await + // An op still waiting for its turn when the stop comes is + // not started; the ones already sending finish. + if core.shutdown.is_stopping() { + return None; + } + Some(core.apply_pending(folder_id, &root, op).await) }); } let mut out = Vec::new(); let mut join_errors = 0usize; while let Some(joined) = set.join_next().await { match joined { - Ok(result) => out.push(result), + Ok(Some(result)) => out.push(result), + Ok(None) => {} + // The runtime ends with the daemon and cancels what is left. + Err(e) if e.is_cancelled() => {} Err(e) => { join_errors += 1; warn!(error = %e, "sync: task panicked"); @@ -1106,6 +1130,13 @@ impl Core { } } } + // The pass goes no further: the next depth's paths may need a folder + // this batch did not get to make, and went on to fail one by one + // (`docs/BUGS.md` B169). + if self.shutdown.is_stopping() { + return Err(PassAbort::Stopped); + } + Ok(()) } /// Apply one [`Pending`] op (async, so it can run concurrently in a batch), diff --git a/crates/pdfs-fuse/src/sync/model.rs b/crates/pdfs-fuse/src/sync/model.rs index afb55e3..059ecc3 100644 --- a/crates/pdfs-fuse/src/sync/model.rs +++ b/crates/pdfs-fuse/src/sync/model.rs @@ -227,6 +227,10 @@ pub(super) enum PassAbort { Interrupted, /// The pass could not establish its diff (a walk or the baseline load failed). Failed(String), + /// The daemon is stopping. What the pass sent before the stop has landed or + /// will be cancelled with the runtime; nothing more is sent, and the next + /// start's pass picks up from the baseline. + Stopped, } impl From for PassAbort { diff --git a/docs/BUGS.md b/docs/BUGS.md index a90552d..9ae422d 100644 --- a/docs/BUGS.md +++ b/docs/BUGS.md @@ -12,6 +12,32 @@ Conventions: --- +## B169 — A stop lets a mirror pass go on until the runtime ends under it + +**Status:** Fixed. +**Found:** 2026-10-04, in the journal of the account run on the working tree after B165. A +stop that came while a mirror pass was uploading logged "background workers did not finish in +time" for `pdfs-sync`, then "sync: task panicked … was cancelled" and a run of "no remote +parent" warnings: the pass had gone on to deeper folders while the stop waited for it, and the +runtime's end cancelled the uploads under it. + +**Where:** `flush_batch`, `walk_remote` and `reconcile_pass` in `crates/pdfs-fuse/src/sync.rs`. + +**Cause.** The mirror pass never asked whether the daemon was stopping. It sends its work a +level of folders at a time and went on to the next level after the stop, until the stop gave up +on it at its 10 s limit (B165). + +**Fix.** Once the daemon is stopping, a batch starts none of its ops still waiting for their +turn, lets the ones already sending finish, and ends the pass. So do the remote walk and the +start of a pass. The folder keeps the state it had before the pass; the next start's pass picks +up from the baseline. + +**Test:** `a_stop_ends_a_mirror_pass_after_the_uploads_it_is_sending` in `sim/daemon.rs` holds a +mirror upload, stops the daemon, then lets the upload through, and checks that the file in the +deeper folder was not sent and the stop did not run into its limit. + +--- + ## B168 — A file removed with its folder while uploading is sent again, to the root **Status:** Fixed. diff --git a/docs/CHANGELOG.md b/docs/CHANGELOG.md index a596c28..e1958f2 100644 --- a/docs/CHANGELOG.md +++ b/docs/CHANGELOG.md @@ -66,6 +66,8 @@ by its local id, in the queue, the stored tree and pins alike. took close to 2 s with 37,000 files at every start (B167). - **Files deleted with their folder while uploading stay deleted.** The mount no longer tries to upload them to the top of your Drive instead (B168). +- **Stopping the daemon during a folder sync no longer waits for the sync.** The uploads in + progress finish, and the rest waits for the next start (B169). - **A file renamed just as its upload finished keeps its new name.** The mount no longer showed the old name, and a second rename no longer failed with "No such file or directory" (B152). - **A change that waits for a name to be freed no longer waits seconds after it is.** A rename or diff --git a/docs/MILESTONE-3.0.0-STATUS.md b/docs/MILESTONE-3.0.0-STATUS.md index b5e644a..669b289 100644 --- a/docs/MILESTONE-3.0.0-STATUS.md +++ b/docs/MILESTONE-3.0.0-STATUS.md @@ -9,7 +9,7 @@ Tracks `docs/MILESTONE-3.0.0.md`. Update it with every commit that moves a phase - **Phase 2**: local ids, inode = local id, one tree for every mount (2.10.0). - **Phase 3, first part**: mkdir, create, rename, unlink and rmdir on the mount are recorded locally and sent from the queue; `"local_first": false` brings back the old path (5e0f30c). -- Fixes from the simulation and account runs since then: B127 and B141 to B168. +- Fixes from the simulation and account runs since then: B127 and B141 to B169. - **Sync issues, first cut** (§6.2): refusals sorted into issues, shown at once in the app, tray and CLI; Export button and `pdfs sync export`; `pdfs sync issues`. @@ -109,7 +109,8 @@ Tracks `docs/MILESTONE-3.0.0.md`. Update it with every commit that moves a phase running, which the run's queue wait takes for an empty queue (the script's side). Looking into it found the start spending 1.7 s rebuilding listings (B167, fixed). Its journal check caught a file removed with its folder while uploading being - sent again, to the root (B168, fixed). + sent again, to the root (B168, fixed), and a stop that let a mirror pass go on + until the runtime ended under it (B169, fixed). 5. **Release**: 2.10.0 is tagged; the 2.9.0 tag still waits for the user's go-ahead. ## After 3.0.0