fix(fs) locks, vacuum, recover fsynced writes

crates/pdfs-cli/src/main.rs

@@ -1336,6 +1336,11 @@ fn cmd_status() -> Result<()> {
             online,
             pending_uploads,
             pending_changes,
+            parked_uploads,
+            failing_ops,
+            failing_error,
+            staged_bytes,
+            staged_oldest_secs,
             ..
         }) => {
             let state = if online { "" } else { ", offline" };
@@ -1344,6 +1349,25 @@ fn cmd_status() -> Result<()> {
                 None => String::new(),
             };
             println!("Mounted at {mountpoint} ({pinned} pinned{state}{queued})");
+            // Everything below is a queue that is not making progress on its
+            // own. Printed only when it exists, because in the ordinary case
+            // there is nothing to say.
+            if staged_bytes > 0 {
+                println!(
+                    "Staged     {} not yet uploaded (oldest {})",
+                    human_bytes(staged_bytes),
+                    human_age(staged_oldest_secs),
+                );
+            }
+            if parked_uploads > 0 {
+                println!("Parked     {parked_uploads} upload(s) waiting for a final filename");
+            }
+            if failing_ops > 0 {
+                println!(
+                    "Stuck      {failing_ops} queued operation(s) keep failing: {}",
+                    failing_error.as_deref().unwrap_or("no error recorded"),
+                );
+            }
         }
         Ok(other) => println!("Mount: unexpected response {other:?}"),
         Err(_) => println!("Mount: not running."),
@@ -1886,6 +1910,18 @@ fn path_arg(path: &Path) -> Result<String> {
         .ok_or_else(|| anyhow!("path is not valid UTF-8"))
 }
 
