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

@@ -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() {

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, &current.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<Outcome, String> {
+    ) -> Result<Outcome, PassAbort> {
         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<String, NodeUid>,
         batch: Vec<Pending>,
         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),

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<String> for PassAbort {