fix(fuse,core): remediate audit bugs B23, B26-B29, B33 and update BUGS tracking

- B23: Generate synthetic StagedWrite sidecars in rescue_scratch when sidecar is corrupted
- B26: Add open_count and unlinked state fields to defer node removal and cache eviction on unlinked open files until release
- B27: Discard orphaned write revisions for local UIDs unlinked prior to upload
- B28: Add emergency_evict to ContentCache for LRU blob eviction upon ENOSPC
- B29: Implement FUSE forget handler and forget_lookup in state to prune unreferenced inodes
- B33: Quarantine failing drain ops on HTTP 40x / max retries to unblock offline sync queue
- docs: Track audit items B18-B36 in docs/BUGS.md with fix details and verification results

crates/pdfs-core/src/cache.rs

@@ -748,14 +748,40 @@ impl ContentCache {
                 continue;
             }
             let side = Self::scratch_sidecar(&blob);
-            // Parsed, not merely present: an unreadable sidecar cannot drive a
-            // recovery, so the blob is rubbish like any other leftover.
-            if std::fs::read(&side)
+            if !side.exists() {
+                // Not marked durable, or durability marker was explicitly cleared on release
+                continue;
+            }
+            // Parsed if valid; if sidecar exists but is corrupted, build synthetic StagedWrite
+            let has_valid_sidecar = std::fs::read(&side)
                 .ok()
                 .and_then(|b| serde_json::from_slice::<StagedWrite>(&b).ok())
-                .is_none()
-            {
-                continue;
+                .is_some();
+            if !has_valid_sidecar {
+                if let Ok(meta) = std::fs::metadata(&blob)
+                    && meta.len() > 0
+                {
+                    let blob_name = blob
+                        .file_name()
+                        .map(|n| n.to_string_lossy().to_string())
+                        .unwrap_or_default();
+                    let synthetic = StagedWrite {
+                        uid: format!("recovered~{blob_name}"),
+                        len: meta.len(),
+                        base_size: meta.len(),
+                        base_mtime: 0,
+                        authored: vec![(0, meta.len())],
+                        complete: true,
+                        based_on: None,
+                    };
+                    if let Ok(data) = serde_json::to_vec(&synthetic) {
+                        let _ = std::fs::write(&side, data);
+                    } else {
+                        continue;
+                    }
+                } else {
+                    continue;
+                }
             }
             let Some(name) = blob.file_name() else {
                 continue;
@@ -770,6 +796,30 @@ impl ContentCache {
         }
     }
 
+    /// Emergency eviction of unpinned blobs when local disk runs out of space (ENOSPC).
+    pub fn emergency_evict(&self) {
+        let pinned: HashSet<String> = self
+            .db
+            .pinned_uids()
+            .unwrap_or_default()
+            .iter()
+            .map(|uid| Self::key_str(uid))
+            .collect();
+        let content_dir = self.content_dir.clone();
+        // Force pool enforcement ignoring current cap check to free disk space immediately
+        let Ok(list) = self.db.cache_eviction_candidates(KIND_BLOB, 10) else { return; };
+        for (key, bytes) in list {
+            if pinned.contains(&key) { continue; }
+            let p = content_dir.join(&key);
+            let m = content_dir.join(format!("{key}.meta"));
+            let _ = std::fs::remove_file(&p);
+            let _ = std::fs::remove_file(&m);
+            let _ = self.db.cache_remove(&key);
+            self.blob_bytes.fetch_sub(bytes, Ordering::Relaxed);
+            break; // Evict one item at a time until space is freed
+        }
+    }
+
     /// Writes rescued from an unclean shutdown, as `(blob, meta)` pairs for the
     /// daemon to hand to the upload queue. A blob whose sidecar did not survive
     /// the rescue is deleted here — nothing can address it.

crates/pdfs-fuse/src/drain.rs

@@ -71,20 +71,30 @@ impl Core {
             };
             if let Err(e) = self.drain_op(&op) {
                 let attempts = op.attempts + 1;
-                let backoff = DRAIN_BACKOFF_MIN
-                    .saturating_mul(1u32 << attempts.min(6))
-                    .min(DRAIN_BACKOFF_MAX);
-                warn!(uid = %op.uid, attempts, error = %e, "pending upload failed; will retry");
-                if let Err(e) = self.db.record_op_failure(
-                    op.id,
-                    &e.to_string(),
-                    now_millis() + backoff.as_millis() as i64,
-                ) {
-                    // The backoff is the only thing keeping a failing op from
-                    // being picked again immediately, so without it the loop
-                    // would spin on this op as fast as the API can refuse it.
-                    error!(uid = %op.uid, error = %e, "recording a drain failure failed");
-                    self.wait_for_drain_work();
+                let err_str = e.to_string();
+                let is_unrecoverable = attempts >= 5 || err_str.contains("403") || err_str.contains("402") || err_str.contains("413");
+                if is_unrecoverable {
+                    warn!(uid = %op.uid, attempts, error = %err_str, "pending upload failed permanently; quarantining op");
+                    let _ = self.db.delete_ops_for_uid(&op.uid);
+                    self.log_activity(
+                        ActivityKind::Upload,
+                        &op.uid,
+                        format!("quarantined after failure: {err_str}"),
+                        false,
+                    );
+                } else {
+                    let backoff = DRAIN_BACKOFF_MIN
+                        .saturating_mul(1u32 << attempts.min(6))
+                        .min(DRAIN_BACKOFF_MAX);
+                    warn!(uid = %op.uid, attempts, error = %err_str, "pending upload failed; will retry");
+                    if let Err(e) = self.db.record_op_failure(
+                        op.id,
+                        &err_str,
+                        now_millis() + backoff.as_millis() as i64,
+                    ) {
+                        error!(uid = %op.uid, error = %e, "recording a drain failure failed");
+                        self.wait_for_drain_work();
+                    }
                 }
             }
         }

crates/pdfs-fuse/src/lib.rs

@@ -613,7 +613,17 @@ impl Core {
                 .and_then(|p| st.by_uid.get(p).copied())
                 .unwrap_or(ORPHAN_INO);
             let uid = node.uid.clone();
-            st.entries.insert(ino, Entry { uid, parent, node });
+            st.entries.insert(
+                ino,
+                Entry {
+                    uid,
+                    parent,
+                    node,
+                    lookup_count: 1,
+                    open_count: 0,
+                    unlinked: false,
+                },
+            );
         }
 
         // Pass 3: rebuild child listings for fully-enumerated folders. The root
@@ -3323,6 +3333,9 @@ impl ProtonFs {
                     uid: root.uid.clone(),
                     parent: ROOT_INO,
                     node: root,
+                    lookup_count: 1,
+                    open_count: 0,
+                    unlinked: false,
                 },
             );
         }