+/// Format an age in seconds for a report line, coarsely — the point of the
+/// number is whether it is minutes or weeks.
+fn human_age(secs: u64) -> String {
+    const UNITS: [(&str, u64); 4] = [("d", 86_400), ("h", 3_600), ("m", 60), ("s", 1)];
+    for (unit, scale) in UNITS {
+        if secs >= scale {
+            return format!("{}{unit}", secs / scale);
+        }
+    }
+    "just now".to_string()
+}
+
 /// Format a byte count for a report line.
 fn human_bytes(bytes: u64) -> String {
     const UNITS: [(&str, f64); 4] = [
@@ -1925,6 +1961,32 @@ fn cmd_cache(action: CacheCmd) -> Result<()> {
     }
 }
 
+/// Ask for a deep integrity check and wait for it to finish.
+///
+/// The daemon answers the request immediately and runs the check on a thread —
+/// it holds the single database connection for as long as it takes, which would
+/// otherwise stall the mount — so a caller that needs the verdict polls for it.
+/// `pdfs doctor` is that caller: it is a diagnostic the user is watching, so
+/// waiting is right, and the bound is generous enough for a large database
+/// without hanging forever if the daemon never finishes.
+fn await_integrity_check() -> Result<CtlResponse> {
+    const POLL: std::time::Duration = std::time::Duration::from_secs(1);
+    const GIVE_UP_AFTER: std::time::Duration = std::time::Duration::from_secs(120);
+
+    let started = std::time::Instant::now();
+    loop {
+        let response = control_request(CtlRequest::CacheInspect { deep: true })?;
+        match &response {
+            CtlResponse::CacheReport {
+                integrity_running: true,
+                integrity_checked: false,
+                ..
+            } if started.elapsed() < GIVE_UP_AFTER => std::thread::sleep(POLL),
+            _ => return Ok(response),
+        }
+    }
+}
+
 fn cmd_cache_inspect(deep: bool) -> Result<()> {
     let response = control_request(CtlRequest::CacheInspect { deep })?;
     if emit_json(&response)? {
@@ -1940,6 +2002,7 @@ fn cmd_cache_inspect(deep: bool) -> Result<()> {
             cache_budget,
             integrity_problems,
             integrity_checked,
+            integrity_running,
         } => {
             println!(
                 "Database   schema v{schema_version}, {}",
@@ -1972,6 +2035,11 @@ fn cmd_cache_inspect(deep: bool) -> Result<()> {
                         println!("  {problem}");
                     }
                 }
+            } else if integrity_running {
+                // The daemon runs the check on a thread rather than holding its
+                // only database connection for the length of it, so the answer
+                // belongs to a later request.
+                println!("\nIntegrity  checking in the background — run this again for the result");
             } else {
                 println!("\nIntegrity  not checked (pass --deep)");
             }
@@ -2209,14 +2277,24 @@ fn cmd_diagnose() -> Result<()> {
                     format!("{pending} waiting to upload")
                 },
             );
-            match control_request(CtlRequest::CacheInspect { deep: true }) {
+            match await_integrity_check() {
                 Ok(CtlResponse::CacheReport {
-                    integrity_problems, ..
+                    integrity_problems,
+                    integrity_checked: true,
+                    ..
                 }) if integrity_problems.is_empty() => report.finding(
                     DiagnoseLevel::Ok,
                     "  database integrity",
                     "SQLite integrity_check passed",
                 ),
+                Ok(CtlResponse::CacheReport {
+                    integrity_checked: false,
+                    ..
+                }) => report.finding(
+                    DiagnoseLevel::Warn,
+                    "  database integrity",
+                    "still checking — run `pdfs cache inspect --deep` for the result",
+                ),
                 Ok(CtlResponse::CacheReport {
                     integrity_problems, ..
                 }) => report.finding(

crates/pdfs-core/src/cache.rs

@@ -21,7 +21,7 @@ use std::fmt;
 use std::io::{Read, Seek, SeekFrom};
 use std::path::{Path, PathBuf};
 use std::sync::Arc;
-use std::sync::atomic::{AtomicU64, Ordering};
+use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
 use std::time::{SystemTime, UNIX_EPOCH};
 
 use proton_drive_rs::proton_sdk::ids::NodeUid;
@@ -146,6 +146,19 @@ struct PinFile {
 /// in the gaps. Uploading that as-is would silently corrupt the file.
 /// `authored` says which ranges are real; `complete` says whether the file can
 /// be uploaded as it stands.
+/// What `staging/` currently holds, from [`ContentCache::staging_usage`].
+#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
+pub struct StagingUsage {
+    /// Bytes across every staged blob.
+    pub bytes: u64,
+    /// Staged blobs, not counting their sidecars.
+    pub files: u64,
+    /// Age of the oldest staged blob, in seconds. Zero when there are none —
+    /// what makes a figure here interesting is it being *large*, which is the
+    /// shape of a write that is never going to drain by itself.
+    pub oldest_secs: u64,
+}
+
 #[derive(Serialize, Deserialize, Debug, Clone)]
 pub struct StagedWrite {
     /// Node the write targeted, in `volume~link` display form.
@@ -287,6 +300,14 @@ pub struct ContentCache {
     /// once [`TOUCH_FLUSH_AT`] have accumulated, and always before an eviction
     /// pass reads the ordering these describe.
     touches: Mutex<HashMap<String, i64>>,
+    /// Whether `recovery/` may hold a write the daemon has not queued yet.
+    ///
+    /// The drain asks for recovered writes on every idle pass — a `read_dir` of
+    /// a directory that is empty in every run that did not crash. The directory
+    /// gains entries in exactly two places (the rescue at open, and the release
+    /// path that parks a still-authoritative scratch file), both of which set
+    /// this; a scan that finds nothing actionable clears it.
+    recovery_pending: AtomicBool,
 }
 
 impl ContentCache {
@@ -350,6 +371,8 @@ impl ContentCache {
             thumb_bytes: AtomicU64::new(0),
             db,
             touches: Mutex::new(HashMap::new()),
+            // The rescue above has already run, so start by looking once.
+            recovery_pending: AtomicBool::new(true),
         };
         cache.reconcile()?;
         // After reconcile, so the totals describe the index that now describes
@@ -1039,7 +1062,14 @@ impl ContentCache {
     /// daemon to hand to the upload queue. A blob whose sidecar did not survive
     /// is not actionable automatically, but is retained for manual recovery:
     /// unexplained bytes are never evidence that user data is disposable.
+    ///
+    /// Asked for on every idle drain pass, so the usual answer — a directory
+    /// that has been empty since the last clean shutdown — costs an atomic load
+    /// rather than a `read_dir` (see [`recovery_pending`](Self::recovery_pending)).
     pub fn recovered_writes(&self) -> Vec<(PathBuf, StagedWrite)> {
+        if !self.recovery_pending.load(Ordering::Relaxed) {
+            return Vec::new();
+        }
         let Ok(entries) = std::fs::read_dir(&self.recovery_dir) else {
             return Vec::new();
         };
@@ -1056,6 +1086,39 @@ impl ContentCache {
                 out.push((blob, meta));
             }
         }
+        // Blobs whose sidecar did not survive stay here for a human and are
+        // never actionable, so "nothing to queue" is the condition that stops
+        // the polling — not "the directory is empty".
+        if out.is_empty() {
+            self.recovery_pending.store(false, Ordering::Relaxed);
+        }
+        out
+    }
+
+    /// Every blob in `staging/`, paired with its sidecar when it has a readable
+    /// one.
+    ///
+    /// For the startup reconcile: staging holds the only copy of writes the
+    /// kernel has already been told succeeded, and three failure paths can
+    /// leave one there with no `pending_op` row pointing at it. Nothing walked
+    /// this directory, so those bytes were invisible to the daemon and to the
+    /// user both — `hydrate_pending` walks the op table and
+    /// [`recovered_writes`](Self::recovered_writes) walks `recovery/`.
+    pub fn staged_writes(&self) -> Vec<(PathBuf, Option<StagedWrite>)> {
+        let Ok(entries) = std::fs::read_dir(&self.staging_dir) else {
+            return Vec::new();
+        };
+        let mut out = Vec::new();
+        for entry in entries.flatten() {
+            let blob = entry.path();
+            if blob.extension().is_some_and(|e| e == "json") {
+                continue;
+            }
+            let meta = std::fs::read(Self::scratch_sidecar(&blob))
+                .ok()
+                .and_then(|b| serde_json::from_slice::<StagedWrite>(&b).ok());
+            out.push((blob, meta));
+        }
         out
     }
 
@@ -1226,6 +1289,10 @@ impl ContentCache {
                         if std::fs::rename(scratch, &recovery).is_ok() {
                             let _ = std::fs::File::open(&self.recovery_dir)
                                 .and_then(|dir| dir.sync_all());
+                            // These bytes are unaccompanied, but re-arm the
+                            // scan anyway: the flag exists to skip a directory
+                            // nothing has written to, and something just did.
+                            self.recovery_pending.store(true, Ordering::Relaxed);
                             return Err(retained(
                                 &recovery,
                                 Error::Other(format!(
@@ -1504,6 +1571,44 @@ impl ContentCache {
             .sum()
     }
 
+    /// Bytes and file count held in `staging/`, with the age of the oldest
+    /// entry in seconds.
+    ///
+    /// Staging is not part of [`usage`](Self::usage) and is not subject to the
+    /// budget — it holds the *only* copy of writes that have not been uploaded,
+    /// so nothing here may ever be evicted to reclaim space. That is precisely
+    /// why it has to be reportable: a parked transient create (three abandoned
+    /// 4 GiB downloads, say) otherwise occupies disk with nothing in any UI
+    /// accounting for it.
+    ///
+    /// A `read_dir` per call, which is fine for the callers it has: a status
+    /// request, polled every couple of seconds against a directory that is
+    /// empty whenever the queue is drained.
+    pub fn staging_usage(&self) -> StagingUsage {
+        let Ok(entries) = std::fs::read_dir(&self.staging_dir) else {
+            return StagingUsage::default();
+        };
+        let mut usage = StagingUsage::default();
+        let now = std::time::SystemTime::now();
+        for entry in entries.flatten() {
+            // Sidecars describe the blob beside them; counting both would double
+            // the file count and report metadata as user bytes.
+            if entry.path().extension().is_some_and(|e| e == "json") {
+                continue;
+            }
+            let Ok(meta) = entry.metadata() else { continue };
+            usage.files += 1;
+            usage.bytes += meta.len();
+            if let Ok(age) = meta.modified().and_then(|m| {
+                now.duration_since(m)
+                    .map_err(|_| std::io::Error::other("modified in the future"))
+            }) {
+                usage.oldest_secs = usage.oldest_secs.max(age.as_secs());
+            }
+        }
+        usage
+    }
+
     /// Configured soft byte cap (`0` = unlimited), for display alongside
     /// [`usage`](Self::usage). This is the whole-cache figure; internally it is
     /// divided between the pools (see [`SPLIT_BLOB`]), but the user set one

crates/pdfs-core/src/control.rs

@@ -1193,6 +1193,28 @@ pub enum Response {
         /// calling a queued `mkdir` an upload is a lie.
         #[serde(default)]
         pending_changes: u64,
+        /// Queued writes held back because the file still wears a transient
+        /// name (a browser's `*.crdownload`, an editor's `*.swp`). They are not
+        /// waiting on the network and will not drain until the finalising
+        /// rename, so they are counted apart from `pending_uploads`.
+        #[serde(default)]
+        parked_uploads: u64,
+        /// Queued ops that have failed enough times to be considered stuck.
+        /// They keep retrying forever — correctly, since their staged bytes are
+        /// the only copy — which is why they have to be visible.
+        #[serde(default)]
+        failing_ops: u64,
+        /// The most recent error from a stuck op, when there is one.
+        #[serde(default)]
+        failing_error: Option<String>,
+        /// Bytes held in `staging/`: writes accepted from the kernel whose only
+        /// copy is on local disk. Never evictable, never counted against the
+        /// cache budget — hence reported separately.
+        #[serde(default)]
+        staged_bytes: u64,
+        /// Age of the oldest staged write, in seconds.
+        #[serde(default)]
+        staged_oldest_secs: u64,
     },
     /// A human-readable success message.
     Ok { message: String },
@@ -1217,6 +1239,12 @@ pub enum Response {
         /// Whether the integrity check actually ran, so an empty problem list
         /// is not mistaken for a clean bill of health it never gave.
         integrity_checked: bool,
+        /// Whether a deep check is running right now. It reads every page of
+        /// the database under the daemon's only connection, so it is answered
+        /// asynchronously: this says "ask again shortly", which is a different
+        /// thing to say than either of the two above.
+        #[serde(default)]
+        integrity_running: bool,
     },
     /// The pin registry.
     Pins { pins: Vec<Pin> },

crates/pdfs-core/src/db/local.rs

@@ -41,32 +41,43 @@ impl Db {
         Ok(next)
     }
 
-    /// Write one batch of walked entries under scan generation `generation`. The
-    /// FTS index is *not* touched here — it is rebuilt once in
-    /// [`local_finish_scan`](Self::local_finish_scan).
+    /// Write one batch of walked entries under scan generation `generation`,
+    /// keeping the FTS index in step with them.
+    ///
+    /// The index is maintained incrementally rather than rebuilt at the end of
+    /// the scan, because a rebuild reads and re-tokenises every row in the table
+    /// — trigrams over a few hundred thousand paths — inside one transaction on
+    /// the single shared connection, which stalls the whole mount for as long as
+    /// it takes. Incremental maintenance is possible here for a reason specific
+    /// to this table: `local_fts` covers `name` and `path`, `path` is the
+    /// table's key, and `name` is derived from it, so an existing row's indexed
+    /// content can never change. Only insertions and deletions touch the index;
+    /// a re-scan that finds the same files does no index work at all.
+    ///
+    /// Hence the update-then-insert shape below rather than an upsert: the
+    /// distinction between "already indexed" and "new row" is exactly what
+    /// decides whether the index needs a write.
     pub fn local_upsert_batch(&self, generation: i64, entries: &[LocalEntry]) -> Result<()> {
         let mut conn = self.conn.lock();
         let tx = conn.transaction()?;
         {
-            let mut stmt = tx.prepare_cached(
-                "INSERT INTO local_files (path, name, is_dir, size, mtime, scan_gen)
-                 VALUES (?1, ?2, ?3, ?4, ?5, ?6)
-                 ON CONFLICT(path) DO UPDATE SET
-                   name     = excluded.name,
-                   is_dir   = excluded.is_dir,
-                   size     = excluded.size,
-                   mtime    = excluded.mtime,
-                   scan_gen = excluded.scan_gen",
+            let mut update = tx.prepare_cached(
+                "UPDATE local_files
+                    SET name = ?2, is_dir = ?3, size = ?4, mtime = ?5, scan_gen = ?6
+                  WHERE path = ?1",
             )?;
+            let mut insert = tx.prepare_cached(
+                "INSERT INTO local_files (path, name, is_dir, size, mtime, scan_gen)
+                 VALUES (?1, ?2, ?3, ?4, ?5, ?6)",
+            )?;
+            let mut index =
+                tx.prepare_cached("INSERT INTO local_fts (rowid, name, path) VALUES (?1, ?2, ?3)")?;
             for e in entries {
-                stmt.execute(params![
-                    e.path,
-                    e.name,
-                    e.is_dir as i64,
-                    e.size,
-                    e.mtime,
-                    generation
-                ])?;
+                let row = params![e.path, e.name, e.is_dir as i64, e.size, e.mtime, generation];
+                if update.execute(row)? == 0 {
+                    insert.execute(row)?;
+                    index.execute(params![tx.last_insert_rowid(), e.name, e.path])?;
+                }
             }
         }
         tx.commit()?;
@@ -74,16 +85,25 @@ impl Db {
     }
 
     /// Close scan generation `generation`: drop every row an older scan wrote
-    /// (those paths are gone from disk), rebuild the FTS index over what remains,
-    /// and stamp the completion time. Returns the number of indexed entries.
+    /// (those paths are gone from disk), drop their FTS entries, and stamp the
+    /// completion time. Returns the number of indexed entries.
+    ///
+    /// The FTS deletes name the old values explicitly, which is how an
+    /// external-content FTS5 table is told to retract a row — see
+    /// [`local_upsert_batch`](Self::local_upsert_batch) for why this is done
+    /// per-row instead of by rebuilding the index.
     pub fn local_finish_scan(&self, generation: i64, finished_at: i64) -> Result<i64> {
         let mut conn = self.conn.lock();
         let tx = conn.transaction()?;
+        tx.execute(
+            "INSERT INTO local_fts (local_fts, rowid, name, path)
+             SELECT 'delete', id, name, path FROM local_files WHERE scan_gen != ?1",
+            params![generation],
+        )?;
         tx.execute(
             "DELETE FROM local_files WHERE scan_gen != ?1",
             params![generation],
         )?;
-        tx.execute_batch("INSERT INTO local_fts(local_fts) VALUES('rebuild');")?;
         tx.execute(
             "INSERT INTO sync_state (key, value) VALUES ('local_indexed_at', ?1)
              ON CONFLICT(key) DO UPDATE SET value = excluded.value",

crates/pdfs-core/src/db/migrations.rs

@@ -9,7 +9,7 @@ use super::Db;
 use crate::Result;
 
 /// Current schema version. Bump on every forward migration added below.
-pub(super) const SCHEMA_VERSION: i64 = 22;
+pub(super) const SCHEMA_VERSION: i64 = 24;
 
 impl Db {
     pub(super) fn migrate(&self) -> Result<()> {
@@ -131,6 +131,24 @@ impl Db {
             // order by `id`: it has only ever cost writes.
             tx.execute_batch("DROP INDEX IF EXISTS activity_time;")?;
         }
+        if current < 23 {
+            // One last full rebuild of the local index, because from here on the
+            // scan maintains it incrementally (`local_upsert_batch`) and so can
+            // no longer repair a `local_fts` that drifted from `local_files`.
+            // Whatever it costs, it costs once, at startup, before the mount
+            // serves anything.
+            let present: bool = tx.query_row(
+                "SELECT COUNT(*) FROM sqlite_master WHERE type = 'table' AND name = 'local_fts'",
+                [],
+                |row| row.get::<_, i64>(0),
+            )? > 0;
+            if present {
+                tx.execute_batch("INSERT INTO local_fts(local_fts) VALUES('rebuild');")?;
+            }
+        }
+        if current < 24 {
+            tx.execute_batch(MIGRATION_V24)?;
+        }
         tx.execute(
             "INSERT INTO sync_state (key, value) VALUES ('schema_version', ?1)
              ON CONFLICT(key) DO UPDATE SET value = excluded.value",
@@ -633,3 +651,22 @@ const MIGRATION_V22: [(&str, &str); 3] = [
         "CREATE INDEX IF NOT EXISTS idx_nodes_parent_live ON nodes(parent_uid) WHERE trashed = 0;",
     ),
 ];
+
+/// V24: the revisions this daemon sealed itself, so restarting does not reopen
+/// the conflict hole B70 layer B closed.
+///
+/// `Core::own_sealed_revs` held these in memory: a queued write whose baseline
+/// names an earlier revision is a conflict *unless* the remote sits at a
+/// revision we sealed, in which case it is one writer stalling and resuming and
+/// must supersede rather than fork a `(sync-conflict)` copy. Losing the map at
+/// shutdown meant the first drain after a restart could not tell the two apart.
+///
+/// `sealed_at` is what bounds the table: entries older than the window in which
+/// a stall→resume is plausible are pruned on write.
+const MIGRATION_V24: &str = "
+CREATE TABLE IF NOT EXISTS own_sealed_rev (
+  uid         TEXT PRIMARY KEY,
+  revision_id TEXT NOT NULL,
+  sealed_at   INTEGER NOT NULL
+);
+";

crates/pdfs-core/src/db/mod.rs

@@ -18,8 +18,10 @@
 use parking_lot::Mutex;
 use rusqlite::Connection;
 
-use crate::Result;
-use std::path::Path;
+use crate::{Error, Result};
+use std::fs::File;
+use std::os::fd::AsRawFd;
+use std::path::{Path, PathBuf};
 
 mod activity;
 mod albums;
@@ -86,12 +88,50 @@ const WAL_AUTOCHECKPOINT_PAGES: i64 = 500;
 /// the inner `Mutex<Connection>`.
 pub struct Db {
     conn: Mutex<Connection>,
+    /// The single-writer lock, held open for as long as this handle lives.
+    ///
+    /// Never read; dropping the `File` is what releases the `flock`, so it has
+    /// to be owned here rather than by `open`'s stack frame. `None` for
+    /// in-memory databases, which nothing else can reach.
+    _single_writer: Option<File>,
 }
 
 impl Db {
     /// Open (creating if absent) the database at `path`, enable WAL, and bring
     /// the schema up to [`SCHEMA_VERSION`].
+    ///
+    /// Takes the single-instance lock first (see
+    /// [`acquire_single_writer_lock`]), and rebuilds from empty if what is on
+    /// disk turns out not to be a readable database (see [`is_corrupt`]).
     pub fn open(path: &Path) -> Result<Self> {
+        let single_writer = acquire_single_writer_lock(path)?;
+        let conn = match Self::open_configured(path) {
+            Ok(conn) => conn,
+            Err(e) if is_corrupt(&e) => {
+                // Every byte in here is derived from the account or from the
+                // content cache directory, so the recovery is to start over —
+                // loudly, and keeping the damaged file for a post-mortem rather
+                // than deleting evidence.
+                tracing::error!(
+                    db = %path.display(),
+                    error = %e,
+                    "cache database is corrupt — moving it aside and rebuilding \
+                     (metadata will be re-fetched; no user data is stored here)"
+                );
+                quarantine_corrupt_db(path)?;
+                Self::open_configured(path)?
+            }
+            Err(e) => return Err(e),
+        };
+        Ok(Self {
+            conn: Mutex::new(conn),
+            _single_writer: single_writer,
+        })
+    }
+
+    /// Open the file, apply every PRAGMA, and migrate. Split out of
+    /// [`open`](Self::open) so the corruption path can retry it verbatim.
+    fn open_configured(path: &Path) -> Result<Connection> {
         let conn = Connection::open(path)?;
         // WAL: readers never block the single writer. NORMAL sync is the
         // standard durability/throughput tradeoff for WAL.
@@ -137,9 +177,17 @@ impl Db {
 
         let db = Self {
             conn: Mutex::new(conn),
+            _single_writer: None,
         };
         db.migrate()?;
-        Ok(db)
+        Ok(db.into_conn())
+    }
+
+    /// Unwrap the connection back out of a temporary handle. Only
+    /// [`open_configured`](Self::open_configured) uses it, to run `migrate`
+    /// (which is written against `&self`) before the real handle exists.
+    fn into_conn(self) -> Connection {
+        self.conn.into_inner()
     }
 
     /// Open an in-memory database. For tests.
@@ -152,6 +200,7 @@ impl Db {
         conn.pragma_update(None, "foreign_keys", "ON")?;
         let db = Self {
             conn: Mutex::new(conn),
+            _single_writer: None,
         };
         db.migrate()?;
         Ok(db)
@@ -165,5 +214,102 @@ impl Db {
     }
 }
 
+/// Whether `path` names the in-memory database rather than a file. Both
+/// spellings SQLite accepts are checked, because neither has a lock file or a
+/// corruption story.
+fn is_in_memory(path: &Path) -> bool {
+    let p = path.as_os_str();
+    p == ":memory:" || p.is_empty()
+}
+
+/// Take the single-instance write lock for the database at `path`.
+///
+/// The daemon is the only writer by design — the front-ends go through the
+/// control socket — but nothing stopped a hand-run `pdfs mount` from starting
+/// alongside the systemd unit, at which point two processes own the same inode
+/// space, the same content cache, and the same drain queue. SQLite's own
+/// locking makes each *statement* safe and does nothing about that.
+///
+/// `flock` on a sibling `.lock` file rather than on the database itself:
+/// SQLite uses POSIX record locks, so the two never interact, and a lock file
+/// left behind carries no state — the lock lives in the kernel and dies with
+/// the process, however it exits.
+fn acquire_single_writer_lock(path: &Path) -> Result<Option<File>> {
+    if is_in_memory(path) {
+        return Ok(None);
+    }
+    let lock_path = lock_path_for(path);
+    let file = std::fs::OpenOptions::new()
+        .create(true)
+        .read(true)
+        .truncate(false)
+        .write(true)
+        .open(&lock_path)?;
+    // SAFETY: `file` owns the descriptor and outlives the call.
+    let rc = unsafe { libc::flock(file.as_raw_fd(), libc::LOCK_EX | libc::LOCK_NB) };
+    if rc != 0 {
+        let e = std::io::Error::last_os_error();
+        if e.raw_os_error() == Some(libc::EWOULDBLOCK) {
+            return Err(Error::Other(format!(
+                "another Proton Drive daemon is already using {} — \
+                 stop it first (`systemctl --user stop proton-drive.service`)",
+                path.display()
+            )));
+        }
+        return Err(Error::Io(e));
+    }
+    Ok(Some(file))
+}
+
+fn lock_path_for(path: &Path) -> PathBuf {
+    let mut name = path.as_os_str().to_owned();
+    name.push(".lock");
+    PathBuf::from(name)
+}
+
+/// Whether `e` says the file is not a usable database, as opposed to any other
+/// failure. These are the only two conditions rebuilding fixes; a disk error, a
+/// permission problem or a newer-than-supported schema must all still fail.
+fn is_corrupt(e: &Error) -> bool {
+    matches!(
+        e,
+        Error::Db(rusqlite::Error::SqliteFailure(inner, _))
+            if inner.code == rusqlite::ErrorCode::DatabaseCorrupt
+                || inner.code == rusqlite::ErrorCode::NotADatabase
+    )
+}
+
+/// Move a damaged database (and its WAL sidecars) out of the way so the next
+/// open starts from empty.
+///
+/// Renamed rather than deleted: this is a cache, but it is also the only
+/// evidence of what went wrong, and it costs one file the user can remove.
+fn quarantine_corrupt_db(path: &Path) -> Result<()> {
+    let stamp = std::time::SystemTime::now()
+        .duration_since(std::time::UNIX_EPOCH)
+        .map(|d| d.as_secs())
+        .unwrap_or(0);
+    for suffix in ["", "-wal", "-shm"] {
+        let mut from = path.as_os_str().to_owned();
+        from.push(suffix);
+        let from = PathBuf::from(from);
+        if !from.exists() {
+            continue;
+        }
+        let mut to = path.as_os_str().to_owned();
+        to.push(format!(".corrupt-{stamp}{suffix}"));
+        let to = PathBuf::from(to);
+        // A failure to rename the sidecars is survivable — SQLite discards a
+        // WAL whose database header no longer matches — but failing to move the
+        // database itself is not, because the retry would hit the same file.
+        match std::fs::rename(&from, &to) {
+            Ok(()) => tracing::warn!(from = %from.display(), to = %to.display(), "quarantined"),
+            Err(e) if suffix.is_empty() => return Err(Error::Io(e)),
+            Err(e) => tracing::warn!(file = %from.display(), error = %e, "could not quarantine"),
+        }
+    }
+    Ok(())
+}
+
 #[cfg(test)]
 mod tests;

crates/pdfs-core/src/db/ops.rs

@@ -119,15 +119,32 @@ pub struct PendingOp {
 }
 
 /// How much work the queue owes the server, by whether it carries bytes.
-#[derive(Debug, Clone, Copy, Default)]
+#[derive(Debug, Clone, Default)]
 pub struct PendingCounts {
     /// Queued `create`/`revision` ops: files whose content is not on the remote.
     pub uploads: i64,
     /// Queued `mkdir`/`rename`/`trash` ops: metadata the remote has not been
     /// told about.
     pub changes: i64,
+    /// Ops parked at [`PARK_UNTIL`]: a transient file's bytes, held back until
+    /// the rename that gives it its finished name. They are queued but will
+    /// never drain on their own, so counting them as "waiting to upload" would
+    /// have the front-end promise progress that is not coming.
+    pub parked: i64,
+    /// Ops that have failed at least [`FAILING_ATTEMPTS`] times. A permanently
+    /// failing op never wedges the queue — the backoff sees to that — which is
+    /// exactly why it needs surfacing: it retries forever, invisibly.
+    pub failing: i64,
+    /// The most recent error text from a failing op, for the front-end to show
+    /// instead of the bare count.
+    pub last_error: Option<String>,
 }
 
+/// Failed attempts after which an op is reported as failing rather than merely
+/// retrying. Six is where the backoff has reached its ceiling, so an op past it
+/// is retrying at the slowest rate it ever will and has been for minutes.
+pub const FAILING_ATTEMPTS: i64 = 6;
+
 /// The outcome of folding freshly written bytes into a queued create.
 #[derive(Debug, Clone)]
 pub struct AttachedBlob {
@@ -564,7 +581,33 @@ impl Db {
             params![OP_REVISION, OP_CREATE],
             |r| r.get(0),
         )?;
-        Ok(PendingCounts { uploads, changes })
+        let parked = conn.query_row(
+            "SELECT COUNT(*) FROM pending_op WHERE next_attempt_at >= ?1",
+            params![PARK_UNTIL],
+            |r| r.get(0),
+        )?;
+        let failing = conn.query_row(
+            "SELECT COUNT(*) FROM pending_op WHERE attempts >= ?1",
+            params![FAILING_ATTEMPTS],
+            |r| r.get(0),
+        )?;
+        let last_error = conn
+            .query_row(
+                "SELECT last_error FROM pending_op
+                  WHERE attempts >= ?1 AND last_error IS NOT NULL
+                  ORDER BY attempts DESC LIMIT 1",
+                params![FAILING_ATTEMPTS],
+                |r| r.get::<_, Option<String>>(0),
+            )
+            .optional()?
+            .flatten();
+        Ok(PendingCounts {
+            uploads,
+            changes,
+            parked,
+            failing,
+            last_error,
+        })
     }
 
     /// Nothing bearing this volume may be handed to the API — it would 404. The
@@ -609,4 +652,78 @@ impl Db {
         )?;
         Ok(())
     }
+
+    /// Every staged blob the queue still refers to.
+    ///
+    /// The startup staging reconcile subtracts this from what is actually in
+    /// `staging/`: the difference is bytes the kernel was told were written and
+    /// that nothing is going to upload.
+    pub fn op_blob_paths(&self) -> Result<std::collections::HashSet<String>> {
+        let conn = self.conn.lock();
+        let mut stmt =
+            conn.prepare("SELECT blob_path FROM pending_op WHERE blob_path IS NOT NULL")?;
+        let rows = stmt.query_map([], |r| r.get::<_, String>(0))?;
+        let mut out = std::collections::HashSet::new();
+        for row in rows {
+            out.insert(row?);
+        }
+        Ok(out)
+    }
+
+    /// Record that this daemon sealed `revision_id` on `uid`, and prune entries
+    /// past [`OWN_SEALED_TTL_MS`].
+    ///
+    /// One row per node, overwritten as the node advances. See
+    /// [`own_sealed_rev`](Self::own_sealed_rev) for what reads it.
+    pub fn set_own_sealed_rev(&self, uid: &str, revision_id: &str, now_ms: i64) -> Result<()> {
+        let conn = self.conn.lock();
+        conn.execute(
+            "INSERT INTO own_sealed_rev (uid, revision_id, sealed_at)
+             VALUES (?1, ?2, ?3)
+             ON CONFLICT(uid) DO UPDATE SET
+               revision_id = excluded.revision_id,
+               sealed_at   = excluded.sealed_at",
+            params![uid, revision_id, now_ms],
+        )?;
+        conn.execute(
+            "DELETE FROM own_sealed_rev WHERE sealed_at < ?1",
+            params![now_ms - OWN_SEALED_TTL_MS],
+        )?;
+        Ok(())
+    }
+
+    /// The last revision of `uid` this daemon sealed itself, if it is still
+    /// within [`OWN_SEALED_TTL_MS`].
+    ///
+    /// The drain asks before treating a moved-on remote as a conflict: a
+    /// revision we sealed means one writer stalled and resumed, which
+    /// supersedes rather than forking a `(sync-conflict)` copy (docs/BUGS.md
+    /// B70 layer B).
+    pub fn own_sealed_rev(&self, uid: &str, now_ms: i64) -> Result<Option<String>> {
+        let conn = self.conn.lock();
+        Ok(conn
+            .query_row(
+                "SELECT revision_id FROM own_sealed_rev
+                  WHERE uid = ?1 AND sealed_at >= ?2",
+                params![uid, now_ms - OWN_SEALED_TTL_MS],
+                |r| r.get(0),
+            )
+            .optional()?)
+    }
+
+    /// Forget what we sealed on `uid` — it has been trashed, so there is no
+    /// revision left to chain onto.
+    pub fn clear_own_sealed_rev(&self, uid: &str) -> Result<()> {
+        let conn = self.conn.lock();
+        conn.execute("DELETE FROM own_sealed_rev WHERE uid = ?1", params![uid])?;
+        Ok(())
+    }
 }
+
+/// How long a revision this daemon sealed stays recognisable as its own.
+///
+/// Bounds a table that would otherwise grow with every write for the life of
+/// the install. A week is far longer than any stall→resume — a browser download
+/// that pauses and continues, an editor that reopens its handle — and short
+/// enough that the table tracks recent work rather than accumulating history.
+pub const OWN_SEALED_TTL_MS: i64 = 7 * 24 * 60 * 60 * 1000;

crates/pdfs-core/src/db/tests.rs

@@ -867,6 +867,67 @@ fn shared_root_pinned_name_survives_database_reopen() {
     let _ = std::fs::remove_file(path);
 }
 
+/// Two writers on one database is the failure mode the single-instance lock
+/// exists to prevent — a hand-run `pdfs mount` next to the systemd unit gives
+/// two daemons the same inode space, content cache, and drain queue. The second
+/// open has to fail, and it has to stop failing once the first handle is gone.
+#[test]
+fn a_second_writer_is_refused_while_the_first_holds_the_database() {
+    let path = std::env::temp_dir().join(format!(
+        "pdfs-single-writer-{}-{}.db",
+        std::process::id(),
+        now_test_id()
+    ));
+    let first = Db::open(&path).unwrap();
+    let second = Db::open(&path);
+    assert!(second.is_err(), "a second writer was allowed in");
+
+    drop(first);
+    // The lock lives in the kernel and dies with the descriptor, so nothing has
+    // to be cleaned up for the next daemon to start.
+    Db::open(&path).expect("reopen after the first handle is dropped");
+
+    let _ = std::fs::remove_file(&path);
+    let _ = std::fs::remove_file(path.with_extension("db.lock"));
+}
+
+/// `cache.db` is derived state — every row can be re-fetched or re-scanned — so
+/// a file that is not a database at all must not stop the daemon starting. It
+/// is moved aside (not deleted: it is the only evidence) and rebuilt.
+#[test]
+fn a_corrupt_database_is_moved_aside_and_rebuilt() {
+    let path = std::env::temp_dir().join(format!(
+        "pdfs-corrupt-{}-{}.db",
+        std::process::id(),
+        now_test_id()
+    ));
+    std::fs::write(&path, b"this is not an SQLite database").unwrap();
+
+    let db = Db::open(&path).unwrap();
+    db.set_state_str("shared_with_me_name", "rebuilt").unwrap();
+    assert_eq!(
+        db.state_str("shared_with_me_name").unwrap().as_deref(),
+        Some("rebuilt")
+    );
+    drop(db);
+
+    let dir = path.parent().unwrap();
+    let stem = path.file_name().unwrap().to_string_lossy().to_string();
+    let quarantined: Vec<_> = std::fs::read_dir(dir)
+        .unwrap()
+        .filter_map(|e| e.ok())
+        .map(|e| e.file_name().to_string_lossy().to_string())
+        .filter(|name| name.starts_with(&format!("{stem}.corrupt-")))
+        .collect();
+    assert_eq!(quarantined.len(), 1, "the damaged file was not kept");
+
+    for name in quarantined {
+        let _ = std::fs::remove_file(dir.join(name));
+    }
+    let _ = std::fs::remove_file(&path);
+    let _ = std::fs::remove_file(path.with_extension("db.lock"));
+}
+
 fn local(path: &str, name: &str, is_dir: bool) -> LocalEntry {
     LocalEntry {
         path: path.into(),
@@ -911,6 +972,25 @@ fn local_scan_indexes_then_prunes_stale_paths() {
     assert_eq!(db.search_local("notes", 10).unwrap().len(), 1);
 }
 
+/// The index is maintained incrementally rather than rebuilt at the end of a
+/// scan, so the case that used to be covered for free — a file the scan sees
+/// again — needs its own test: a re-scan must neither lose the entry nor
+/// duplicate it.
+#[test]
+fn rescanning_an_unchanged_file_keeps_exactly_one_index_entry() {
+    let db = Db::open_in_memory().unwrap();
+    for (generation, mtime) in [(1_000, 5), (2_000, 9)] {
+        let g = db.local_begin_scan().unwrap();
+        let mut e = local("/home/u/docs/report.pdf", "report.pdf", false);
+        e.mtime = mtime;
+        db.local_upsert_batch(g, &[e]).unwrap();
+        assert_eq!(db.local_finish_scan(g, generation).unwrap(), 1);
+        let hits = db.search_local("report", 10).unwrap();
+        assert_eq!(hits.len(), 1, "generation {generation}");
+        assert_eq!(hits[0].mtime, mtime);
+    }
+}
+
 /// Queries below the trigram minimum still match, via the `LIKE` fallback.
 #[test]
 fn local_search_short_query_like_fallback() {

crates/pdfs-fuse/src/control.rs

@@ -148,7 +148,13 @@ fn handle_control_conn(core: &Core, username: &str, mountpoint: &Path, stream: U
         Ok(CtlRequest::Status) => {
             let pins = core.cache.list_pins();
             let queued = core.db.pending_op_counts().unwrap_or_default();
+            let staging = core.cache.staging_usage();
             CtlResponse::Status {
+                parked_uploads: queued.parked.max(0) as u64,
+                failing_ops: queued.failing.max(0) as u64,
+                failing_error: queued.last_error.clone(),
+                staged_bytes: staging.bytes,
+                staged_oldest_secs: staging.oldest_secs,
                 username: username.to_string(),
                 mountpoint: mountpoint.display().to_string(),
                 pinned: pins.len(),
@@ -562,19 +568,20 @@ fn handle_control_conn(core: &Core, username: &str, mountpoint: &Path, stream: U
         }
         Ok(CtlRequest::CacheInspect { deep }) => match core.db.stats() {
             Ok(stats) => {
-                // The deep check reads every page, so it runs only on request;
-                // `integrity_checked` tells the caller which answer it is
-                // holding, since "no problems found" and "never looked" are
-                // both an empty list.
-                let (integrity_problems, integrity_checked) = if deep {
-                    match core.db.integrity_check() {
-                        Ok(problems) => (problems, true),
-                        Err(e) => (vec![format!("integrity check failed: {e:?}")], true),
-                    }
+                // The deep check reads every page — tens of seconds on a real
+                // database, all of it holding the connection every FUSE handler
+                // needs — so it runs on a thread and this request returns
+                // whatever the last completed check found. `integrity_checked`
+                // tells the caller which answer it is holding, since "no
+                // problems found" and "never looked" are both an empty list;
+                // `integrity_running` says a fresh answer is on its way.
+                let (integrity_problems, integrity_checked, integrity_running) = if deep {
+                    core.start_or_read_integrity_check()
                 } else {
-                    (Vec::new(), false)
+                    (Vec::new(), false, false)
                 };
                 CtlResponse::CacheReport {
+                    integrity_running,
                     schema_version: stats.schema_version,
                     db_bytes: stats.total_bytes(),
                     db_reclaimable_bytes: stats.reclaimable_bytes(),
@@ -587,17 +594,16 @@ fn handle_control_conn(core: &Core, username: &str, mountpoint: &Path, stream: U
             }
             Err(e) => CtlResponse::error(CoreError::internal(format!("read db stats: {e:?}"))),
         },
-        Ok(CtlRequest::CacheVacuum) => match core.db.vacuum() {
-            Ok(outcome) => CtlResponse::Ok {
-                message: format!(
-                    "vacuumed: {:.1} MiB freed ({:.1} MiB → {:.1} MiB), {} WAL frame(s) checkpointed",
-                    outcome.freed_bytes() as f64 / 1_048_576.0,
-                    outcome.before_bytes as f64 / 1_048_576.0,
-                    outcome.after_bytes as f64 / 1_048_576.0,
-                    outcome.wal_frames_checkpointed,
-                ),
+        // Ack and vacuum on a thread. `VACUUM` rewrites the whole database
+        // under the connection mutex — minutes on a large one — and the
+        // request's own socket timeout is 10 s, so running it inline both froze
+        // the mount and lost its own answer. Progress shows up as a job in
+        // `GetQueueStatus`; the outcome goes to the log.
+        Ok(CtlRequest::CacheVacuum) => match core.start_vacuum() {
+            Ok(()) => CtlResponse::Ok {
+                message: "compacting the cache database in the background".into(),
             },
-            Err(e) => CtlResponse::error(CoreError::internal(format!("vacuum: {e:?}"))),
+            Err(e) => CtlResponse::error(e),
         },
         Ok(CtlRequest::GetQueueStatus) => CtlResponse::Transfers {
             items: core.transfers.snapshot(),

crates/pdfs-fuse/src/drain.rs

@@ -355,6 +355,9 @@ impl Core {
         }
         self.db.complete_trash_op(op.id, &uid)?;
         self.own_sealed_revs.lock().remove(&uid);
+        if let Err(e) = self.db.clear_own_sealed_rev(&uid.to_string()) {
+            debug!(%uid, error = %e, "clearing the sealed-revision record failed");
+        }
         self.invalidate_trash();
         self.log_activity(ActivityKind::Trash, &name, "trashed", true);
         info!(%uid, name, "pending trash landed");
@@ -629,7 +632,16 @@ impl Core {
         // instead of forking — so the finished bytes win the name rather than
         // being exiled to a `(sync-conflict)` copy.
         let remote_rev = node_revision_id(&node);
-        let own_rev = self.own_sealed_revs.lock().get(uid).cloned();
+        // In memory first, then the table it is written through to: the map is
+        // this process's cache of it, and after a restart only the table has
+        // the answer (which is the whole reason it is persisted).
+        let own_rev = match self.own_sealed_revs.lock().get(uid).cloned() {
+            Some(rev) => Some(rev),
+            None => self
+                .db
+                .own_sealed_rev(&uid.to_string(), now_millis())
+                .unwrap_or_default(),
+        };
         if is_own_self_supersede(meta.complete, remote_rev.as_deref(), own_rev.as_deref()) {
             debug!(%uid, ?remote_rev,
                    "queued write chains onto our own sealed revision; not a conflict");
@@ -985,7 +997,16 @@ impl Core {
             // change and does not fork (B70 layer B; consulted in
             // `revision_conflict`).
             if let Some(rev) = sealed_rev.clone() {
-                self.own_sealed_revs.lock().insert(uid.clone(), rev);
+                self.own_sealed_revs.lock().insert(uid.clone(), rev.clone());
+                // Write through: losing this at shutdown reopened the fork
+                // window B70 layer B closed, for the first drain after a
+                // restart.
+                if let Err(e) = self
+                    .db
+                    .set_own_sealed_rev(&uid.to_string(), &rev, now_millis())
+                {
+                    warn!(%uid, error = %e, "persisting our sealed revision failed");
+                }
             }
             self.for_each_state(|st| {
                 for aw in st.active_writes.values_mut() {

crates/pdfs-fuse/src/lib.rs

@@ -425,6 +425,14 @@ struct Core {
     /// storage moves slowly enough that a minute-old number is a better answer
     /// than either a stall or the zeroes the default implementation returns.
     quota: Arc<Mutex<Option<(std::time::Instant, i64, i64)>>>,
+    /// Database maintenance that has been moved off the request thread.
+    ///
+    /// `VACUUM` and a deep `integrity_check` both hold the single connection for
+    /// tens of seconds against a real install, which stops every FUSE handler
+    /// that needs the database — so the handler acknowledges the request and a
+    /// thread does the work, exactly as the other long control requests already
+    /// do. See [`Maintenance`].
+    maintenance: Arc<Mutex<Maintenance>>,
     /// Folders whose listing was enumerated cheaply and is having its file sizes
     /// filled in right now, so a burst of `stat`s over a fresh listing starts one
     /// upgrade rather than one per entry. See [`Core::spawn_size_upgrade`].
@@ -510,6 +518,21 @@ struct Core {
     states: Arc<StateRegistry>,
 }
 
+/// Background database maintenance: what is running, and what the last deep
+/// integrity check found.
+///
+/// One at a time — both operations are whole-database and there is nothing to
+/// gain from overlapping them — and the result of a check outlives the request
+/// that asked for it, because the request is answered before the check finishes.
+#[derive(Default)]
+struct Maintenance {
+    /// A vacuum or integrity check is running right now.
+    running: bool,
+    /// Findings of the most recent completed deep check. `None` means none has
+    /// finished this run, which is *not* the same as "found nothing".
+    integrity: Option<Vec<String>>,
+}
+
 /// How long a change this daemon made stays recognisable as its own. Generous
 /// against a feed that can lag, but finite: past it, an event for the node is
 /// applied like any other. See [`Core::self_changes`].
@@ -695,6 +718,82 @@ enum SwitchBlocked {
 }
 
 impl Core {
+    /// Start a background `VACUUM`, or refuse because maintenance is already
+    /// running.
+    ///
+    /// Returns as soon as the thread is spawned; the caller acknowledges the
+    /// request. The outcome is a log line and a job that disappears from
+    /// `GetQueueStatus` when it is done — a vacuum has nothing to report that a
+    /// follow-up `CacheInspect` cannot show.
+    fn start_vacuum(&self) -> CoreResult<()> {
+        {
+            let mut m = self.maintenance.lock();
+            if m.running {
+                return Err(CoreError::conflict(
+                    "database maintenance is already running",
+                ));
+            }
+            m.running = true;
+        }
+        let core = self.clone();
+        std::thread::spawn(move || {
+            let job = core.transfers.begin_job("Compacting the cache database");
+            job.detail("Rewriting the database file");
+            let started = Instant::now();
+            match core.db.vacuum() {
+                Ok(outcome) => info!(
+                    freed_bytes = outcome.freed_bytes(),
+                    before_bytes = outcome.before_bytes,
+                    after_bytes = outcome.after_bytes,
+                    wal_frames = outcome.wal_frames_checkpointed,
+                    elapsed_ms = started.elapsed().as_millis() as u64,
+                    "cache database vacuumed"
+                ),
+                Err(e) => warn!(error = %e, "vacuum failed"),
+            }
+            core.maintenance.lock().running = false;
+        });
+        Ok(())
+    }
+
+    /// Answer a deep `CacheInspect`: hand back the last completed integrity
+    /// check, and start one if none is running.
+    ///
+    /// Returns `(problems, checked, running)`. A first request gets
+    /// `(_, false, true)` and the answer lands on a later one — the check reads
+    /// every page of the database under the connection mutex, so the choice is
+    /// between answering late and freezing the mount until it finishes.
+    fn start_or_read_integrity_check(&self) -> (Vec<String>, bool, bool) {
+        let mut m = self.maintenance.lock();
+        if let Some(problems) = m.integrity.clone() {
+            return (problems, true, m.running);
+        }
+        if m.running {
+            return (Vec::new(), false, true);
+        }
+        m.running = true;
+        drop(m);
+
+        let core = self.clone();
+        std::thread::spawn(move || {
+            let job = core.transfers.begin_job("Checking the cache database");
+            job.detail("Verifying every page");
+            let problems = match core.db.integrity_check() {
+                Ok(problems) => problems,
+                Err(e) => vec![format!("integrity check failed: {e:?}")],
+            };
+            if problems.is_empty() {
+                info!("cache database integrity check found no problems");
+            } else {
+                warn!(problems = problems.len(), "cache database integrity check");
+            }
+            let mut m = core.maintenance.lock();
+            m.integrity = Some(problems);
+            m.running = false;
+        });
+        (Vec::new(), false, true)
+    }
+
     /// Register an inode space so background work can reach it. Called once for
     /// the primary mount and once per on-demand fork.
     fn register_state(&self, mountpoint: &Path) {
@@ -1218,6 +1317,100 @@ impl Core {
         }
     }
 
+    /// Re-attach staged blobs that no queued op refers to.
+    ///
+    /// `staging/` holds the only copy of writes the kernel has already been
+    /// told succeeded. Three release-path failures put bytes there and return
+    /// `EIO` without leaving a `pending_op` row behind — an access change
+    /// mid-release, a write overlapping an undrained incomplete edit, a create
+    /// that drained out from under its own handle. Until now nothing ever
+    /// walked the directory again: `hydrate_pending` reads the op table and
+    /// [`recover_fsynced_writes`](Self::recover_fsynced_writes) reads
+    /// `recovery/`, so an orphaned staged blob was invisible to the daemon and
+    /// to the user alike, and stayed that way until someone found the file.
+    ///
+    /// Runs once at mount, after `hydrate_pending`, so `pending` already
+    /// describes what the queue owns. An orphan whose sidecar names a real node
+    /// is queued as an ordinary revision — the condition that made it fail is
+    /// gone by definition, since the op it collided with has since drained.
+    /// Anything that cannot be addressed (no sidecar, an unparseable uid, a
+    /// `local~` placeholder whose create is gone) is left in place and named in
+    /// the log: unexplained bytes are never evidence that user data is
+    /// disposable.
+    fn reconcile_staging(&self) {
+        let staged = self.cache.staged_writes();
+        if staged.is_empty() {
+            return;
+        }
+        let claimed = match self.db.op_blob_paths() {
+            Ok(paths) => paths,
+            Err(e) => {
+                error!(error = %e, "cannot read queued blob paths; skipping staging reconcile");
+                return;
+            }
+        };
+        let mut queued = 0usize;
+        let mut stranded = 0usize;
+        for (blob, meta) in staged {
+            if claimed.contains(&blob.to_string_lossy().into_owned()) {
+                continue;
+            }
+            let addressable = meta.as_ref().and_then(|m| parse_node_uid(&m.uid));
+            let Some((meta, uid)) = meta.zip(addressable).filter(|(_, uid)| !is_local_uid(uid))
+            else {
+                stranded += 1;
+                warn!(blob = %blob.display(),
+                      "staged bytes belong to no queued upload and name no node; kept for recovery");
+                continue;
+            };
+            if self.pending.lock().contains_key(&uid) {
+                // A newer write for the same node is queued and carries its own
+                // blob. These bytes are older; keeping them is right, uploading
+                // them over the newer ones is not.
+                stranded += 1;
+                warn!(%uid, blob = %blob.display(),
+                      "staged bytes are superseded by a queued write; kept for recovery");
+                continue;
+            }
+            let op = PendingOp {
+                id: 0,
+                kind: OP_REVISION.to_string(),
+                uid: uid.to_string(),
+                parent_uid: None,
+                name: None,
+                blob_path: Some(blob.to_string_lossy().into_owned()),
+                meta_json: Some(serde_json::to_string(&meta).unwrap_or_default()),
+                created_at: now_millis(),
+                attempts: 0,
+                last_error: None,
+                next_attempt_at: now_millis(),
+            };
+            match self.db.enqueue_op(&op) {
+                Ok((_, superseded)) => {
+                    if let Some(old) = superseded {
+                        self.cache.discard_staged(Path::new(&old));
+                    }
+                    self.pending.lock().insert(
+                        uid.clone(),
+                        PendingRevision {
+                            path: blob.clone(),
+                            meta,
+                        },
+                    );
+                    queued += 1;
+                }
+                Err(e) => {
+                    stranded += 1;
+                    error!(%uid, blob = %blob.display(), error = %e,
+                           "cannot queue an orphaned staged write; bytes kept");
+                }
+            }
+        }
+        if queued > 0 || stranded > 0 {
+            info!(queued, stranded, "reconciled orphaned staged writes");
+        }
+    }
+
     /// Queue the writes that an unclean shutdown caught between `fsync(2)` and
     /// `close(2)`.
     ///

crates/pdfs-fuse/src/mount.rs

@@ -357,6 +357,7 @@ pub fn mount(
     let core = Core {
         client: client.clone(),
         rt: rt.clone(),
+        maintenance: Arc::new(Mutex::new(Default::default())),
         primary_root_uid: root.uid.clone(),
         primary: true,
         state: Arc::new(Mutex::new(State {
@@ -416,6 +417,9 @@ pub fn mount(
     // aside at open. After `hydrate_pending`, so a recovered partial write can
     // see an already-queued write to the same node.
     core.recover_fsynced_writes();
+    // And finally the staged blobs neither of those accounts for: bytes a
+    // failed release left behind with nothing pointing at them.
+    core.reconcile_staging();
     {
         let core = core.clone();
         std::thread::Builder::new()