release(1.8.1)

crates/pdfs-fuse/src/background.rs

@@ -1,6 +1,7 @@
 //! Remote-event invalidation and local-index background workers.
 
 use super::*;
+use crate::shutdown::Shutdown;
 
 /// Drop `uid` from one mount's tree and tell that mount's kernel session the
 /// entry is gone. Shared by the trash and delete paths, which differ only in
@@ -437,6 +438,7 @@ pub(super) fn run_local_index(
     indexing: Arc<AtomicBool>,
     transfers: Arc<TransferRegistry>,
     mountpoint: PathBuf,
+    shutdown: Arc<Shutdown>,
 ) {
     loop {
         let age = db.local_indexed_at().ok().flatten();
@@ -445,7 +447,11 @@ pub(super) fn run_local_index(
         if stale {
             scan_local_once(&db, &indexing, &transfers, &mountpoint);
         }
-        std::thread::sleep(LOCAL_INDEX_CHECK);
+        // Interruptible, so teardown joins this thread instead of leaving it to
+        // keep rewriting the index for a mount that is gone (bugs.md B44).
+        if !shutdown.sleep(LOCAL_INDEX_CHECK) {
+            return;
+        }
     }
 }
 

crates/pdfs-fuse/src/control.rs

@@ -1084,6 +1084,12 @@ pub(crate) fn run_control_socket(
 ) {
     info!("control socket listening");
     for conn in listener.incoming() {
+        // Checked on every accept, including the deliberate poke teardown sends
+        // to wake this thread out of a blocking accept (bugs.md B44).
+        if core.shutdown.is_stopping() {
+            info!("control socket closing");
+            return;
+        }
         match conn {
             Ok(stream) => {
                 let Some(permit) = ControlHandlerPermit::acquire() else {

crates/pdfs-fuse/src/drain.rs

@@ -260,6 +260,13 @@ impl Core {
     /// [`Db::claim_next_due_op`]: pdfs_core::db::Db::claim_next_due_op
     pub(crate) fn run_pending_drain(&self, primary: bool) {
         loop {
+            // Between ops, never inside one: an op that has started is either
+            // retired or released, so stopping here cannot leave the queue with
+            // a row claimed by a worker that no longer exists (bugs.md B44).
+            if self.shutdown.is_stopping() {
+                debug!(primary, "drain worker stopping");
+                return;
+            }
             let now = now_millis();
             // One row, chosen and *claimed* by the database. Reading the whole
             // queue to pick one op made a long queue quadratic to drain, and

crates/pdfs-fuse/src/filesystem.rs

@@ -1,6 +1,7 @@
 //! Kernel-facing FUSE callback adapter.
 
 use super::*;
+use crate::state::DirListing;
 
 impl Filesystem for ProtonFs {
     /// Negotiate kernel-side behaviour once, before any other callback.
@@ -162,15 +163,46 @@ impl Filesystem for ProtonFs {
         }
     }
 
+    /// Freeze this directory's listing for the life of the handle.
+    ///
+    /// `readdir` is paged, and its continuation offset only means anything
+    /// against a listing that does not move underneath it. Taking the snapshot
+    /// here — once per `opendir`, which the kernel pairs exactly with one
+    /// `releasedir` — is what makes the offsets stable, so a create or a trash
+    /// during an enumeration can no longer make the caller skip or repeat an
+    /// entry (bugs.md B43).
+    ///
+    /// Enumerating a cold folder is a remote call, so it goes to a worker for
+    /// the same reason `lookup` and `readdir` do.
+    fn opendir(&self, _req: &Request, ino: INodeNo, _flags: OpenFlags, reply: ReplyOpen) {
+        let ino = ino.0;
+        if self.core.children_cached(ino) {
+            self.serve_opendir(ino, reply);
+            return;
+        }
+        let fs = self.clone();
+        self.core
+            .workers
+            .run(Lane::Meta, move || fs.serve_opendir(ino, reply));
+    }
+
     fn readdir(
         &self,
         _req: &Request,
         ino: INodeNo,
-        _fh: FileHandle,
+        fh: FileHandle,
         offset: u64,
         reply: ReplyDirectory,
     ) {
         let ino = ino.0;
+        // The snapshot is the whole point: served from it, this is a pure
+        // in-memory walk with no chance of a remote call, so it never needs a
+        // worker. Only a handle whose `opendir` failed to snapshot (or a kernel
+        // that issued `readdir` without one) falls back to the live path.
+        if let Some(listing) = self.core.state().dir_snapshots.get(&fh.0).cloned() {
+            reply_listing(&listing, offset, reply);
+            return;
+        }
         // Same split as `lookup`: only the cold, remote-enumerating path pays
         // for a worker.
         if self.core.children_cached(ino) {
@@ -183,6 +215,18 @@ impl Filesystem for ProtonFs {
             .run(Lane::Meta, move || fs.serve_readdir(ino, offset, reply));
     }
 
+    fn releasedir(
+        &self,
+        _req: &Request,
+        _ino: INodeNo,
+        fh: FileHandle,
+        _flags: OpenFlags,
+        reply: ReplyEmpty,
+    ) {
+        self.core.state().dir_snapshots.remove(&fh.0);
+        reply.ok();
+    }
+
     /// A read-only open is a map lookup and answers inline; a write open has to
     /// allocate a scratch file and, when a queued revision is the base, duplicate
     /// a blob that is the whole file — so it is served from a worker rather than
@@ -912,6 +956,58 @@ pub struct ProtonFs {
     gid: u32,
 }
 
+/// How many times a write `open` will re-read its base after a `release`
+/// published a newer queued revision underneath it (bugs.md B32).
+///
+/// One retry covers the race, which needs a close and an open of the same file
+/// to overlap on different threads. More than that is not contention, it is a
+/// file being republished continuously, and looping on it would keep a FUSE
+/// worker busy indefinitely.
+const OPEN_BASE_ATTEMPTS: u32 = 3;
+
+/// One directory's entries as `readdir` reports them: `.` and `..` first, then
+/// the interned children in `children` order.
+///
+/// Split out because it is built from two places that must agree — `opendir`
+/// freezes it, and the fallback `readdir` builds it live — and a divergence
+/// between them would show up as entries appearing or vanishing depending on
+/// whether the kernel happened to send an `opendir`.
+fn build_listing(st: &State, ino: u64) -> DirListing {
+    let parent = st.entries.get(&ino).map_or(ROOT_INO, |e| e.parent);
+    // "." and ".." occupy offsets 0 and 1; real children follow.
+    let mut listing = vec![
+        (ino, true, ".".to_string()),
+        (parent, true, "..".to_string()),
+    ];
+    if let Some(kids) = st.children.get(&ino) {
+        for &kid in kids {
+            if let Some(e) = st.entries.get(&kid) {
+                listing.push((kid, e.node.is_folder(), e.node.name.clone()));
+            }
+        }
+    }
+    listing
+}
+
+/// Fill a `readdir` reply from `listing`, resuming at `offset`.
+///
+/// The cookie handed back is the index of the *next* entry, so a resumed call
+/// continues exactly where the last one stopped — which is only true because
+/// the listing behind those indexes is a snapshot (bugs.md B43).
+fn reply_listing(listing: &DirListing, offset: u64, mut reply: ReplyDirectory) {
+    for (i, (ino, is_dir, name)) in listing.iter().enumerate().skip(offset as usize) {
+        let ft = if *is_dir {
+            FileType::Directory
+        } else {
+            FileType::RegularFile
+        };
+        if reply.add(INodeNo(*ino), (i + 1) as u64, ft, name) {
+            break;
+        }
+    }
+    reply.ok();
+}
+
 pub(crate) fn access_allowed(is_dir: bool, access: Access, mask: AccessFlags) -> bool {
     (!mask.contains(AccessFlags::W_OK) || access.writable())
         && (!mask.contains(AccessFlags::X_OK) || is_dir)
@@ -1027,40 +1123,39 @@ impl ProtonFs {
     }
 
     /// The body of [`Filesystem::readdir`], on whichever thread ends up serving it.
-    fn serve_readdir(&self, ino: u64, offset: u64, mut reply: ReplyDirectory) {
+    /// The body of [`Filesystem::opendir`]: enumerate the folder if it is cold,
+    /// then publish one frozen listing under a fresh handle.
+    ///
+    /// A failure to enumerate is reported as a failed `opendir` rather than
+    /// deferred to the first `readdir` — the caller gets the errno at the syscall
+    /// that can still act on it.
+    fn serve_opendir(&self, ino: u64, reply: ReplyOpen) {
+        if let Err(e) = self.core.ensure_children(ino) {
+            reply.error(e);
+            return;
+        }
+        let mut st = self.core.state();
+        let listing = Arc::new(build_listing(&st, ino));
+        // Shares the counter with write handles, so a directory handle can never
+        // collide with a file handle (both are `fh` values in one kernel space).
+        let fh = st.next_fh;
+        st.next_fh += 1;
+        st.dir_snapshots.insert(fh, listing);
+        drop(st);
+        reply.opened(FileHandle(fh), FopenFlags::empty());
+    }
+
+    /// The live, un-snapshotted `readdir`. Reached only by a handle that has no
+    /// snapshot — see [`Filesystem::readdir`].
+    fn serve_readdir(&self, ino: u64, offset: u64, reply: ReplyDirectory) {
         if let Err(e) = self.core.ensure_children(ino) {
             reply.error(e);
             return;
         }
         let st = self.core.state();
-        let parent = st.entries.get(&ino).map_or(ROOT_INO, |e| e.parent);
-
-        // "." and ".." occupy offsets 0 and 1; real children follow.
-        let mut listing: Vec<(u64, FileType, String)> = vec![
-            (ino, FileType::Directory, ".".to_string()),
-            (parent, FileType::Directory, "..".to_string()),
-        ];
-        if let Some(kids) = st.children.get(&ino) {
-            for &kid in kids {
-                if let Some(e) = st.entries.get(&kid) {
-                    let ft = if e.node.is_folder() {
-                        FileType::Directory
-                    } else {
-                        FileType::RegularFile
-                    };
-                    listing.push((kid, ft, e.node.name.clone()));
-                }
-            }
-        }
+        let listing = build_listing(&st, ino);
         drop(st);
-
-        for (i, (ino, ft, name)) in listing.into_iter().enumerate().skip(offset as usize) {
-            // The stored offset is that of the *next* entry to resume at.
-            if reply.add(INodeNo(ino), (i + 1) as u64, ft, &name) {
-                break;
-            }
-        }
-        reply.ok();
+        reply_listing(&listing, offset, reply);
     }
 
     fn attr(&self, ino: u64, node: &Node, access: Access) -> FileAttr {
@@ -1142,66 +1237,101 @@ impl ProtonFs {
         // any network access. An incomplete blob still has gaps referring to
         // the remote base; stacking another write on it cannot be represented
         // safely by WriteHandle, so fail the open rather than risk corruption.
-        let pending_base = self.core.pending.lock().get(&uid).cloned();
-        if pending_base.as_ref().is_some_and(|p| !p.meta.complete) {
-            if let Some(entry) = self.core.state().entries.get_mut(&ino.0) {
-                entry.open_count = entry.open_count.saturating_sub(1);
-            }
-            error!(%uid, "refusing write over incomplete queued revision");
-            reply.error(Errno::EIO);
-            return;
-        }
-        let (file, path) = match self.core.cache.create_scratch() {
-            Ok(x) => x,
-            Err(e) => {
-                if let Some(entry) = self.core.state().entries.get_mut(&ino.0) {
-                    entry.open_count = entry.open_count.saturating_sub(1);
+        //
+        // Reading that blob and installing the handle cannot be done under one
+        // lock — the copy is the size of the file — so the base is re-checked at
+        // the moment the handle goes in, and a `release` that published a newer
+        // revision in between sends this around again. Without the re-check the
+        // new handle carried the *older* base, and its own release published
+        // content that silently discarded the revision closed a moment earlier
+        // (bugs.md B32). A second attempt is already unlikely and a third would
+        // mean something is republishing continuously, so give up rather than
+        // spin: the caller's `open` fails instead of its data being lost.
+        //
+        // `reply` is consumed by whichever arm answers, so the loop yields an
+        // outcome and the answer is sent once, after it.
+        let outcome = 'install: {
+            for attempt in 0..OPEN_BASE_ATTEMPTS {
+                let pending_base = self.core.pending.lock().get(&uid).cloned();
+                if pending_base.as_ref().is_some_and(|p| !p.meta.complete) {
+                    error!(%uid, "refusing write over incomplete queued revision");
+                    break 'install Err(Errno::EIO);
                 }
-                error!(%uid, error = %e, "create scratch file failed");
-                reply.error(Errno::EIO);
-                return;
+                let (file, path) = match self.core.cache.create_scratch() {
+                    Ok(x) => x,
+                    Err(e) => {
+                        error!(%uid, error = %e, "create scratch file failed");
+                        break 'install Err(Errno::EIO);
+                    }
+                };
+                let mut initial_written = Intervals::default();
+                if let Some(pending) = &pending_base {
+                    // Reflink where the filesystem can: this blob is the whole
+                    // file, and a byte-for-byte copy of a multi-GB pending
+                    // revision is the kind of work `open` must not do
+                    // (`ContentCache::clone_or_copy`).
+                    if let Err(e) = ContentCache::clone_or_copy(&pending.path, &path) {
+                        let _ = std::fs::remove_file(&path);
+                        error!(%uid, source = %pending.path.display(), error = %e,
+                            "copy queued revision into write scratch failed");
+                        break 'install Err(Errno::EIO);
+                    }
+                    initial_written.add(0, pending.meta.len);
+                }
+                // Sampled *before* the state lock, never under it: no site in
+                // the daemon holds `pending` and `state` at once (see
+                // `stamp_pending_sizes`) and this must not become the first.
+                // What that costs is a window of a few instructions instead of
+                // a lock-tight comparison — against the multi-gigabyte copy
+                // above, which is the window that actually loses data.
+                let current = self.core.pending_blob(&uid);
+                let mut st = self.core.state();
+                // Only a handle this open is about to *create* can carry a stale
+                // base; if one already exists, that handle is authoritative and
+                // the scratch prepared above is discarded either way.
+                let creating = !st.active_writes.contains_key(&ino.0);
+                if creating && current != pending_base.map(|p| p.path) {
+                    drop(st);
+                    let _ = std::fs::remove_file(&path);
+                    warn!(%uid, attempt, "queued revision changed while opening for write; retrying");
+                    continue;
+                }
+                let fh = st.next_fh;
+                st.next_fh += 1;
+                let aw = st.active_writes.entry(ino.0).or_insert_with(|| {
+                    WriteHandle {
+                        ino: ino.0,
+                        uid: uid.clone(),
+                        file: Arc::new(file),
+                        path,
+                        written: initial_written,
+                        // Starts at the current size; reads in [0, base_size)
+                        // come from the base until overwritten.
+                        len: base_size,
+                        base_size,
+                        base_mtime,
+                        base_revision_id,
+                        dirty: false,
+                        open_count: 0,
+                    }
+                });
+                aw.open_count += 1;
+                st.handles.insert(fh, ino.0);
+                break 'install Ok(fh);
             }
+            error!(%uid, "queued revision kept changing while opening for write");
+            Err(Errno::EIO)
         };
-        let mut initial_written = Intervals::default();
-        if let Some(pending) = &pending_base {
-            // Reflink where the filesystem can: this blob is the whole file, and
-            // a byte-for-byte copy of a multi-GB pending revision is the kind of
-            // work `open` must not do (`ContentCache::clone_or_copy`).
-            if let Err(e) = ContentCache::clone_or_copy(&pending.path, &path) {
+        match outcome {
+            Ok(fh) => reply.opened(FileHandle(fh), FopenFlags::empty()),
+            Err(error) => {
+                // The lookup count was raised before any of this could fail.
                 if let Some(entry) = self.core.state().entries.get_mut(&ino.0) {
                     entry.open_count = entry.open_count.saturating_sub(1);
                 }
-                let _ = std::fs::remove_file(&path);
-                error!(%uid, source = %pending.path.display(), error = %e,
-                    "copy queued revision into write scratch failed");
-                reply.error(Errno::EIO);
-                return;
+                reply.error(error);
             }
-            initial_written.add(0, pending.meta.len);
         }
-        let mut st = self.core.state();
-        let fh = st.next_fh;
-        st.next_fh += 1;
-        let aw = st.active_writes.entry(ino.0).or_insert_with(|| {
-            WriteHandle {
-                ino: ino.0,
-                uid,
-                file: Arc::new(file),
-                path,
-                written: initial_written,
-                // Starts at the current size; reads in [0, base_size) come from
-                // the base until overwritten.
-                len: base_size,
-                base_size,
-                base_mtime,
-                base_revision_id,
-                dirty: false,
-                open_count: 0,
-            }
-        });
-        aw.open_count += 1;
-        st.handles.insert(fh, ino.0);
-        reply.opened(FileHandle(fh), FopenFlags::empty());
     }
 
     /// The body of [`Filesystem::fsync`], run off the dispatch loop.

crates/pdfs-fuse/src/lib.rs

@@ -55,7 +55,7 @@ use fuser::{
 };
 use futures::StreamExt as _;
 use pdfs_core::batch;
-use pdfs_core::cache::{BLOCK_SIZE, Baseline, ContentCache, StagedWrite};
+use pdfs_core::cache::{BLOCK_SIZE, Baseline, ContentCache, StagedWrite, block_len};
 use pdfs_core::config::{AppDirs, SweepMode};
 use pdfs_core::control::{
     ActivityEntry, ActivityKind, DirEntry, ErrorKind, LocalHit, PhotoKind, PublicLinkInfo,
@@ -92,6 +92,7 @@ mod profile;
 mod reads;
 mod revisions;
 mod sharing;
+mod shutdown;
 mod state;
 mod sweep;
 mod sync;
@@ -410,6 +411,10 @@ struct Core {
     /// kernel read-ahead, our own prefetch) share a single download and decrypt.
     /// See [`BlockFlight`].
     block_flight: Arc<BlockFlight>,
+    /// One lock per file being repaired by [`Core::repair_block`], so the
+    /// concurrent block reads that all discover the same untrustworthy block
+    /// table share a single whole-file download instead of one each.
+    content_repairs: Arc<tokio::sync::Mutex<HashMap<NodeUid, Arc<tokio::sync::Mutex<()>>>>>,
     /// Per-file sequential-read detection driving [`Core::prefetch`].
     prefetch: Arc<Prefetch>,
     /// Permits bounding speculative block fetches, so prefetch cannot queue in
@@ -450,6 +455,9 @@ struct Core {
     /// Nudges the drain workers: set true and notify to have them re-examine
     /// the queue instead of waiting out their backoff.
     drain_wake: Arc<(Mutex<bool>, Condvar)>,
+    /// The daemon's stop signal, shared with every background loop this `Core`
+    /// was cloned into so teardown can end and join them (bugs.md B44).
+    shutdown: Arc<shutdown::Shutdown>,
     /// How long this node's last upload actually took, in milliseconds, with
     /// the time it was measured for eviction. Feeds
     /// [`Core::revision_debounce`]: a file that takes 30 seconds to send needs
@@ -1708,7 +1716,9 @@ impl Core {
     fn run_online_probe(&self) {
         let mut delay = ONLINE_PROBE_MIN;
         loop {
-            std::thread::sleep(delay);
+            if !self.shutdown.sleep(delay) {
+                return;
+            }
             match self.rt.block_on(self.client.get_my_files_folder()) {
                 Ok(root) => {
                     {
@@ -3225,6 +3235,15 @@ impl Core {
         Ok(())
     }
 
+    /// Which staged blob is currently queued for `uid`, if any.
+    ///
+    /// The path identifies the revision: publication stages a fresh file and
+    /// swaps the map entry to point at it, so an unchanged path means the base a
+    /// caller sampled earlier is still the base now (bugs.md B32).
+    fn pending_blob(&self, uid: &NodeUid) -> Option<PathBuf> {
+        self.pending.lock().get(uid).map(|p| p.path.clone())
+    }
+
     /// Nudge the drain worker to re-examine the queue now.
     fn wake_drain(&self) {
         let (lock, cv) = &*self.drain_wake;

crates/pdfs-fuse/src/mount.rs

@@ -365,6 +365,7 @@ pub fn mount(
         readers: Arc::new(Mutex::new(HashMap::new())),
         block_ring: Arc::new(Mutex::new(BlockRing::default())),
         block_flight: Arc::new(BlockFlight::default()),
+        content_repairs: Arc::new(tokio::sync::Mutex::new(HashMap::new())),
         prefetch: Arc::new(Prefetch::default()),
         prefetch_budget: Arc::new(tokio::sync::Semaphore::new(PREFETCH_BUDGET)),
         workers: Arc::new(Workers::new(FUSE_WORKERS)?),
@@ -376,6 +377,7 @@ pub fn mount(
         pending: Arc::new(Mutex::new(HashMap::new())),
         hidden: Arc::new(Mutex::new(HashSet::new())),
         drain_wake: Arc::new((Mutex::new(false), Condvar::new())),
+        shutdown: Arc::new(crate::shutdown::Shutdown::default()),
         upload_times: Arc::new(Mutex::new(HashMap::new())),
         upload_cancel: Arc::new(Mutex::new(HashMap::new())),
         timeline_refreshing: Arc::new(AtomicBool::new(false)),
@@ -420,16 +422,24 @@ pub fn mount(
     // long as it ran; the claim column (`Db::claim_next_due_op`) keeps the
     // workers off each other's nodes, which is the only ordering the queue
     // actually needs. Worker 0 additionally runs the queue's idle chores.
+    //
+    // Every long-lived thread started from here on is kept in `workers` and
+    // joined at the end of this function. Leaving them running turned an
+    // in-process remount into a second generation of drain workers, sync engines
+    // and sweeps all operating on a mount that no longer exists (bugs.md B44).
+    let mut workers: Vec<std::thread::JoinHandle<()>> = Vec::new();
     for worker in 0..DRAIN_WORKERS {
         let core = core.clone();
-        std::thread::Builder::new()
-            .name(format!("pdfs-drain-{worker}"))
-            .spawn(move || core.run_pending_drain(worker == 0))?;
+        workers.push(
+            std::thread::Builder::new()
+                .name(format!("pdfs-drain-{worker}"))
+                .spawn(move || core.run_pending_drain(worker == 0))?,
+        );
     }
 
     // Start the folder-sync engine. It watches every mirror folder, polls the
     // remotes, and reconciles on its own thread — never in front of a FUSE call.
-    sync::spawn(core.clone(), sync_rx);
+    workers.extend(sync::spawn(core.clone(), sync_rx));
 
     // Reconcile leftover `(sync-conflict …)` copies: drop the ones that turned
     // out identical to the live file, surface the ones that genuinely diverge.
@@ -441,18 +451,22 @@ pub fn mount(
     } else {
         info!(mode = ?sweep_mode, "conflict sweep enabled");
         let core = core.clone();
-        std::thread::Builder::new()
-            .name("pdfs-conflict-sweep".into())
-            .spawn(move || core.run_conflict_sweep_loop())?;
+        workers.push(
+            std::thread::Builder::new()
+                .name("pdfs-conflict-sweep".into())
+                .spawn(move || core.run_conflict_sweep_loop())?,
+        );
     }
 
     // Mounted from the cache: watch for the network coming back so the mount can
     // stop being read-only-ish without the user restarting the daemon.
     if !online {
         let core = core.clone();
-        std::thread::Builder::new()
-            .name("pdfs-online-probe".into())
-            .spawn(move || core.run_online_probe())?;
+        workers.push(
+            std::thread::Builder::new()
+                .name("pdfs-online-probe".into())
+                .spawn(move || core.run_online_probe())?,
+        );
     }
 
     // Keep the launcher prompt's "This computer" index fresh. Its own thread:
@@ -462,9 +476,12 @@ pub fn mount(
         let indexing = core.indexing.clone();
         let transfers = core.transfers.clone();
         let mountpoint = mountpoint.to_path_buf();
-        std::thread::Builder::new()
-            .name("pdfs-localindex".into())
-            .spawn(move || run_local_index(db, indexing, transfers, mountpoint))?;
+        let shutdown = core.shutdown.clone();
+        workers.push(
+            std::thread::Builder::new()
+                .name("pdfs-localindex".into())
+                .spawn(move || run_local_index(db, indexing, transfers, mountpoint, shutdown))?,
+        );
     }
 
     // Bind the control socket before the FUSE session takes over the thread. A
@@ -501,23 +518,29 @@ pub fn mount(
         let control_core = core.clone();
         let username = username.clone();
         let mountpoint = mountpoint.to_path_buf();
-        if let Err(error) = std::thread::Builder::new()
+        match std::thread::Builder::new()
             .name("pdfs-control".into())
             .spawn(move || run_control_socket(control_core, username, mountpoint, listener))
         {
-            let secondaries = core.mounts.lock().close_and_drain();
-            for (id, mount) in secondaries {
-                if let Err(unmount_error) = mount.teardown() {
-                    warn!(id, error = %unmount_error, "secondary teardown after control startup failure failed");
+            Ok(handle) => workers.push(handle),
+            Err(error) => {
+                // A failure here abandons the mount, so the background threads
+                // started above have to be told to stop with it.
+                stop_workers(&core, control_socket, workers);
+                let secondaries = core.mounts.lock().close_and_drain();
+                for (id, mount) in secondaries {
+                    if let Err(unmount_error) = mount.teardown() {
+                        warn!(id, error = %unmount_error, "secondary teardown after control startup failure failed");
+                    }
                 }
+                if let Err(unmount_error) = teardown_session(&core.session_live, || {
+                    umount_session_unblocked(bg, main_conn)
+                }) {
+                    warn!(error = %unmount_error, "unmount after control startup failure failed");
+                }
+                let _ = std::fs::remove_file(control_socket);
+                return Err(error);
             }
-            if let Err(unmount_error) = teardown_session(&core.session_live, || {
-                umount_session_unblocked(bg, main_conn)
-            }) {
-                warn!(error = %unmount_error, "unmount after control startup failure failed");
-            }
-            let _ = std::fs::remove_file(control_socket);
-            return Err(error);
         }
     }
     // Re-establish on-demand mounts only after the primary session is live.
@@ -626,10 +649,43 @@ pub fn mount(
         }
     }
 
-    let _ = std::fs::remove_file(control_socket);
+    // Last, because the workers may legitimately be finishing an upload, and
+    // nothing they can still do is harmful once the mounts are down.
+    stop_workers(&core, control_socket, workers);
     Ok(outcome)
 }
 
+/// Signal every background loop, wake the ones that are blocked, and join them.
+///
+/// The mount used to return with its drain workers, sync engine, sweep, online
+/// probe, indexer and control listener all still running on `Core` clones. The
+/// process usually exited straight afterwards and hid it; an in-process remount
+/// did not, and started a second full set on top of the first (bugs.md B44).
+///
+/// Two of the waits need waking rather than merely flagging: the drain workers
+/// block on their own condvar (`wake_drain`), and the control listener blocks
+/// inside `accept`, which only returns when a connection arrives — so it gets
+/// one, from us, after the flag is already set. A join that takes a while is
+/// expected and correct: a worker part-way through an upload finishes it rather
+/// than abandoning the user's bytes mid-flight.
+fn stop_workers(core: &Core, control_socket: &Path, workers: Vec<std::thread::JoinHandle<()>>) {
+    core.shutdown.stop();
+    core.wake_drain();
+    let _ = core.sync_tx.send(sync::SyncMsg::Stop);
+    // Connect while the socket still exists — that is the whole point of the
+    // poke — then unlink it so nothing else can arrive behind us. The listener
+    // re-checks the flag after every accept, so even a connection that races the
+    // unlink cannot put it back to sleep.
+    let _ = std::os::unix::net::UnixStream::connect(control_socket);
+    let _ = std::fs::remove_file(control_socket);
+    for worker in workers {
+        if worker.join().is_err() {
+            warn!("a background worker panicked before shutdown");
+        }
+    }
+    debug!("background workers stopped");
+}
+
 #[cfg(test)]
 mod tests {
     use std::sync::atomic::{AtomicBool, Ordering};

crates/pdfs-fuse/src/reads.rs

@@ -26,6 +26,11 @@ const PREFETCH_MAX: u64 = 8;
 /// Window a sequential read starts at, and returns to after a seek.
 const PREFETCH_MIN: u64 = 2;
 
+/// Largest file [`Core::repair_block`] will pull down whole to work around a
+/// revision whose block table understates it. Past this a single read would cost
+/// gigabytes of download, so the read fails instead and says to pin the file.
+pub(super) const REPAIR_MAX: u64 = 512 * 1024 * 1024;
+
 /// Blocks that may be prefetching at once, across every file. Prefetch takes a
 /// permit and gives up rather than queueing, so it can never sit in front of a
 /// demand read in the SDK's flat block semaphore.
@@ -563,6 +568,10 @@ impl Core {
     /// The disk store is handed to a blocking thread — it is a 4 MiB write plus
     /// a cache-index update, and neither the reactor nor the kernel read waiting
     /// on these bytes should be held up by it.
+    ///
+    /// A block that comes back shorter than the file's size says it must be is
+    /// never served, cached or promoted: it is repaired from the whole-file
+    /// download, or the read fails. See [`Core::repair_block`] (bugs.md B84).
     async fn read_block(
         &self,
         reader: &Arc<RevisionReader>,
@@ -573,11 +582,36 @@ impl Core {
         cache_blocks: bool,
     ) -> BlockResult {
         let bstart = bidx * BLOCK_SIZE;
-        let blen = BLOCK_SIZE.min(fsize - bstart);
-        let bytes = reader.read_at(bstart, blen).await.map_err(|e| {
-            warn!(%uid, bstart, blen, error = %e, "block read failed");
-            Errno::EIO
-        })?;
+        let blen = block_len(fsize, bidx);
+        // The reader derives every block boundary from the revision's recorded
+        // block sizes. If they do not add up to the size the node reports, this
+        // file's boundaries are wrong, and the failure is not confined to short
+        // reads: sizes that *overstate* a block shift every block after it and
+        // return full-length reads of the wrong bytes. So a disagreement of any
+        // kind takes the file off the range path entirely (bugs.md B84).
+        let bytes = if reader.size() != fsize {
+            warn!(
+                %uid, bidx, fsize,
+                reader_size = reader.size(),
+                reader_blocks = reader.block_sizes().len(),
+                "the revision's block table disagrees with the node size"
+            );
+            self.repair_block(uid, mtime, fsize, bidx).await?
+        } else {
+            let bytes = reader.read_at(bstart, blen).await.map_err(|e| {
+                warn!(%uid, bstart, blen, error = %e, "block read failed");
+                Errno::EIO
+            })?;
+            if bytes.len() as u64 == blen {
+                bytes
+            } else {
+                warn!(
+                    %uid, bidx, blen, got = bytes.len(), fsize,
+                    "block came back shorter than the file's size implies"
+                );
+                self.repair_block(uid, mtime, fsize, bidx).await?
+            }
+        };
         let bytes = Arc::new(bytes);
         if cache_blocks {
             let cache = self.cache.clone();
@@ -593,6 +627,129 @@ impl Core {
         Ok(bytes)
     }
 
+    /// Recover block `bidx` of a file whose per-block reads cannot be trusted,
+    /// by fetching the whole revision through the manifest-verified download and
+    /// serving the block out of that.
+    ///
+    /// The range path derives every block boundary from the revision's recorded
+    /// block sizes. When those understate the file, reads come back short with
+    /// no error anywhere, and the caller hands userspace a truncated file that
+    /// looks like a clean EOF (bugs.md B84). `download_file_to` does not do that
+    /// arithmetic — it streams the block list — so it returns the whole file
+    /// where the range path cannot, which is why pinning was the workaround.
+    /// This does the same thing on the user's behalf, once, and the resulting
+    /// blob answers every later read of the file before the block path is
+    /// consulted at all.
+    ///
+    /// Bounded by [`REPAIR_MAX`]: past that, one unlucky read would pull down
+    /// gigabytes, and a clear `EIO` plus a `pdfs pin` is the better trade.
+    async fn repair_block(
+        &self,
+        uid: &NodeUid,
+        mtime: i64,
+        fsize: u64,
+        bidx: u64,
+    ) -> Result<Vec<u8>, Errno> {
+        let blen = block_len(fsize, bidx);
+        let bstart = bidx * BLOCK_SIZE;
+        if fsize > REPAIR_MAX {
+            warn!(
+                %uid, fsize,
+                "file is too large to repair from a whole-file download; pin it to read it"
+            );
+            return Err(Errno::EIO);
+        }
+        // One download per file, not per block: every block of a file with a bad
+        // block table discovers the same problem at once.
+        let lock = {
+            let mut locks = self.content_repairs.lock().await;
+            locks.entry(uid.clone()).or_default().clone()
+        };
+        let guard = lock.clone().lock_owned().await;
+        let downloaded = match self.cache.read_range(uid, mtime, fsize, bstart, blen) {
+            // The first arrival did the work; the rest just read its blob.
+            Some(bytes) if bytes.len() as u64 == blen => Ok(Some(bytes)),
+            _ => self
+                .download_whole_revision(uid, mtime, fsize)
+                .await
+                .map(|()| None),
+        };
+        // Before releasing, so the map entry is only dropped once no waiter is
+        // still holding it.
+        drop(guard);
+        self.release_repair_lock(uid, lock).await;
+        if let Some(bytes) = downloaded? {
+            return Ok(bytes);
+        }
+        match self.cache.read_range(uid, mtime, fsize, bstart, blen) {
+            Some(bytes) if bytes.len() as u64 == blen => {
+                info!(%uid, bidx, "served a short block from a repaired whole-file download");
+                Ok(bytes)
+            }
+            _ => {
+                error!(%uid, bidx, fsize, "repairing a short block did not produce the block");
+                Err(Errno::EIO)
+            }
+        }
+    }
+
+    /// Download the whole of `uid`'s active revision to disk and adopt it as the
+    /// cached blob, so the range path is bypassed entirely for this file.
+    ///
+    /// Streamed into a scratch file rather than memory: this runs for files up
+    /// to [`REPAIR_MAX`], and buffering that per concurrent reader is not a cost
+    /// a read should carry. A download that does not produce exactly `fsize`
+    /// bytes is discarded — a short blob would fail the cache's own validity
+    /// check anyway, and storing it would only hide the disagreement one layer
+    /// further down.
+    async fn download_whole_revision(
+        &self,
+        uid: &NodeUid,
+        mtime: i64,
+        fsize: u64,
+    ) -> Result<(), Errno> {
+        let (mut file, path) = self.cache.create_scratch().map_err(|error| {
+            error!(%uid, %error, "creating a scratch file for a repair download failed");
+            Errno::EIO
+        })?;
+        let outcome = self.client.download_file_to(uid, &mut file).await;
+        let stored = (|| {
+            outcome.map_err(|error| {
+                warn!(%uid, %error, "repair download failed");
+                Errno::EIO
+            })?;
+            file.sync_all().map_err(|error| {
+                warn!(%uid, %error, "syncing a repair download failed");
+                Errno::EIO
+            })?;
+            let got = file.metadata().map(|m| m.len()).unwrap_or(0);
+            if got != fsize {
+                warn!(%uid, fsize, got, "repair download did not match the node size");
+                return Err(Errno::EIO);
+            }
+            self.cache
+                .store_file(uid, mtime, fsize, &path)
+                .map_err(|error| {
+                    error!(%uid, %error, "adopting a repair download as cached content failed");
+                    Errno::EIO
+                })
+        })();
+        // The blob is a hardlink to the same inode, so unlinking the scratch
+        // name leaves it intact.
+        let _ = std::fs::remove_file(&path);
+        stored
+    }
+
+    /// Drop a repair lock's map entry once nobody else holds it, so a file
+    /// repaired long ago does not keep a mutex alive forever.
+    async fn release_repair_lock(&self, uid: &NodeUid, lock: Arc<tokio::sync::Mutex<()>>) {
+        let mut locks = self.content_repairs.lock().await;
+        // Two: the map's and ours. Anything more means a waiter still needs it.
+        if Arc::strong_count(&lock) <= 2 {
+            locks.remove(uid);
+        }
+    }
+
     /// Warm the blocks after a sequential read into the caches, in the
     /// background, so the next read finds them instead of paying a round-trip.
     ///
@@ -761,6 +918,15 @@ impl Core {
             let bstart = bidx * BLOCK_SIZE;
             // Every slot is populated: cache hits up front, misses by the fetch above.
             let block = block.expect("block fetched or cached");
+            // Both sources guarantee the length the file's size implies — the
+            // cache validates it on read, the network path in `read_block` — so
+            // this refuses rather than quietly contributing fewer bytes than the
+            // range covers, which is what turned a bad block into a truncated
+            // file (bugs.md B84).
+            if block.len() as u64 != block_len(fsize, bidx) {
+                error!(%uid, bidx, len = block.len(), fsize, "assembling a read hit a short block");
+                return Err(Errno::EIO);
+            }
             let s = (offset.max(bstart) - bstart) as usize;
             let e = (end.min(bstart + block.len() as u64) - bstart) as usize;
             if s < e {

crates/pdfs-fuse/src/shutdown.rs

@@ -0,0 +1,96 @@
+//! One shutdown signal shared by every long-lived worker in the daemon.
+//!
+//! The mount used to end by unmounting and returning, leaving the drain
+//! workers, the sync engine, the conflict sweep, the online probe and the local
+//! indexer running on `Core` clones. In a process that exits immediately that is
+//! invisible; in an in-process remount — which is how the daemon recovers from a
+//! mount going away underneath it — it leaks a whole generation of threads that
+//! keep writing to the database and calling the API on behalf of a mount that no
+//! longer exists (bugs.md B44).
+//!
+//! The primitive is deliberately small: a flag plus a condvar. Every loop that
+//! would otherwise call `thread::sleep` calls [`Shutdown::sleep`] instead, which
+//! returns early — and reports `false` — the moment [`Shutdown::stop`] is
+//! called, so teardown does not have to wait out a five-minute sweep interval to
+//! join a thread.
+
+use std::time::Duration;
+
+use parking_lot::{Condvar, Mutex};
+
+/// The daemon's stop signal. Set once, never cleared: a `Core` that has been
+/// told to stop belongs to a mount that is going away.
+#[derive(Default)]
+pub(crate) struct Shutdown {
+    stopping: Mutex<bool>,
+    changed: Condvar,
+}
+
+impl Shutdown {
+    /// Whether teardown has begun. Loops check this where they would otherwise
+    /// start another unit of work.
+    pub(crate) fn is_stopping(&self) -> bool {
+        *self.stopping.lock()
+    }
+
+    /// Begin teardown and wake everything waiting in [`sleep`](Self::sleep).
+    /// Idempotent, because more than one path can reach teardown (a signal, the
+    /// kernel mount ending, a failed startup).
+    pub(crate) fn stop(&self) {
+        let mut stopping = self.stopping.lock();
+        *stopping = true;
+        self.changed.notify_all();
+    }
+
+    /// Wait up to `duration`, or until teardown starts.
+    ///
+    /// Returns `false` if the daemon is stopping — including when it was already
+    /// stopping on entry, so `while shutdown.sleep(interval)` is a complete and
+    /// correct loop condition and no separate check is needed at the top.
+    pub(crate) fn sleep(&self, duration: Duration) -> bool {
+        let mut stopping = self.stopping.lock();
+        if !*stopping {
+            self.changed.wait_for(&mut stopping, duration);
+        }
+        !*stopping
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::sync::Arc;
+    use std::time::Instant;
+
+    #[test]
+    fn sleeping_returns_early_once_stopped() {
+        let shutdown = Arc::new(Shutdown::default());
+        assert!(!shutdown.is_stopping());
+
+        let waker = shutdown.clone();
+        std::thread::spawn(move || {
+            std::thread::sleep(Duration::from_millis(20));
+            waker.stop();
+        });
+
+        let started = Instant::now();
+        // An hour's worth of interval, cut short by the stop.
+        assert!(!shutdown.sleep(Duration::from_secs(3600)));
+        assert!(
+            started.elapsed() < Duration::from_secs(5),
+            "stop must interrupt the wait, not wait it out"
+        );
+        assert!(shutdown.is_stopping());
+        // Already stopping: the next wait does not sleep at all.
+        let started = Instant::now();
+        assert!(!shutdown.sleep(Duration::from_secs(3600)));
+        assert!(started.elapsed() < Duration::from_secs(5));
+    }
+
+    #[test]
+    fn sleeping_runs_to_the_timeout_while_running() {
+        let shutdown = Shutdown::default();
+        assert!(shutdown.sleep(Duration::from_millis(10)));
+        assert!(!shutdown.is_stopping());
+    }
+}

crates/pdfs-fuse/src/state.rs

@@ -141,6 +141,11 @@ pub(crate) struct PendingRevision {
     pub(crate) meta: StagedWrite,
 }
 
+/// One directory's entries as `readdir` reports them: `(ino, is_dir, name)` per
+/// entry, in reply order. The kernel's file-type enum is mapped in at reply
+/// time so this module stays free of `fuser`.
+pub(crate) type DirListing = Vec<(u64, bool, String)>;
+
 /// Mutable inode bookkeeping, guarded by a mutex because fuser drives the
 /// `Filesystem` trait through `&self`.
 pub(crate) struct State {
@@ -158,6 +163,16 @@ pub(crate) struct State {
     /// have no entry here.
     pub(crate) handles: HashMap<u64, u64>,
     pub(crate) next_fh: u64,
+    /// One frozen listing per open directory handle, taken at `opendir`.
+    ///
+    /// `readdir` resumes from a byte offset into the listing it is walking, so
+    /// that listing has to stay the same across the calls of one `getdents`
+    /// sequence. Rebuilding it live meant a create or a trash landing between
+    /// two pages shifted every later entry, and the caller silently skipped or
+    /// repeated whatever crossed the page boundary. Each entry is
+    /// `(ino, is_dir, name)` — the kernel type is mapped at reply time so this
+    /// module stays free of `fuser` (bugs.md B43).
+    pub(crate) dir_snapshots: HashMap<u64, Arc<DirListing>>,
     /// Resident inode attributes whose effective access changed during an
     /// intern/root refresh. Core drains this set and notifies the matching
     /// kernel only after releasing the State lock.
@@ -317,6 +332,7 @@ impl State {
             active_writes: HashMap::new(),
             handles: HashMap::new(),
             next_fh: 1,
+            dir_snapshots: HashMap::new(),
             access_changes: HashSet::new(),
             share_access,
             db,

crates/pdfs-fuse/src/sweep.rs

@@ -67,10 +67,16 @@ impl Core {
     /// Sweep for stale conflict copies forever, on its own thread. See the module
     /// docs for the identical/divergent policy.
     pub(crate) fn run_conflict_sweep_loop(&self) {
-        std::thread::sleep(CONFLICT_SWEEP_WARMUP);
+        // Both waits are interruptible, so teardown does not have to sit out a
+        // five-minute interval to join this thread (bugs.md B44).
+        if !self.shutdown.sleep(CONFLICT_SWEEP_WARMUP) {
+            return;
+        }
         loop {
             self.sweep_conflicts_once();
-            std::thread::sleep(CONFLICT_SWEEP_INTERVAL);
+            if !self.shutdown.sleep(CONFLICT_SWEEP_INTERVAL) {
+                return;
+            }
         }
     }
 

crates/pdfs-fuse/src/sync.rs

@@ -54,7 +54,7 @@ use engine::{classify, settle_with};
 #[cfg(test)]
 use local_scan::is_write_mode;
 use local_scan::open_for_write_set;
-use model::{LocalItem, Outcome, PassAbort, Pending, RemoteItem};
+use model::{LocalItem, LocalSig, Outcome, PassAbort, Pending, RemoteItem};
 pub(crate) use planner::base_name;
 #[cfg(test)]
 use planner::conflict_path;
@@ -687,6 +687,7 @@ impl Core {
                             rel_path: rel.clone(),
                             remote_uid: Some(remote.uid.to_string()),
                             local_mtime: local.mtime,
+                            local_mtime_ns: local.mtime_ns,
                             local_size: local.size,
                             remote_rev: Some(remote.mtime.to_string()),
                             remote_hash: Some(remote.size.to_string()),
@@ -717,6 +718,9 @@ impl Core {
                         uid: remote.uid.clone(),
                         mtime: remote.mtime,
                         size: remote.size,
+                        // What the walk saw here, so the apply step can tell a
+                        // stale plan from a live one before it overwrites.
+                        local: l.map(LocalSig::from),
                     });
                 }
                 FilePlan::Conflict => {
@@ -739,7 +743,9 @@ impl Core {
                     }
                 },
                 FilePlan::DeleteLocal => {
-                    if let Err(e) = std::fs::remove_file(local_root.join(rel_to_path(rel))) {
+                    if let Err(e) =
+                        removed_locally(std::fs::remove_file(local_root.join(rel_to_path(rel))))
+                    {
                         warn!(rel, error = %e, "sync: remove local file failed");
                         self.log_activity(
                             ActivityKind::Trash,
@@ -797,7 +803,9 @@ impl Core {
         // Deferred folder deletions, deepest first.
         delete_local_dirs.sort_by_key(|p| std::cmp::Reverse(p.matches('/').count()));
         for rel in delete_local_dirs {
-            if let Err(e) = std::fs::remove_dir_all(local_root.join(rel_to_path(&rel))) {
+            if let Err(e) =
+                removed_locally(std::fs::remove_dir_all(local_root.join(rel_to_path(&rel))))
+            {
                 warn!(rel, error = %e, "sync: remove local folder failed");
                 self.log_activity(ActivityKind::Trash, base_name(&rel), e.to_string(), false);
                 outcome.errors += 1;
@@ -885,6 +893,7 @@ impl Core {
                     LocalItem {
                         is_dir: true,
                         mtime: 0,
+                        mtime_ns: None,
                         size: 0,
                         open_for_write: false,
                     },
@@ -903,6 +912,7 @@ impl Core {
                     LocalItem {
                         is_dir: false,
                         mtime: system_mtime(&meta),
+                        mtime_ns: system_mtime_ns(&meta),
                         size: meta.len() as i64,
                         open_for_write: ofw,
                     },
@@ -1069,6 +1079,7 @@ impl Core {
                     rel_path: rel.to_string(),
                     remote_uid: Some(uid.to_string()),
                     local_mtime: 0,
+                    local_mtime_ns: None,
                     local_size: 0,
                     remote_rev: None,
                     remote_hash: None,
@@ -1192,8 +1203,9 @@ impl Core {
                 uid,
                 mtime,
                 size,
+                local,
             } => {
-                self.download_file(folder_id, local_root, rel, uid, *mtime, *size)
+                self.download_file(folder_id, local_root, rel, uid, *mtime, *size, *local)
                     .await?;
                 Ok(Applied::Downloaded)
             }
@@ -1211,7 +1223,11 @@ impl Core {
                         .map_err(|e| format!("set aside conflict copy for {rel}: {e}"))?;
                     info!(rel, "sync: kept local changes as a conflict copy");
                 }
-                self.download_file(folder_id, local_root, rel, uid, *mtime, *size)
+                // The local copy has just been moved aside, so the state this
+                // download expects to find at the path is "nothing" — anything
+                // that appears while it runs is a *new* racing write, and gets
+                // its own conflict copy rather than being overwritten (B39).
+                self.download_file(folder_id, local_root, rel, uid, *mtime, *size, None)
                     .await?;
                 Ok(Applied::Conflict)
             }
@@ -1250,6 +1266,9 @@ impl Core {
         let file = std::fs::File::open(&path).map_err(|e| format!("open {rel}: {e}"))?;
         let meta = file.metadata().map_err(|e| format!("stat {rel}: {e}"))?;
         let mtime = system_mtime(&meta);
+        // Taken from the descriptor being streamed, so it names the content that
+        // actually goes up rather than whatever is at the path when we finish.
+        let streamed = LocalSig::from(&meta);
         // Count the bytes as they are read, so this upload shows up in
         // `GetQueueStatus` next to a manual one. The uid isn't known until the
         // draft is sealed, so the transfer registers without one.
@@ -1276,7 +1295,9 @@ impl Core {
             )
             .await
             .map_err(|e| format!("upload {rel}: {e}"))?;
-        self.record_file_baseline(folder_id, rel, &path, &uid).await
+        warn_if_torn(rel, &path, streamed);
+        self.record_file_baseline(folder_id, rel, Some(streamed), &uid)
+            .await
     }
 
     /// Upload a changed local file as a new revision of an existing remote node.
@@ -1291,6 +1312,7 @@ impl Core {
         let file = std::fs::File::open(&path).map_err(|e| format!("open {rel}: {e}"))?;
         let meta = file.metadata().map_err(|e| format!("stat {rel}: {e}"))?;
         let mtime = system_mtime(&meta);
+        let streamed = LocalSig::from(&meta);
         let reader = OwnedCountingReader::new(
             file,
             self.transfers.begin(
@@ -1304,11 +1326,20 @@ impl Core {
             .upload_new_revision_from(uid, reader, meta.len() as i64, Vec::new(), Some(mtime))
             .await
             .map_err(|e| format!("upload revision {rel}: {e}"))?;
-        self.record_file_baseline(folder_id, rel, &path, uid).await
+        warn_if_torn(rel, &path, streamed);
+        self.record_file_baseline(folder_id, rel, Some(streamed), uid)
+            .await
     }
 
     /// Download a remote file to its local path (atomically via a temp file),
     /// stamp the local mtime to match the remote, then record baseline.
+    ///
+    /// `expected_local` is the identity planning saw at the destination. The
+    /// rename below is the destructive step, and the download in between can run
+    /// for minutes, so the destination is re-examined right before it: a local
+    /// file that no longer matches the plan is set aside as a conflict copy
+    /// instead of being replaced (bugs.md B39).
+    #[allow(clippy::too_many_arguments)]
     async fn download_file(
         &self,
         folder_id: i64,
@@ -1317,6 +1348,7 @@ impl Core {
         uid: &NodeUid,
         mtime: i64,
         size: i64,
+        expected_local: Option<LocalSig>,
     ) -> Result<(), String> {
         let path = local_root.join(rel_to_path(rel));
         if let Some(parent) = path.parent() {
@@ -1351,6 +1383,13 @@ impl Core {
             out.flush().map_err(|e| format!("flush tmp {rel}: {e}"))?;
             out.sync_all().map_err(|e| format!("sync tmp {rel}: {e}"))?;
         }
+        // The last moment at which the destination can still be saved. Anything
+        // other than the state planning classified means a writer got there
+        // first, and its bytes are the ones nothing else has a copy of.
+        if let Err(e) = keep_racing_local_edit(&path, expected_local) {
+            let _ = std::fs::remove_file(&tmp);
+            return Err(format!("set aside racing local edit for {rel}: {e}"));
+        }
         std::fs::rename(&tmp, &path).map_err(|e| format!("place {rel}: {e}"))?;
         // Match local mtime to the remote's so neither side looks "changed" next pass.
         let f = std::fs::File::options()
@@ -1366,23 +1405,36 @@ impl Core {
                 .and_then(|dir| dir.sync_all())
                 .map_err(|e| format!("sync directory for {rel}: {e}"))?;
         }
-        self.record_file_baseline(folder_id, rel, &path, uid).await
+        // Here the file on disk *is* the transferred content: it was published
+        // by a rename and stamped just above, so its stat is the identity.
+        self.record_file_baseline(folder_id, rel, local_sig(&path), uid)
+            .await
     }
 
-    /// Refresh a file's baseline from ground truth: the local stat and the
-    /// remote node's reported signature. Called after every upload/download so
-    /// the very next reconcile sees the path as unchanged.
+    /// Refresh a file's baseline from ground truth: the local identity these
+    /// bytes came from and the remote node's reported signature. Called after
+    /// every upload/download so the very next reconcile sees the path as
+    /// unchanged.
+    ///
+    /// `local` is the identity the *transferred content* corresponds to, which
+    /// is not always the file's current stat. An upload streams the live file,
+    /// so it passes the identity it read the first byte at: if a writer changed
+    /// the file mid-stream, the remote now holds a torn revision, and recording
+    /// the pre-upload identity is what makes the next pass see the path as
+    /// locally changed and upload a whole, clean revision over it. Recording the
+    /// post-upload stat instead declared the torn revision synchronized and left
+    /// it as the file's content on Drive (bugs.md B40).
     async fn record_file_baseline(
         &self,
         folder_id: i64,
         rel: &str,
-        local_path: &Path,
+        local: Option<LocalSig>,
         uid: &NodeUid,
     ) -> Result<(), String> {
-        let (lmtime, lsize) = match std::fs::metadata(local_path) {
-            Ok(m) => (system_mtime(&m), m.len() as i64),
-            Err(e) => return Err(format!("stat {rel}: {e}")),
+        let Some(local) = local else {
+            return Err(format!("stat {rel}: no local identity to baseline"));
         };
+        let (lmtime, lsize) = (local.mtime, local.size);
         // Re-fetch the node so the stored remote signature is exactly what a walk
         // will report next time.
         let (rmtime, rsize) = match self.client.enumerate_nodes(std::slice::from_ref(uid)).await {
@@ -1402,6 +1454,7 @@ impl Core {
                     rel_path: rel.to_string(),
                     remote_uid: Some(uid.to_string()),
                     local_mtime: lmtime,
+                    local_mtime_ns: local.mtime_ns,
                     local_size: lsize,
                     remote_rev: Some(rmtime.to_string()),
                     remote_hash: Some(rsize.to_string()),
@@ -1411,6 +1464,88 @@ impl Core {
     }
 }
 
+/// Say so when a file changed under an upload that was streaming it.
+///
+/// Purely diagnostic — the correction is already in place, because the baseline
+/// records the identity the upload *read* rather than the one on disk now, so
+/// the next pass classifies the path as locally changed and replaces the torn
+/// revision. Without the log there is nothing in the journal to explain the
+/// extra revision (bugs.md B40).
+fn warn_if_torn(rel: &str, path: &Path, streamed: LocalSig) {
+    let now = local_sig(path);
+    if !now.is_some_and(|now| now.same_content(&streamed)) {
+        warn!(
+            rel,
+            ?streamed,
+            ?now,
+            "sync: file changed while uploading; the remote revision may be torn \
+             and will be replaced on the next pass"
+        );
+    }
+}
+
+/// The identity a reconcile pass uses to recognise a local file it has already
+/// looked at: `None` when nothing is at the path.
+fn local_sig(path: &Path) -> Option<LocalSig> {
+    std::fs::metadata(path).ok().map(|m| LocalSig {
+        mtime: system_mtime(&m),
+        mtime_ns: system_mtime_ns(&m),
+        size: m.len() as i64,
+    })
+}
+
+/// Guard the destination of a download: if what is at `path` is no longer what
+/// planning classified, move it aside as a conflict copy before the caller
+/// renames over it.
+///
+/// A reconcile pass decides what to do from a walk that finished before the
+/// transfer started, and a download of a large file can run for minutes. Without
+/// this, a local edit completed inside that window was replaced by the remote
+/// version with no copy of it kept anywhere — the one shape of loss a sync
+/// engine must not have (bugs.md B39). Keeping the copy makes it re-upload as a
+/// new file on the next pass, which is the same resolution a both-sides-changed
+/// conflict already gets.
+///
+/// A path that is empty when it was expected to be is not a conflict: the
+/// download is about to recreate it, and there is nothing to preserve.
+fn keep_racing_local_edit(path: &Path, expected: Option<LocalSig>) -> std::io::Result<()> {
+    let actual = local_sig(path);
+    let unchanged = match (actual, expected) {
+        (Some(a), Some(e)) => a.same_content(&e),
+        // Nothing there: the download is about to create it, and there is
+        // nothing to preserve either way.
+        (None, _) => true,
+        (Some(_), None) => false,
+    };
+    if unchanged {
+        return Ok(());
+    }
+    warn!(
+        path = %path.display(),
+        ?expected,
+        ?actual,
+        "sync: local file changed while downloading; keeping it as a conflict copy"
+    );
+    preserve_conflict_copy(path, now_secs()).map(|_| ())
+}
+
+/// Read a local removal as the sync engine has to read it: the *state* it was
+/// asking for is "this path is gone", so a path that was already gone is a
+/// success, not a failure.
+///
+/// It matters because the baseline is only dropped on success. Treating
+/// `NotFound` as an error left the baseline in place for a file that no longer
+/// exists, so the same removal was re-attempted and re-reported as an error on
+/// every pass, forever. Every other error is still an error, and keeping the
+/// baseline is exactly right there: the surviving file must not be mistaken for
+/// a new local creation on the next pass and uploaded back (bugs.md B42).
+fn removed_locally(result: std::io::Result<()>) -> std::io::Result<()> {
+    match result {
+        Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(()),
+        other => other,
+    }
+}
+
 /// Publish a conflict copy without overwriting an earlier conflict bearing the
 /// same timestamp. The original is removed only after the new copy is durable.
 fn preserve_conflict_copy(path: &Path, stamp: i64) -> std::io::Result<PathBuf> {
@@ -1476,10 +1611,77 @@ fn system_mtime(meta: &std::fs::Metadata) -> i64 {
         .map(|d| d.as_secs() as i64)
         .unwrap_or(0)
 }
+
+/// The same modification time in nanoseconds, when the filesystem reports one.
+///
+/// Whole seconds cannot distinguish an edit made in the same second as the last
+/// sync from no edit at all, and if the size did not change either, that file
+/// was never uploaded (bugs.md B25). `None` for a clock or filesystem that
+/// cannot answer, which falls the comparison back to seconds rather than
+/// asserting an mtime of zero.
+fn system_mtime_ns(meta: &std::fs::Metadata) -> Option<i64> {
+    meta.modified()
+        .ok()
+        .and_then(|t| t.duration_since(SystemTime::UNIX_EPOCH).ok())
+        .and_then(|d| i64::try_from(d.as_nanos()).ok())
+}
 #[cfg(test)]
 mod tests {
     use super::*;
 
+    /// A removal of something already gone has produced the state the sync
+    /// engine wanted, so its baseline is dropped rather than retried forever.
+    /// Anything else stays an error and keeps the baseline (bugs.md B42).
+    #[test]
+    fn an_already_absent_path_counts_as_removed() {
+        let root = std::env::temp_dir().join(format!("pdfs-b42-{}", std::process::id()));
+        let missing = root.join("never-existed");
+        assert!(removed_locally(std::fs::remove_file(&missing)).is_ok());
+        assert!(removed_locally(std::fs::remove_dir_all(&missing)).is_ok());
+
+        // A directory in the way of `remove_file` is a real failure, and must
+        // stay one: the file survives, and its baseline has to survive with it.
+        let occupied = root.join("a-directory");
+        std::fs::create_dir_all(&occupied).unwrap();
+        assert!(removed_locally(std::fs::remove_file(&occupied)).is_err());
+        let _ = std::fs::remove_dir_all(&root);
+    }
+
+    /// The download guard: a destination that still matches the plan is
+    /// replaced silently, one that has moved on since is kept (bugs.md B39).
+    #[test]
+    fn a_download_keeps_a_local_edit_it_did_not_plan_for() {
+        let root = std::env::temp_dir().join(format!("pdfs-b39-{}", std::process::id()));
+        std::fs::create_dir_all(&root).unwrap();
+        let path = root.join("report.txt");
+
+        // Nothing there, nothing expected: the download simply creates it.
+        assert!(keep_racing_local_edit(&path, None).is_ok());
+        assert!(!path.exists());
+
+        // Unchanged since planning: no copy, the caller overwrites it.
+        std::fs::write(&path, b"planned").unwrap();
+        let planned = local_sig(&path).unwrap();
+        keep_racing_local_edit(&path, Some(planned)).unwrap();
+        assert_eq!(std::fs::read(&path).unwrap(), b"planned");
+
+        // Changed since planning: the racing bytes are the only copy of
+        // themselves, so they must survive the overwrite as a conflict copy.
+        std::fs::write(&path, b"a much longer local edit").unwrap();
+        keep_racing_local_edit(&path, Some(planned)).unwrap();
+        let kept: Vec<_> = std::fs::read_dir(&root)
+            .unwrap()
+            .filter_map(|e| e.ok())
+            .map(|e| std::fs::read(e.path()).unwrap_or_default())
+            .collect();
+        assert!(
+            kept.iter()
+                .any(|bytes| bytes == b"a much longer local edit"),
+            "the racing edit must still exist somewhere: {kept:?}"
+        );
+        let _ = std::fs::remove_dir_all(&root);
+    }
+
     /// Feed `n` `Reconcile` events `gap` apart on another thread, starting
     /// immediately. Returns the sender's join handle so the test can keep it
     /// alive (dropping the sender early would look like a disconnect).
@@ -1565,6 +1767,7 @@ mod tests {
 
     fn local_item(is_dir: bool) -> LocalItem {
         LocalItem {
+            mtime_ns: None,
             is_dir,
             mtime: 1,
             size: 1,
@@ -1764,6 +1967,7 @@ mod tests {
                 remote_uid: Some("v~l".into()),
                 local_mtime: 0,
                 local_size: 0,
+                local_mtime_ns: None,
                 remote_rev: Some(mtime.to_string()),
                 remote_hash: Some(size.to_string()),
             },
@@ -1801,6 +2005,7 @@ mod tests {
                 remote_uid: Some("v~l".into()),
                 local_mtime: 0,
                 local_size: 0,
+                local_mtime_ns: None,
                 remote_rev: None,
                 remote_hash: None,
             },
@@ -1815,6 +2020,7 @@ mod tests {
             remote_uid: Some("v~l".into()),
             local_mtime: 10,
             local_size: 20,
+            local_mtime_ns: None,
             remote_rev: Some("1700".into()),
             remote_hash: Some("4096".into()),
         };
@@ -1828,6 +2034,7 @@ mod tests {
             remote_uid: Some("v~l".into()),
             local_mtime: 0,
             local_size: 0,
+            local_mtime_ns: None,
             remote_rev: None,
             remote_hash: None,
         };
@@ -1863,6 +2070,7 @@ mod tests {
             uid: uid(),
             mtime: 0,
             size: 10,
+            local: None,
         };
         assert_eq!(op.kind(), ActivityKind::Download);
     }

crates/pdfs-fuse/src/sync/engine.rs

@@ -32,17 +32,24 @@ pub(crate) enum SyncMsg {
     ReconcileAll,
     /// Re-read the folder set and adjust the filesystem watches (after add/remove).
     Rewatch,
+    /// The daemon is tearing down: finish the current pass and end the engine
+    /// thread, so the mount can join it (bugs.md B44).
+    Stop,
 }
 
 /// Start the sync engine: a control thread that owns the filesystem watcher and
 /// serialises reconcile passes, plus a periodic poll thread. The engine receives
 /// on `rx`; senders live in [`Core::sync_tx`].
-pub(crate) fn spawn(core: Core, rx: Receiver<SyncMsg>) {
-    if let Err(e) = std::thread::Builder::new()
+pub(crate) fn spawn(core: Core, rx: Receiver<SyncMsg>) -> Option<std::thread::JoinHandle<()>> {
+    match std::thread::Builder::new()
         .name("pdfs-sync".into())
         .spawn(move || engine_loop(core, rx))
     {
-        warn!(error = %e, "failed to start sync engine");
+        Ok(handle) => Some(handle),
+        Err(e) => {
+            warn!(error = %e, "failed to start sync engine");
+            None
+        }
     }
 }
 
@@ -74,11 +81,11 @@ fn engine_loop(core: Core, rx: Receiver<SyncMsg>) {
     // Periodic remote poll.
     {
         let tx = core.sync_tx.clone();
+        let shutdown = core.shutdown.clone();
         std::thread::Builder::new()
             .name("pdfs-sync-poll".into())
             .spawn(move || {
-                loop {
-                    std::thread::sleep(POLL_INTERVAL);
+                while shutdown.sleep(POLL_INTERVAL) {
                     if tx.send(SyncMsg::ReconcileAll).is_err() {
                         break;
                     }
@@ -94,8 +101,12 @@ fn engine_loop(core: Core, rx: Receiver<SyncMsg>) {
         let mut ids: HashSet<i64> = HashSet::new();
         let mut all = false;
         let mut do_rewatch = false;
-        classify(msg, &mut ids, &mut all, &mut do_rewatch);
-        settle(&rx, &mut ids, &mut all, &mut do_rewatch);
+        if !classify(msg, &mut ids, &mut all, &mut do_rewatch)
+            || !settle(&rx, &mut ids, &mut all, &mut do_rewatch)
+        {
+            info!("sync engine stopping");
+            return;
+        }
 
         if do_rewatch {
             rewatch(&core, &mut watcher, &watched);
@@ -116,9 +127,9 @@ fn engine_loop(core: Core, rx: Receiver<SyncMsg>) {
 fn poll_only_loop(core: &Core, rx: Receiver<SyncMsg>) {
     {
         let tx = core.sync_tx.clone();
+        let shutdown = core.shutdown.clone();
         std::thread::spawn(move || {
-            loop {
-                std::thread::sleep(POLL_INTERVAL);
+            while shutdown.sleep(POLL_INTERVAL) {
                 if tx.send(SyncMsg::ReconcileAll).is_err() {
                     break;
                 }
@@ -130,8 +141,12 @@ fn poll_only_loop(core: &Core, rx: Receiver<SyncMsg>) {
         let mut ids: HashSet<i64> = HashSet::new();
         let mut all = false;
         let mut do_rewatch = false;
-        classify(msg, &mut ids, &mut all, &mut do_rewatch);
-        settle(&rx, &mut ids, &mut all, &mut do_rewatch);
+        if !classify(msg, &mut ids, &mut all, &mut do_rewatch)
+            || !settle(&rx, &mut ids, &mut all, &mut do_rewatch)
+        {
+            info!("sync engine stopping");
+            return;
+        }
         if all {
             reconcile_all(core);
         } else {
@@ -158,7 +173,7 @@ pub(super) fn settle_with(
     do_rewatch: &mut bool,
     quiet: Duration,
     cap: Duration,
-) {
+) -> bool {
     let deadline = Instant::now() + cap;
     loop {
         let now = Instant::now();
@@ -170,28 +185,49 @@ pub(super) fn settle_with(
         match rx.recv_timeout(wait) {
             // Another event: the burst is still going, so the quiet window
             // starts again from here.
-            Ok(m) => classify(m, ids, all, do_rewatch),
+            // A stop lands mid-burst: give up the rest of the burst rather
+            // than settling for it, and let the caller end the loop.
+            Ok(m) => {
+                if !classify(m, ids, all, do_rewatch) {
+                    return false;
+                }
+            }
             // Quiet for a full window: whatever was being written has stopped.
             Err(RecvTimeoutError::Timeout) => break,
             // Every sender is gone; the caller's loop is about to end anyway.
             Err(RecvTimeoutError::Disconnected) => break,
         }
     }
+    true
 }
 
 /// [`settle_with`] at the real timings.
-fn settle(rx: &Receiver<SyncMsg>, ids: &mut HashSet<i64>, all: &mut bool, do_rewatch: &mut bool) {
-    settle_with(rx, ids, all, do_rewatch, DEBOUNCE, MAX_SETTLE);
+fn settle(
+    rx: &Receiver<SyncMsg>,
+    ids: &mut HashSet<i64>,
+    all: &mut bool,
+    do_rewatch: &mut bool,
+) -> bool {
+    settle_with(rx, ids, all, do_rewatch, DEBOUNCE, MAX_SETTLE)
 }
 
-pub(super) fn classify(msg: SyncMsg, ids: &mut HashSet<i64>, all: &mut bool, rewatch: &mut bool) {
+/// Fold one message into the pending work of the current burst. Returns `false`
+/// for [`SyncMsg::Stop`], which is not work but an instruction to end the loop.
+pub(super) fn classify(
+    msg: SyncMsg,
+    ids: &mut HashSet<i64>,
+    all: &mut bool,
+    rewatch: &mut bool,
+) -> bool {
     match msg {
         SyncMsg::Reconcile(id) => {
             ids.insert(id);
         }
         SyncMsg::ReconcileAll => *all = true,
         SyncMsg::Rewatch => *rewatch = true,
+        SyncMsg::Stop => return false,
     }
+    true
 }
 
 /// Reconcile every mirror folder in turn.

crates/pdfs-fuse/src/sync/model.rs

@@ -2,9 +2,73 @@
 
 use super::*;
 
+/// What a reconcile pass remembers about the local side of one path: enough to
+/// recognise the same content again, and no more.
+///
+/// `mtime_ns` is the whole reason this is a struct rather than a pair — see
+/// [`StoredSyncEntry::local_mtime_ns`](pdfs_core::db::StoredSyncEntry).
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub(super) struct LocalSig {
+    pub(super) mtime: i64,
+    pub(super) mtime_ns: Option<i64>,
+    pub(super) size: i64,
+}
+
+impl LocalSig {
+    /// Whether these two observations describe the same local content.
+    ///
+    /// Nanoseconds decide it when both sides have them; otherwise the comparison
+    /// falls back to whole seconds, which is what a baseline row written before
+    /// schema 28 — or a filesystem that reports no sub-second time — can support
+    /// (bugs.md B25).
+    pub(super) fn same_content(&self, other: &Self) -> bool {
+        if self.size != other.size {
+            return false;
+        }
+        match (self.mtime_ns, other.mtime_ns) {
+            (Some(a), Some(b)) => a == b,
+            _ => self.mtime == other.mtime,
+        }
+    }
+}
+
+impl From<&LocalItem> for LocalSig {
+    fn from(item: &LocalItem) -> Self {
+        Self {
+            mtime: item.mtime,
+            mtime_ns: item.mtime_ns,
+            size: item.size,
+        }
+    }
+}
+
+impl From<&std::fs::Metadata> for LocalSig {
+    fn from(meta: &std::fs::Metadata) -> Self {
+        Self {
+            mtime: system_mtime(meta),
+            mtime_ns: system_mtime_ns(meta),
+            size: meta.len() as i64,
+        }
+    }
+}
+
+/// The baseline's record of the local side, as [`LocalSig`] compares it.
+impl From<&StoredSyncEntry> for LocalSig {
+    fn from(entry: &StoredSyncEntry) -> Self {
+        Self {
+            mtime: entry.local_mtime,
+            mtime_ns: entry.local_mtime_ns,
+            size: entry.local_size,
+        }
+    }
+}
+
 pub(super) struct LocalItem {
     pub(super) is_dir: bool,
     pub(super) mtime: i64,
+    /// [`LocalItem::mtime`] to nanosecond precision, when the filesystem
+    /// reports one.
+    pub(super) mtime_ns: Option<i64>,
     pub(super) size: i64,
     /// True when another process holds this file open for writing (detected via
     /// `/proc/*/fd`). An open-for-write file is kept in the map — so it is not
@@ -91,11 +155,18 @@ pub(super) enum Pending {
     UploadRevision { rel: String, uid: NodeUid },
     /// Download remote `uid` to the local path, stamping `mtime`. `size` is the
     /// remote's reported size, used as the transfer's expected total.
+    ///
+    /// `local` is what the planning walk saw at this path, or `None` if nothing
+    /// was there. A download can run for minutes, so the apply step revalidates
+    /// against it immediately before it renames over the destination: a writer
+    /// that finished an edit in the meantime must not be overwritten without a
+    /// conflict copy (bugs.md B39).
     Download {
         rel: String,
         uid: NodeUid,
         mtime: i64,
         size: i64,
+        local: Option<LocalSig>,
     },
     /// Both sides changed: set the local copy aside, then download remote `uid`.
     Conflict {

crates/pdfs-fuse/src/sync/planner.rs

@@ -218,7 +218,10 @@ pub(super) fn plan_file(
     }
 
     let local_changed = local.is_some_and(|item| {
-        baseline.is_none_or(|base| base.local_mtime != item.mtime || base.local_size != item.size)
+        // Nanoseconds where both sides have them (bugs.md B25): whole seconds
+        // could not see an edit made in the same second as the last sync that
+        // left the file the same length, so it was never uploaded.
+        baseline.is_none_or(|base| !LocalSig::from(item).same_content(&LocalSig::from(base)))
     });
     let remote_changed = remote.is_some_and(|item| {
         baseline.is_none_or(|base| remote_sig(base) != Some((item.mtime, item.size)))
@@ -252,6 +255,7 @@ mod file_plan_tests {
 
     fn local(mtime: i64, size: i64) -> LocalItem {
         LocalItem {
+            mtime_ns: None,
             is_dir: false,
             mtime,
             size,
@@ -278,12 +282,49 @@ mod file_plan_tests {
             rel_path: "file".into(),
             remote_uid: Some("v~l".into()),
             local_mtime,
+            local_mtime_ns: None,
             local_size,
             remote_rev: Some(remote_mtime.to_string()),
             remote_hash: Some(remote_size.to_string()),
         }
     }
 
+    /// Whole seconds cannot see an edit made inside the same second that leaves the
+    /// file the same length, so a mirror silently stopped uploading it. With a
+    /// nanosecond time on both sides it is an ordinary local change (bugs.md B25).
+    #[test]
+    fn a_same_second_same_size_edit_is_still_a_local_change() {
+        let mut base = baseline(10, 20, 30, 40);
+        base.local_mtime_ns = Some(10_000_000_000);
+        let edited = LocalItem {
+            mtime_ns: Some(10_400_000_000),
+            ..local(10, 20)
+        };
+        let remote = remote(30, 40);
+        assert_eq!(
+            plan_file(Some(&edited), Some(&remote), Some(&base)),
+            FilePlan::UploadRevision
+        );
+
+        // The same file untouched still reads as unchanged.
+        let same = LocalItem {
+            mtime_ns: Some(10_000_000_000),
+            ..local(10, 20)
+        };
+        assert_eq!(
+            plan_file(Some(&same), Some(&remote), Some(&base)),
+            FilePlan::Unchanged
+        );
+
+        // And a row from before the column exists compares on seconds, exactly
+        // as it did when it was written — no mass re-upload on upgrade.
+        let old = baseline(10, 20, 30, 40);
+        assert_eq!(
+            plan_file(Some(&edited), Some(&remote), Some(&old)),
+            FilePlan::Unchanged
+        );
+    }
+
     #[test]
     fn file_plan_covers_every_presence_and_change_shape() {
         let same_local = local(10, 20);