From 1d5fdfb6b99dd32c193467413201fa95eccc972b Mon Sep 17 00:00:00 2001 From: Nils Pukropp Date: Sun, 19 Jul 2026 12:48:48 +0200 Subject: [PATCH] some performance optimizations --- crates/pdfs-core/src/cache.rs | 271 ++++++++++++++++++++++---- crates/pdfs-core/src/db/cache.rs | 44 ++++- crates/pdfs-core/src/db/migrations.rs | 15 +- crates/pdfs-core/src/db/ops.rs | 57 ++++++ crates/pdfs-core/src/db/tests.rs | 254 ++++++++++++++++++++++++ crates/pdfs-fuse/src/drain.rs | 12 +- crates/pdfs-fuse/src/lib.rs | 33 ---- 7 files changed, 601 insertions(+), 85 deletions(-) diff --git a/crates/pdfs-core/src/cache.rs b/crates/pdfs-core/src/cache.rs index 5c2c20b..2930919 100644 --- a/crates/pdfs-core/src/cache.rs +++ b/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>) { 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(); diff --git a/crates/pdfs-core/src/db/cache.rs b/crates/pdfs-core/src/db/cache.rs index 2db9b56..817cdd8 100644 --- a/crates/pdfs-core/src/db/cache.rs +++ b/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 { + 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> { + 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> { let conn = self.conn.lock(); let mut stmt = conn.prepare( diff --git a/crates/pdfs-core/src/db/migrations.rs b/crates/pdfs-core/src/db/migrations.rs index 290118f..96b8ff8 100644 --- a/crates/pdfs-core/src/db/migrations.rs +++ b/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); +"; diff --git a/crates/pdfs-core/src/db/ops.rs b/crates/pdfs-core/src/db/ops.rs index 990ceaa..81498ae 100644 --- a/crates/pdfs-core/src/db/ops.rs +++ b/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> { + 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> { let conn = self.conn.lock(); let mut stmt = conn.prepare( diff --git a/crates/pdfs-core/src/db/tests.rs b/crates/pdfs-core/src/db/tests.rs index 7447a58..02b1b36 100644 --- a/crates/pdfs-core/src/db/tests.rs +++ b/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` 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 + ); +} diff --git a/crates/pdfs-fuse/src/drain.rs b/crates/pdfs-fuse/src/drain.rs index 6768657..33d2829 100644 --- a/crates/pdfs-fuse/src/drain.rs +++ b/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(); diff --git a/crates/pdfs-fuse/src/lib.rs b/crates/pdfs-fuse/src/lib.rs index 0ed7224..c717ae3 100644 --- a/crates/pdfs-fuse/src/lib.rs +++ b/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"));