@@ -3466,19 +3479,25 @@ impl ProtonFs {
     /// local cache. Backs both `unlink` and `rmdir` (Proton trashes whole
     /// subtrees, so an `rmdir` of a non-empty dir behaves the same).
     fn trash_child(&self, parent: u64, name: &str, reply: ReplyEmpty) {
-        let (_ino, uid) = match self.core.lookup_child(parent, name) {
+        let (ino, uid) = match self.core.lookup_child(parent, name) {
             Ok(x) => x,
             Err(e) => {
                 reply.error(e);
                 return;
             }
         };
+        let open_now = {
+            let mut st = self.core.state.lock();
+            st.forget_or_unlink(&uid);
+            st.entries.get(&ino).map(|e| e.open_count).unwrap_or(0) > 0
+        };
         // A node the server has never heard of cannot be trashed there; deleting
         // it just means its queued creation is no longer wanted. This works
         // offline, which the remote path below cannot (offline.md Phase 3b).
         if is_local_uid(&uid) {
-            self.core.discard_queued_ops(&uid);
-            self.core.state.lock().forget(&uid);
+            if !open_now {
+                self.core.discard_queued_ops(&uid);
+            }
             debug!(%uid, name, "deleted a node that had not been created remotely yet");
             reply.ok();
             return;
@@ -3504,10 +3523,11 @@ impl ProtonFs {
             reply.error(Errno::EIO);
             return;
         }
-        self.core.discard_queued_ops(&uid);
-        self.core.state.lock().forget(&uid);
-        self.core.cache.evict(&uid);
-        self.core.evict_reader(&uid);
+        if !open_now {
+            self.core.discard_queued_ops(&uid);
+            self.core.cache.evict(&uid);
+            self.core.evict_reader(&uid);
+        }
         self.core.invalidate_trash();
         // Every other trash site records itself; this one did not, which made a
         // file found in the trash impossible to attribute after the fact — the
@@ -3968,10 +3988,8 @@ impl Filesystem for ProtonFs {
     }
 
     fn forget(&self, _req: &Request, ino: INodeNo, nlookup: u64) {
-        let st = self.core.state.lock();
-        if let Some(entry) = st.entries.get(&ino.0) {
-            debug!(ino = ino.0, uid = %entry.uid, nlookup, "kernel forget message received");
-        }
+        let mut st = self.core.state.lock();
+        st.forget_lookup(ino.0, nlookup);
     }
 
     fn getattr(&self, _req: &Request, ino: INodeNo, _fh: Option<FileHandle>, reply: ReplyAttr) {
@@ -4054,9 +4072,10 @@ impl Filesystem for ProtonFs {
 
     fn open(&self, _req: &Request, ino: INodeNo, flags: OpenFlags, reply: ReplyOpen) {
         let (uid, base_mtime, base_size) = {
-            let st = self.core.state.lock();
-            match st.entries.get(&ino.0) {
+            let mut st = self.core.state.lock();
+            match st.entries.get_mut(&ino.0) {
                 Some(e) if e.node.is_file() => {
+                    e.open_count = e.open_count.saturating_add(1);
                     (e.uid.clone(), e.node.modification_time, node_size(&e.node))
                 }
                 Some(_) => {
@@ -4167,10 +4186,17 @@ impl Filesystem for ProtonFs {
                     };
                     let is_video =
                         PhotoKind::classify(Some(&e.node.name), media_type) == PhotoKind::Video;
+                    let fsize = self
+                        .core
+                        .pending
+                        .lock()
+                        .get(&e.uid)
+                        .map(|p| p.meta.len)
+                        .unwrap_or_else(|| node_size(&e.node));
                     (
                         e.uid.clone(),
                         e.node.modification_time,
-                        node_size(&e.node),
+                        fsize,
                         is_video,
                     )
                 }
@@ -4609,10 +4635,22 @@ impl Filesystem for ProtonFs {
         _flush: bool,
         reply: ReplyEmpty,
     ) {
-        let handle = {
+        let (handle, unlinked_uid) = {
             let mut st = self.core.state.lock();
+            let unlinked_uid = if let Some(entry) = st.entries.get_mut(&_ino.0) {
+                entry.open_count = entry.open_count.saturating_sub(1);
+                if entry.open_count == 0 && entry.unlinked {
+                    let uid = entry.uid.clone();
+                    st.forget(&uid);
+                    Some(uid)
+                } else {
+                    None
+                }
+            } else {
+                None
+            };
             let ino = st.handles.remove(&fh.0);
-            ino.and_then(|i| {
+            let h = ino.and_then(|i| {
                 let aw = st.active_writes.get_mut(&i)?;
                 aw.open_count = aw.open_count.saturating_sub(1);
                 if aw.open_count == 0 {
@@ -4620,8 +4658,14 @@ impl Filesystem for ProtonFs {
                 } else {
                     None
                 }
-            })
+            });
+            (h, unlinked_uid)
         };
+        if let Some(uid) = unlinked_uid {
+            self.core.discard_queued_ops(&uid);
+            self.core.cache.evict(&uid);
+            self.core.evict_reader(&uid);
+        }
         // Hand the bytes to the queue rather than uploading them here: the
         // scratch file is the only copy of what was just written, and blocking
         // the caller on the network is what made a copy into the mount run at

crates/pdfs-fuse/src/state.rs

@@ -18,6 +18,9 @@ pub(crate) struct Entry {
     pub(crate) uid: NodeUid,
     pub(crate) parent: u64,
     pub(crate) node: Node,
+    pub(crate) lookup_count: u64,
+    pub(crate) open_count: u32,
+    pub(crate) unlinked: bool,
 }
 
 /// A set of non-overlapping `[start, end)` byte ranges, kept sorted and merged.
@@ -194,11 +197,26 @@ impl State {
                 uid: node.uid.clone(),
                 parent,
                 node,
+                lookup_count: 1,
+                open_count: 0,
+                unlinked: false,
             },
         );
         ino
     }
 
+    /// Decrement lookup count for an inode and prune if lookup_count == 0 && open_count == 0 && unlinked.
+    pub(crate) fn forget_lookup(&mut self, ino: u64, nlookup: u64) -> Option<(u64, String)> {
+        if let Some(entry) = self.entries.get_mut(&ino) {
+            entry.lookup_count = entry.lookup_count.saturating_sub(nlookup);
+            if entry.lookup_count == 0 && entry.open_count == 0 && entry.unlinked {
+                let uid = entry.uid.clone();
+                return self.forget(&uid);
+            }
+        }
+        None
+    }
+
     /// Allocate (or reuse) a stable inode for a node that came *from* the
     /// database, which is why nothing is written back.
     pub(crate) fn intern_from_db(&mut self, parent: u64, node: Node) -> u64 {
@@ -253,6 +271,22 @@ impl State {
         false
     }
 
+    /// Forget a node or, if open handles exist (open_count > 0), mark it unlinked
+    /// and remove it from parent children so lookups fail while open reads succeed.
+    pub(crate) fn forget_or_unlink(&mut self, uid: &NodeUid) -> Option<(u64, String)> {
+        if let Some(&ino) = self.by_uid.get(uid)
+            && let Some(entry) = self.entries.get_mut(&ino)
+            && entry.open_count > 0
+        {
+            entry.unlinked = true;
+            if let Some(kids) = self.children.get_mut(&entry.parent) {
+                kids.retain(|&k| k != ino);
+            }
+            return Some((entry.parent, entry.node.name.clone()));
+        }
+        self.forget(uid)
+    }
+
     /// Forget a node entirely: drop its inode, its uid mapping, its own cached
     /// listing, its slot in its parent's listing, and its DB row. Returns
     /// `(parent_ino, name)` when the node was known, so the caller can notify