some performance optimizations

crates/pdfs-core/src/cache.rs

@@ -51,6 +51,12 @@ fn now_millis() -> i64 {
 /// `download_range` fetch with no straddling.
 pub const BLOCK_SIZE: u64 = 1 << 22;
 
+/// How many eviction candidates to pull per batch when a pool is over budget.
+/// Large enough that a normal overshoot (a store or two past the cap) is settled
+/// in one query, small enough that a badly-over-budget cache does not read its
+/// whole index to drop the oldest few.
+const EVICT_BATCH: usize = 64;
+
 /// Validity tag stored alongside a cached blob. A blob is fresh only if both
 /// fields still match the node's current metadata.
 #[derive(Serialize, Deserialize, Clone, Copy, PartialEq, Eq)]
@@ -161,6 +167,19 @@ pub struct ContentCache {
     /// (a Settings-page change) via [`set_budget`](Self::set_budget) without
     /// taking a lock on every cache read.
     max_bytes: AtomicU64,
+    /// Running byte totals for the two budgeted pools, so the overwhelmingly
+    /// common "still under budget" answer costs an atomic load instead of a
+    /// database query.
+    ///
+    /// These are a *fast path*, not the source of truth — the `cache_entries`
+    /// index is. They are seeded from it at open (right after `reconcile`, so
+    /// they describe what is actually on disk), advanced on each store, and
+    /// **re-seeded from the index after every eviction pass**, which is what
+    /// keeps a long-running daemon from drifting: the only operations that
+    /// change a total without going through here are external file deletions,
+    /// and the next open reconciles those.
+    blob_bytes: AtomicU64,
+    block_bytes: AtomicU64,
     /// Unified metadata DB. Its `cache_entries` table is the LRU index: every
     /// store/read/evict updates it, and the budget enforcers query it instead of
     /// scanning the cache directories (plan.md P4).
@@ -203,9 +222,14 @@ impl ContentCache {
             staging_dir,
             pins_path,
             max_bytes: AtomicU64::new(max_bytes),
+            blob_bytes: AtomicU64::new(0),
+            block_bytes: AtomicU64::new(0),
             db,
         };
         cache.reconcile()?;
+        // After reconcile, so the totals describe the index that now describes
+        // the disk.
+        cache.reseed_totals();
         cache.import_legacy_pins()?;
         Ok(cache)
     }
@@ -359,6 +383,9 @@ impl ContentCache {
     /// crash mid-store fails validation rather than serving truncated data.
     pub fn store(&self, uid: &NodeUid, mtime: i64, size: u64, bytes: &[u8]) -> Result<()> {
         let blob = self.blob_path(uid);
+        // `with_extension` is safe *here* only because a blob path is a bare
+        // 64-char hex key with no `.`, so this appends rather than replaces.
+        // `store_thumbnail` cannot use the same idiom — see the note there.
         let tmp = blob.with_extension("tmp");
         std::fs::write(&tmp, bytes)?;
         std::fs::rename(&tmp, &blob)?;
@@ -368,6 +395,8 @@ impl ContentCache {
         // newest entry is weighed (and ordered most-recent) like any other.
         self.db
             .cache_touch(&Self::key(uid), KIND_BLOB, bytes.len() as u64, now_millis())?;
+        self.blob_bytes
+            .fetch_add(bytes.len() as u64, Ordering::Relaxed);
         self.enforce_budget();
         Ok(())
     }
@@ -435,38 +464,95 @@ impl ContentCache {
             bytes.len() as u64,
             now_millis(),
         )?;
+        self.block_bytes
+            .fetch_add(bytes.len() as u64, Ordering::Relaxed);
         self.enforce_block_budget();
         Ok(())
     }
 
+    /// Re-read both pool totals from the cache index, which is authoritative.
+    fn reseed_totals(&self) {
+        if let Ok(n) = self.db.cache_total_bytes(KIND_BLOB) {
+            self.blob_bytes.store(n, Ordering::Relaxed);
+        }
+        if let Ok(n) = self.db.cache_total_bytes(KIND_BLOCK) {
+            self.block_bytes.store(n, Ordering::Relaxed);
+        }
+    }
+
+    /// The running total for a pool.
+    fn pool(&self, kind: &str) -> &AtomicU64 {
+        if kind == KIND_BLOCK {
+            &self.block_bytes
+        } else {
+            &self.blob_bytes
+        }
+    }
+
     /// Evict least-recently-used block-cache files until the block dir fits
     /// `max_bytes`. No-op when the cap is disabled (`0`). All blocks are
     /// evictable — pinned files are served from whole-file blobs, never blocks.
     fn enforce_block_budget(&self) {
+        self.enforce_pool(KIND_BLOCK, &self.block_dir, None);
+    }
+
+    /// Evict least-recently-used entries of one pool until it fits the cap.
+    ///
+    /// Runs on every store, so the under-budget case is the one that matters:
+    /// it is an atomic load and nothing else. Only once the running total says
+    /// we are over does this touch the database, and then it works in bounded
+    /// batches rather than reading the whole index — a cache that is 10 GB over
+    /// budget should not materialize every row to drop the oldest few.
+    ///
+    /// `exempt` names keys that count toward the total but are never victims
+    /// (pinned blobs). A pass that can find no eligible victim stops rather than
+    /// looping: a cache held entirely by pins legitimately stays over budget.
+    fn enforce_pool(&self, kind: &str, dir: &Path, exempt: Option<&HashSet<String>>) {
         let cap = self.cap();
         if cap == 0 {
             return;
         }
-        // LRU-ordered (oldest first) block entries straight from the index — no
-        // directory scan. All blocks are evictable; pinned files are served from
-        // whole-file blobs, never blocks.
-        let Ok(entries) = self.db.cache_entries_by_kind(KIND_BLOCK) else {
-            return;
+        let total = self.pool(kind);
+        if total.load(Ordering::Relaxed) <= cap {
+            return; // the common case: one atomic load, no query
+        }
+
+        // Over budget by the running count — now do the accurate work.
+        let mut running = match self.db.cache_total_bytes(kind) {
+            Ok(n) => n,
+            Err(_) => return,
         };
-        let mut total: u64 = entries.iter().map(|(_, size)| *size).sum();
-        if total <= cap {
-            return;
-        }
-        for (key, size) in entries {
-            if total <= cap {
+        let mut skipped = 0usize;
+        while running > cap {
+            // Fetch a batch past whatever we have already skipped as exempt.
+            let want = skipped + EVICT_BATCH;
+            let Ok(batch) = self.db.cache_eviction_candidates(kind, want) else {
                 break;
+            };
+            if batch.len() <= skipped {
+                break; // nothing new to consider
+            }
+            let mut evicted_any = false;
+            for (key, size) in batch.into_iter().skip(skipped) {
+                if running <= cap {
+                    break;
+                }
+                if exempt.is_some_and(|set| set.contains(&key)) {
+                    skipped += 1; // counts toward the total, never a victim
+                    continue;
+                }
+                let _ = std::fs::remove_file(dir.join(&key));
+                let _ = std::fs::remove_file(dir.join(format!("{key}.meta")));
+                let _ = self.db.cache_remove(&key);
+                running = running.saturating_sub(size);
+                evicted_any = true;
+            }
+            if !evicted_any {
+                break; // every remaining candidate is exempt
             }
-            // `key` is the block file's name (and its `.meta` sibling's stem).
-            let _ = std::fs::remove_file(self.block_dir.join(&key));
-            let _ = std::fs::remove_file(self.block_dir.join(format!("{key}.meta")));
-            let _ = self.db.cache_remove(&key);
-            total = total.saturating_sub(size);
         }
+        // The index is the truth; realign the fast path with it.
+        self.reseed_totals();
     }
 
     /// Create a fresh, empty read-write scratch file for a disk-backed write
@@ -563,6 +649,8 @@ impl ContentCache {
         }
         // Forget the blob and all of this uid's block rows in the LRU index.
         let _ = self.db.cache_remove_all(&Self::key(uid));
+        // Rows just left the index; the running totals must follow.
+        self.reseed_totals();
     }
 
     fn thumb_blob(&self, uid: &NodeUid, ttype: i32) -> PathBuf {
@@ -622,6 +710,14 @@ impl ContentCache {
     /// Cache `bytes` as the `ttype` thumbnail for `uid`, tagged with `mtime`.
     /// Blob written to a temp file then renamed; the meta tag is written last so
     /// a crash mid-store fails validation rather than serving a torn thumbnail.
+    ///
+    /// The temp name carries `ttype`, and must. `Path::with_extension` would
+    /// *replace* the `.t{ttype}` suffix rather than append to it, giving every
+    /// thumbnail type of one node the same staging file — so a type-1 and a
+    /// type-2 store racing (the gallery caches type 1 from a control-socket
+    /// thread while `getxattr` caches type 2 on the FUSE dispatch loop) would
+    /// publish one type's bytes under the other's name. See
+    /// `concurrent_thumbnail_types_do_not_share_a_temp_file`.
     pub fn store_thumbnail(
         &self,
         uid: &NodeUid,
@@ -630,7 +726,9 @@ impl ContentCache {
         bytes: &[u8],
     ) -> Result<()> {
         let blob = self.thumb_blob(uid, ttype);
-        let tmp = blob.with_extension("tmp");
+        let tmp = self
+            .thumb_dir
+            .join(format!("{}.t{ttype}.tmp", Self::key(uid)));
         std::fs::write(&tmp, bytes)?;
         std::fs::rename(&tmp, &blob)?;
         std::fs::write(self.thumb_meta(uid, ttype), serde_json::to_vec(&mtime)?)?;
@@ -643,7 +741,10 @@ impl ContentCache {
     /// pins can legitimately stay over budget.
     fn enforce_budget(&self) {
         let cap = self.cap();
-        if cap == 0 {
+        if cap == 0 || self.blob_bytes.load(Ordering::Relaxed) <= cap {
+            // The common case, settled without a query — and in particular
+            // without the recursive pin CTE below, which used to run on every
+            // single store.
             return;
         }
         // Pinned blobs (by cache key) are exempt from eviction. Resolves direct
@@ -656,30 +757,8 @@ impl ContentCache {
             .iter()
             .map(|uid| Self::key_str(uid))
             .collect();
-
-        // LRU-ordered (oldest first) blob entries from the index — no directory
-        // scan. Pinned blobs still count toward the total (so pins alone can hold
-        // the cache over budget) but are never chosen as victims.
-        let Ok(entries) = self.db.cache_entries_by_kind(KIND_BLOB) else {
-            return;
-        };
-        let mut total: u64 = entries.iter().map(|(_, size)| *size).sum();
-        if total <= cap {
-            return;
-        }
-        for (key, size) in entries {
-            if total <= cap {
-                break;
-            }
-            if pinned.contains(&key) {
-                continue; // counts toward total but is never a victim
-            }
-            // `key` is the blob file's name (and its `.meta` sibling's stem).
-            let _ = std::fs::remove_file(self.content_dir.join(&key));
-            let _ = std::fs::remove_file(self.content_dir.join(format!("{key}.meta")));
-            let _ = self.db.cache_remove(&key);
-            total = total.saturating_sub(size);
-        }
+        let content_dir = self.content_dir.clone();
+        self.enforce_pool(KIND_BLOB, &content_dir, Some(&pinned));
     }
 
     /// Whether `uid` is pinned — directly, or because an ancestor folder is
@@ -804,6 +883,7 @@ impl ContentCache {
                 freed = freed.saturating_add(size);
             }
         }
+        self.reseed_totals();
         freed
     }
 }
@@ -963,6 +1043,62 @@ mod tests {
         assert!(!c.is_cached(&u, 100, 3));
     }
 
+    /// The under-budget path — which is every store until the cache fills — must
+    /// not touch the database at all. It is on the cold-read hot path: one call
+    /// per cached 4 MiB block, under the connection lock every FUSE metadata
+    /// call also needs.
+    #[test]
+    fn budget_check_is_free_when_under_budget() {
+        use std::time::Instant;
+
+        // Cap far above what we store, so every store stays under budget.
+        let (c, _d) = cache_capped(1 << 30);
+        let payload = vec![0u8; 4096];
+        // Populate enough entries that a full scan would be visibly slower.
+        for i in 0..2000u64 {
+            c.store_block(&uid("a"), 1, 1 << 30, i, &payload).unwrap();
+        }
+
+        let t = Instant::now();
+        for i in 2000..2500u64 {
+            c.store_block(&uid("a"), 1, 1 << 30, i, &payload).unwrap();
+        }
+        let per_store = t.elapsed() / 500;
+
+        println!("B4: store_block while under budget — {per_store:?} each");
+        // Generous bound: the point is that it does not grow with the 2000
+        // existing entries, not that any particular microsecond count holds.
+        assert!(
+            per_store < std::time::Duration::from_millis(1),
+            "under-budget store should not scan the index; took {per_store:?}"
+        );
+    }
+
+    /// The running totals are a fast path, not a second source of truth: after
+    /// an eviction pass they are realigned with the index, so repeated
+    /// over-budget stores cannot drift them.
+    #[test]
+    fn running_totals_stay_aligned_with_the_index() {
+        let (c, _d) = cache_capped(8);
+        for i in 0..12u64 {
+            c.store_block(&uid("a"), 1, 100, i, b"aaaa").unwrap();
+            std::thread::sleep(std::time::Duration::from_millis(2));
+        }
+        let indexed: u64 = c
+            .db
+            .cache_entries_by_kind("block")
+            .unwrap()
+            .iter()
+            .map(|(_, s)| s)
+            .sum();
+        assert_eq!(
+            c.block_bytes.load(Ordering::Relaxed),
+            indexed,
+            "running total matches the index after repeated eviction"
+        );
+        assert!(indexed <= 8, "and the cap is actually held: {indexed}");
+    }
+
     #[test]
     fn budget_evicts_lru_unpinned() {
         // Cap fits two 4-byte blobs but not three.
@@ -1039,6 +1175,57 @@ mod tests {
         assert_eq!(c.budget(), 8);
     }
 
+    /// **The A2 reproduce.** Thumbnail types 1 and 2 of one node must not share
+    /// a staging file. They are stored from different subsystems — the Photos
+    /// gallery caches type 1 from a control-socket thread while `getxattr`
+    /// caches type 2 on the FUSE dispatch loop — so the two genuinely overlap.
+    ///
+    /// The failure is silent: the loser's rename either publishes the *other*
+    /// type's bytes under this type's name, or vanishes with ENOENT. Both are
+    /// reported here, because a fix that only removed the error would leave the
+    /// corruption.
+    #[test]
+    fn concurrent_thumbnail_types_do_not_share_a_temp_file() {
+        use std::sync::Arc as StdArc;
+
+        let (c, _d) = cache();
+        let c = StdArc::new(c);
+        let u = uid("a");
+        // Big enough that the write is not a single atomic-looking syscall, and
+        // distinguishable by content and length.
+        let one = vec![0xA1u8; 96 * 1024];
+        let two = vec![0xB2u8; 64 * 1024];
+
+        let mut handles = Vec::new();
+        for (ttype, payload) in [(1i32, one.clone()), (2i32, two.clone())] {
+            let c = c.clone();
+            let u = u.clone();
+            handles.push(std::thread::spawn(move || {
+                let mut errors = 0usize;
+                for _ in 0..400 {
+                    if c.store_thumbnail(&u, ttype, 100, &payload).is_err() {
+                        errors += 1;
+                    }
+                }
+                errors
+            }));
+        }
+
+        let mut store_errors = 0usize;
+        for h in handles {
+            store_errors += h.join().expect("store thread panicked");
+        }
+
+        // Whatever is on disk for each type must be that type's bytes.
+        if let Some(got) = c.read_thumbnail(&u, 1, 100) {
+            assert_eq!(got, one, "type 1 served type 2's bytes");
+        }
+        if let Some(got) = c.read_thumbnail(&u, 2, 100) {
+            assert_eq!(got, two, "type 2 served type 1's bytes");
+        }
+        assert_eq!(store_errors, 0, "a store lost its staging file to the other type");
+    }
+
     #[test]
     fn thumbnail_store_read_and_evict() {
         let (c, _d) = cache();

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

@@ -64,9 +64,49 @@ impl Db {
         Ok(())
     }
 
+    /// Total bytes held by entries of `kind`.
+    ///
+    /// The budget enforcers' first question, and almost always their last: the
+    /// common case is "under budget, evict nothing". Answering it with a `SUM`
+    /// costs one index range rather than materializing every row, which matters
+    /// because `enforce_block_budget` runs on every cached block — once per
+    /// 4 MiB of every cold read — while holding the shared connection.
+    pub fn cache_total_bytes(&self, kind: &str) -> Result<u64> {
+        let conn = self.conn.lock();
+        let total: i64 = conn.query_row(
+            "SELECT COALESCE(SUM(size_bytes), 0) FROM cache_entries WHERE kind = ?1",
+            params![kind],
+            |r| r.get(0),
+        )?;
+        Ok(total.max(0) as u64)
+    }
+
+    /// The `limit` least-recently-accessed entries of `kind`, oldest first.
+    ///
+    /// Eviction wants a handful of victims, not the whole table. Bounded so a
+    /// pass over a large cache does not read every row to drop a few, and so the
+    /// `(kind, last_accessed)` index can answer it as a range scan with an early
+    /// stop.
+    pub fn cache_eviction_candidates(&self, kind: &str, limit: usize) -> Result<Vec<(String, u64)>> {
+        let conn = self.conn.lock();
+        let mut stmt = conn.prepare_cached(
+            "SELECT cache_key, size_bytes FROM cache_entries
+             WHERE kind = ?1 ORDER BY last_accessed ASC LIMIT ?2",
+        )?;
+        let rows = stmt.query_map(params![kind, limit as i64], |r| {
+            Ok((r.get::<_, String>(0)?, r.get::<_, i64>(1)?.max(0) as u64))
+        })?;
+        let mut out = Vec::new();
+        for r in rows {
+            out.push(r?);
+        }
+        Ok(out)
+    }
+
     /// Every cache entry of `kind`, ordered least-recently-accessed first, as
-    /// `(cache_key, size_bytes)`. The budget enforcer sums the sizes and evicts
-    /// from the front until the cache fits.
+    /// `(cache_key, size_bytes)`. Used by the whole-cache passes
+    /// (`clear_unpinned`); incremental eviction wants
+    /// [`cache_eviction_candidates`](Self::cache_eviction_candidates).
     pub fn cache_entries_by_kind(&self, kind: &str) -> Result<Vec<(String, u64)>> {
         let conn = self.conn.lock();
         let mut stmt = conn.prepare(

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 = 13;
+pub(super) const SCHEMA_VERSION: i64 = 14;
 
 impl Db {
     pub(super) fn migrate(&self) -> Result<()> {
@@ -75,6 +75,9 @@ impl Db {
         if current < 13 {
             tx.execute_batch(MIGRATION_V13)?;
         }
+        if current < 14 {
+            tx.execute_batch(MIGRATION_V14)?;
+        }
         tx.execute(
             "INSERT INTO sync_state (key, value) VALUES ('schema_version', ?1)
              ON CONFLICT(key) DO UPDATE SET value = excluded.value",
@@ -327,3 +330,13 @@ const MIGRATION_V13: &str = "
 ALTER TABLE photos ADD COLUMN media_type TEXT;
 ALTER TABLE photos ADD COLUMN kind INTEGER NOT NULL DEFAULT 0;
 ";
+
+/// Schema v14: index the content-cache LRU. `cache_entries` carried only its
+/// `cache_key` primary key, so every budget check and every eviction pass was a
+/// full scan plus a sort — on a path that runs once per cached 4 MiB block,
+/// under the connection lock every FUSE metadata call also needs. `(kind,
+/// last_accessed)` turns both into a range scan: the `SUM` reads one kind's
+/// slice, and eviction reads its front and stops.
+const MIGRATION_V14: &str = "
+CREATE INDEX cache_entries_lru ON cache_entries(kind, last_accessed);
+";

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

@@ -269,6 +269,63 @@ impl Db {
 
     /// Every queued op, oldest first. The drain worker replays them in this order
     /// so a file's writes land in the order they were made.
+    /// The oldest queued op that is due and not blocked, or `None`.
+    ///
+    /// This is the drain loop's per-iteration query, and it is deliberately not
+    /// `pending_ops().find(...)`. That variant read every row — including each
+    /// op's full `meta_json` — sorted them, and threw all but one away, once per
+    /// drained op *and* once per retry, while holding the single connection
+    /// mutex that every FUSE `lookup` also needs. A queue of a few thousand ops
+    /// (an ordinary offline `cp -r`) made draining quadratic in the queue length.
+    ///
+    /// Both filters are in SQL so the row is decided by the database rather than
+    /// materialized and rejected in Rust:
+    ///
+    /// * `next_attempt_at <= now` — the backoff.
+    /// * a `local~` parent is skipped. A node created inside a folder that was
+    ///   itself created offline cannot be sent anywhere until that folder is
+    ///   real. Ops replay in queue order so the parent normally drains first;
+    ///   this matters when the parent is backing off, where the child must wait
+    ///   rather than burn its own retries.
+    ///
+    /// `id` is `INTEGER PRIMARY KEY`, i.e. the rowid, so `ORDER BY id LIMIT 1`
+    /// walks the table in insertion order and stops at the first match — no
+    /// sort, and no separate index to maintain. The common case (something is
+    /// due) exits on the first row.
+    pub fn next_due_op(&self, now: i64) -> Result<Option<PendingOp>> {
+        let conn = self.conn.lock();
+        let mut stmt = conn.prepare_cached(&format!(
+            "SELECT id, kind, uid, parent_uid, name, blob_path, meta_json, created_at,
+                    attempts, last_error, next_attempt_at
+             FROM pending_op
+             WHERE next_attempt_at <= ?1
+               AND (parent_uid IS NULL OR substr(parent_uid, 1, {n}) <> '{v}~')
+             ORDER BY id LIMIT 1",
+            v = LOCAL_VOLUME,
+            n = LOCAL_VOLUME.len() + 1,
+        ))?;
+        let op = stmt
+            .query_row(params![now], |r| {
+                Ok(PendingOp {
+                    id: r.get(0)?,
+                    kind: r.get(1)?,
+                    uid: r.get(2)?,
+                    parent_uid: r.get(3)?,
+                    name: r.get(4)?,
+                    blob_path: r.get(5)?,
+                    meta_json: r.get(6)?,
+                    created_at: r.get(7)?,
+                    attempts: r.get(8)?,
+                    last_error: r.get(9)?,
+                    next_attempt_at: r.get(10)?,
+                })
+            })
+            .optional()?;
+        Ok(op)
+    }
+
+    /// Every queued op, oldest first. For status read-outs and the CLI — the
+    /// drain wants [`next_due_op`](Self::next_due_op) instead.
     pub fn pending_ops(&self) -> Result<Vec<PendingOp>> {
         let conn = self.conn.lock();
         let mut stmt = conn.prepare(

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

@@ -1061,3 +1061,257 @@ fn schema_objects_exist() {
         .unwrap();
     assert_eq!(count, 3);
 }
+
+/// A queued op carrying a realistically-sized `meta_json`, which is what made
+/// the old scan expensive: the drain read and parsed every op's sidecar to pick
+/// one.
+fn bulk_op(i: usize, parent: &str) -> PendingOp {
+    PendingOp {
+        id: 0,
+        kind: OP_REVISION.to_string(),
+        uid: format!("vol~link{i}"),
+        parent_uid: Some(parent.to_string()),
+        name: Some(format!("file-{i}.bin")),
+        blob_path: Some(format!("/staging/vol~link{i}-{i}")),
+        // A StagedWrite sidecar, roughly the shape and size of a real one.
+        meta_json: Some(format!(
+            r#"{{"uid":"vol~link{i}","len":8388608,"base_size":8388608,
+                 "base_mtime":1700000000,"authored":[[0,8388608]],"complete":true,
+                 "based_on":{{"mtime":1700000000,"size":8388608}}}}"#
+        )),
+        created_at: 1,
+        attempts: 0,
+        last_error: None,
+        next_attempt_at: 0,
+    }
+}
+
+/// **The B3 measurement.** The drain picks one op per iteration. Doing that by
+/// reading the whole queue is quadratic in queue length; doing it in SQL with
+/// `LIMIT 1` is flat. Asserts the shape of the difference, not a wall-clock
+/// number — the point is that one grows with the queue and the other does not.
+#[test]
+fn next_due_op_does_not_scale_with_queue_length() {
+    use std::time::Instant;
+
+    let db = Db::open_in_memory().unwrap();
+    let root = uid("root").to_string();
+    const N: usize = 2000;
+    for i in 0..N {
+        db.enqueue_op(&bulk_op(i, &root)).unwrap();
+    }
+
+    // Both must agree on which op is next.
+    let scanned = db
+        .pending_ops()
+        .unwrap()
+        .into_iter()
+        .find(|o| o.next_attempt_at <= 10);
+    let queried = db.next_due_op(10).unwrap();
+    assert_eq!(
+        scanned.as_ref().map(|o| o.id),
+        queried.as_ref().map(|o| o.id),
+        "the new query must pick the op the old scan picked"
+    );
+    assert_eq!(scanned.map(|o| o.uid), queried.map(|o| o.uid));
+
+    // Simulate a drain pass: pick the next op, retire it, repeat.
+    let rounds = 200;
+
+    let t0 = Instant::now();
+    for _ in 0..rounds {
+        let _ = db
+            .pending_ops()
+            .unwrap()
+            .into_iter()
+            .find(|o| o.next_attempt_at <= 10);
+    }
+    let scan = t0.elapsed();
+
+    let t1 = Instant::now();
+    for _ in 0..rounds {
+        let _ = db.next_due_op(10).unwrap();
+    }
+    let query = t1.elapsed();
+
+    println!("B3: {rounds} picks over a {N}-op queue — scan {scan:?}, query {query:?}");
+    assert!(
+        query * 20 < scan,
+        "expected the LIMIT 1 query to be far cheaper than a full scan; \
+         scan={scan:?} query={query:?}"
+    );
+}
+
+/// The readiness filter moved into SQL, so it needs its own coverage there: an
+/// op whose parent was itself created offline is not yet sendable and must be
+/// skipped in favour of a later one that is.
+#[test]
+fn next_due_op_skips_ops_blocked_on_a_local_parent() {
+    let db = Db::open_in_memory().unwrap();
+    let root = uid("root").to_string();
+
+    db.enqueue_op(&bulk_op(1, "local~dir")).unwrap();
+    db.enqueue_op(&bulk_op(2, &root)).unwrap();
+
+    let next = db.next_due_op(10).unwrap().expect("an op is due");
+    assert_eq!(next.uid, "vol~link2", "skipped the local-parent op");
+
+    // A NULL parent is not blocked.
+    let mut orphan = bulk_op(3, &root);
+    orphan.parent_uid = None;
+    db.enqueue_op(&orphan).unwrap();
+    db.delete_op(next.id).unwrap();
+    assert_eq!(db.next_due_op(10).unwrap().unwrap().uid, "vol~link3");
+}
+
+/// Backoff still gates: nothing is returned before an op is due.
+#[test]
+fn next_due_op_respects_backoff() {
+    let db = Db::open_in_memory().unwrap();
+    let root = uid("root").to_string();
+    let id = db.enqueue_op(&bulk_op(1, &root)).unwrap().0;
+    db.record_op_failure(id, "boom", 5_000).unwrap();
+
+    assert!(db.next_due_op(4_999).unwrap().is_none(), "still backing off");
+    assert!(db.next_due_op(5_000).unwrap().is_some(), "due now");
+}
+
+/// **The B4 measurement.** `enforce_block_budget` runs on *every* `store_block`,
+/// i.e. once per 4 MiB of every cold read. The old path read and sorted every
+/// row of the cache index to answer a question that is almost always "no, we are
+/// under budget" — under the shared connection lock, so it also stalled FUSE
+/// metadata calls.
+///
+/// Asserts the shape: the under-budget check must not scale with cache size.
+#[test]
+fn cache_total_bytes_does_not_scale_with_cache_size() {
+    use std::time::Instant;
+
+    let db = Db::open_in_memory().unwrap();
+    // A 20 GB block cache at 4 MiB a block is ~5000 rows.
+    const N: usize = 5000;
+    const BLOCK: u64 = 4 << 20;
+    for i in 0..N {
+        db.cache_touch(&format!("k{i}.b0"), "block", BLOCK, i as i64)
+            .unwrap();
+    }
+
+    // The aggregate must agree with summing the rows.
+    let summed: u64 = db
+        .cache_entries_by_kind("block")
+        .unwrap()
+        .iter()
+        .map(|(_, s)| s)
+        .sum();
+    assert_eq!(db.cache_total_bytes("block").unwrap(), summed);
+    assert_eq!(summed, N as u64 * BLOCK);
+    // Kinds are counted apart.
+    assert_eq!(db.cache_total_bytes("blob").unwrap(), 0);
+
+    let rounds = 500;
+
+    let t0 = Instant::now();
+    for _ in 0..rounds {
+        let entries = db.cache_entries_by_kind("block").unwrap();
+        let _: u64 = entries.iter().map(|(_, s)| *s).sum();
+    }
+    let scan = t0.elapsed();
+
+    let t1 = Instant::now();
+    for _ in 0..rounds {
+        let _ = db.cache_total_bytes("block").unwrap();
+    }
+    let aggregate = t1.elapsed();
+
+    println!("B4: {rounds} budget checks over {N} entries — scan {scan:?}, aggregate {aggregate:?}");
+    // The SUM is much cheaper — no row materialization, no allocation, no sort —
+    // but it is still O(rows): summing N values reads N index entries. That is
+    // why `ContentCache` keeps a running total on top of this and only falls
+    // back to the query when it has to evict. See
+    // `budget_check_is_free_when_under_budget`.
+    assert!(
+        aggregate * 2 < scan,
+        "expected the SUM to be materially cheaper than reading every row; \
+         scan={scan:?} aggregate={aggregate:?}"
+    );
+}
+
+/// Victims still come out least-recently-accessed first, now in bounded batches
+/// rather than one unbounded read of the table.
+#[test]
+fn cache_eviction_candidates_are_lru_ordered_and_limited() {
+    let db = Db::open_in_memory().unwrap();
+    for i in 0..10u64 {
+        // Insert newest-first so insertion order cannot be mistaken for LRU order.
+        db.cache_touch(&format!("k{i}"), "blob", 100, (10 - i) as i64)
+            .unwrap();
+    }
+    db.cache_touch("other", "block", 100, 0).unwrap();
+
+    let batch = db.cache_eviction_candidates("blob", 3).unwrap();
+    assert_eq!(batch.len(), 3, "honours the limit");
+    let keys: Vec<&str> = batch.iter().map(|(k, _)| k.as_str()).collect();
+    assert_eq!(keys, ["k9", "k8", "k7"], "least-recently-accessed first");
+    assert!(
+        batch.iter().all(|(k, _)| k != "other"),
+        "a different kind is never a candidate"
+    );
+}
+
+/// **The B5 checkpoint.** `improvements.md` P2.3 proposes concurrent SQLite
+/// reads, on the premise that the single `Mutex<Connection>` serializes the FUSE
+/// workers. This measures what the connection is actually asked to do now that
+/// B3 and B4 have landed, so the proposal is decided on evidence.
+///
+/// Note what is *not* here: `lookup`/`getattr`/`readdir` do not read this
+/// database in the steady state — they serve from `State::entries` in memory and
+/// only write through on a cold fill. The per-read DB operation is
+/// `cache_accessed`, one `UPDATE` per cache hit, which is what this drives.
+#[test]
+fn db_contention_under_fuse_worker_load() {
+    use std::sync::Arc;
+    use std::time::Instant;
+
+    let db = Arc::new(Db::open_in_memory().unwrap());
+    for i in 0..5000u64 {
+        db.cache_touch(&format!("k{i}.b0"), "block", 4 << 20, i as i64)
+            .unwrap();
+    }
+
+    // Eight workers, matching FUSE_WORKERS, each doing what a served block read
+    // does to the database: one LRU touch.
+    const WORKERS: usize = 8;
+    const PER_WORKER: usize = 2000;
+
+    let t = Instant::now();
+    let mut handles = Vec::new();
+    for w in 0..WORKERS {
+        let db = db.clone();
+        handles.push(std::thread::spawn(move || {
+            for i in 0..PER_WORKER {
+                db.cache_accessed(&format!("k{}.b0", (w * PER_WORKER + i) % 5000), i as i64)
+                    .unwrap();
+            }
+        }));
+    }
+    for h in handles {
+        h.join().unwrap();
+    }
+    let concurrent = t.elapsed();
+
+    // The same work on one thread, for the serialization baseline.
+    let t = Instant::now();
+    for i in 0..(WORKERS * PER_WORKER) {
+        db.cache_accessed(&format!("k{}.b0", i % 5000), i as i64)
+            .unwrap();
+    }
+    let serial = t.elapsed();
+
+    let total = WORKERS * PER_WORKER;
+    println!(
+        "B5: {total} LRU touches — {WORKERS} threads {concurrent:?}, 1 thread {serial:?} \
+         (per-op {:?} vs {:?})",
+        concurrent / total as u32,
+        serial / total as u32
+    );
+}

crates/pdfs-fuse/src/drain.rs

@@ -35,7 +35,7 @@ use super::transfers::CountingReader;
 use super::{
     Core, DRAIN_BACKOFF_MAX, DRAIN_BACKOFF_MIN, DRAIN_IDLE_POLL, ROOT_INO, conflict_name,
     is_already_exists, is_gone, is_local_uid_str, media_type_for, node_size, now_millis, now_secs,
-    op_is_ready, parse_node_uid,
+    parse_node_uid,
 };
 
 impl Core {
@@ -55,12 +55,10 @@ impl Core {
     pub(crate) fn run_pending_drain(&self) {
         loop {
             let now = now_millis();
-            let due = self
-                .db
-                .pending_ops()
-                .unwrap_or_default()
-                .into_iter()
-                .find(|op| op.next_attempt_at <= now && op_is_ready(op));
+            // One row, chosen by the database. Reading the whole queue to pick
+            // one op made a long queue quadratic to drain, and held the shared
+            // connection — and so every FUSE metadata call — for the duration.
+            let due = self.db.next_due_op(now).unwrap_or_default();
 
             let Some(op) = due.filter(|_| self.online.load(Ordering::Relaxed)) else {
                 self.wait_for_drain_work();

crates/pdfs-fuse/src/lib.rs

@@ -3097,17 +3097,6 @@ fn conflict_name(name: &str, stamp: i64) -> String {
     }
 }
 
-/// Whether an op can be attempted now, as opposed to waiting on another op.
-///
-/// A node created inside a folder that was itself created offline cannot be sent
-/// anywhere until that folder is real — the API has never heard of `local~…`.
-/// Ops replay in queue order, so the parent normally drains first and rewrites
-/// this one; the check matters when the parent is instead backing off after a
-/// failure, where the child must wait rather than burn its own retries.
-fn op_is_ready(op: &PendingOp) -> bool {
-    !op.parent_uid.as_deref().is_some_and(is_local_uid_str)
-}
-
 /// Fabricate the node the server would have returned, for a `create`/`mkdir`
 /// that could not reach it. Everything the kernel asks about a fresh node —
 /// name, kind, size, times — is knowable locally; the uid is the only invention,
@@ -4784,28 +4773,6 @@ mod local_uid_tests {
         assert!(!is_local_uid_str("vol1~local"));
     }
 
-    #[test]
-    fn an_op_waits_for_a_parent_that_is_still_a_placeholder() {
-        let op = |parent: Option<&str>| PendingOp {
-            id: 0,
-            kind: OP_CREATE.to_string(),
-            uid: "local~child".to_string(),
-            parent_uid: parent.map(str::to_string),
-            name: Some("f.txt".to_string()),
-            blob_path: None,
-            meta_json: None,
-            created_at: 0,
-            attempts: 0,
-            last_error: None,
-            next_attempt_at: 0,
-        };
-        // Sending this to the API would 404: the folder does not exist yet.
-        assert!(!op_is_ready(&op(Some("local~dir"))));
-        assert!(op_is_ready(&op(Some("vol1~dir"))));
-        // A revision op names no parent and is always ready.
-        assert!(op_is_ready(&op(None)));
-    }
-
     #[test]
     fn a_placeholder_file_reports_itself_as_empty_and_unsealed() {
         let parent = NodeUid::new(VolumeId::from("vol1"), LinkId::from("dir"));