diff --git a/Cargo.lock b/Cargo.lock index 4661970..ae31087 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3805,7 +3805,7 @@ dependencies = [ [[package]] name = "proton-drive-rs" -version = "0.1.8" +version = "0.1.9" dependencies = [ "async-trait", "base64", @@ -3830,7 +3830,7 @@ dependencies = [ [[package]] name = "proton-sdk" -version = "0.1.8" +version = "0.1.9" dependencies = [ "aes-gcm 0.11.0", "async-trait", @@ -3843,6 +3843,7 @@ dependencies = [ "hex", "hkdf 0.13.0", "hmac 0.13.0", + "lru", "num-bigint 0.5.1", "num-traits", "pgp", diff --git a/crates/pdfs-core/src/cache.rs b/crates/pdfs-core/src/cache.rs index 2930919..6d0765e 100644 --- a/crates/pdfs-core/src/cache.rs +++ b/crates/pdfs-core/src/cache.rs @@ -34,6 +34,27 @@ use crate::error::Result; const KIND_BLOB: &str = "blob"; /// `cache_entries.kind` tag for on-demand block-cache chunks. const KIND_BLOCK: &str = "block"; +/// `cache_entries.kind` tag for cached thumbnails. +const KIND_THUMB: &str = "thumb"; + +/// How the configured budget is divided between the three pools, in percent. +/// +/// The alternative — one shared cap across all kinds — was rejected: the pools +/// have genuinely different lifetimes, and sharing lets one starve another. A +/// single streaming read is millions of block bytes and would evict the whole +/// blob cache behind it; a gallery scroll would do the same with thumbnails. +/// +/// Whole-file blobs get the bulk: they are what "available offline" means, and +/// they are the only pool a user explicitly asks for (by pinning). Blocks are +/// transient partial reads, cheaply re-fetched. Thumbnails are tiny +/// individually — a few percent buys tens of thousands of them — but there is +/// one per photo, so the pool needs a ceiling, which is what C6 was about. +/// +/// Must sum to 100; asserted at compile time below. +const SPLIT_BLOB: u64 = 70; +const SPLIT_BLOCK: u64 = 25; +const SPLIT_THUMB: u64 = 5; +const _: () = assert!(SPLIT_BLOB + SPLIT_BLOCK + SPLIT_THUMB == 100); /// Current wall-clock time in unix *milliseconds*, for the LRU `last_accessed` /// column. Milliseconds (not seconds) so two cache events in the same second @@ -180,6 +201,7 @@ pub struct ContentCache { /// and the next open reconciles those. blob_bytes: AtomicU64, block_bytes: AtomicU64, + thumb_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). @@ -224,6 +246,7 @@ impl ContentCache { max_bytes: AtomicU64::new(max_bytes), blob_bytes: AtomicU64::new(0), block_bytes: AtomicU64::new(0), + thumb_bytes: AtomicU64::new(0), db, }; cache.reconcile()?; @@ -265,6 +288,7 @@ impl ContentCache { for (dir, kind) in [ (&self.content_dir, KIND_BLOB), (&self.block_dir, KIND_BLOCK), + (&self.thumb_dir, KIND_THUMB), ] { let Ok(rd) = std::fs::read_dir(dir) else { continue; @@ -272,7 +296,15 @@ impl ContentCache { for entry in rd.flatten() { let name = entry.file_name(); let Some(name) = name.to_str() else { continue }; - if name.ends_with(".meta") || name.ends_with(".tmp") { + if name.ends_with(".meta") { + continue; + } + // A `.tmp` is a staging file from a store that never completed + // its rename — a crashed or killed run. Nothing will ever claim + // it, and no pass other than this one looks in these + // directories, so it would sit there forever. Sweep it. + if name.ends_with(".tmp") { + let _ = std::fs::remove_file(entry.path()); continue; } let Ok(meta) = entry.metadata() else { continue }; @@ -470,25 +502,45 @@ impl ContentCache { Ok(()) } - /// Re-read both pool totals from the cache index, which is authoritative. + /// Re-read every pool total 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); + for kind in [KIND_BLOB, KIND_BLOCK, KIND_THUMB] { + if let Ok(n) = self.db.cache_total_bytes(kind) { + self.pool(kind).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 + match kind { + KIND_BLOCK => &self.block_bytes, + KIND_THUMB => &self.thumb_bytes, + _ => &self.blob_bytes, } } + /// This pool's share of the configured budget (`0` = unlimited, as for the + /// whole cache). See [`SPLIT_BLOB`] for why the budget is divided rather + /// than shared. + fn pool_cap(&self, kind: &str) -> u64 { + let cap = self.cap(); + if cap == 0 { + return 0; + } + let share = match kind { + KIND_BLOCK => SPLIT_BLOCK, + KIND_THUMB => SPLIT_THUMB, + _ => SPLIT_BLOB, + }; + // Multiply before dividing: `cap / 100 * share` truncates a small budget + // to zero, which `pool_cap`'s callers would read as "unlimited" — the + // opposite of what a small budget asks for. + // + // At least one byte, for the same reason. + (cap.saturating_mul(share) / 100).max(1) + } + /// 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. @@ -508,7 +560,7 @@ impl ContentCache { /// (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(); + let cap = self.pool_cap(kind); if cap == 0 { return; } @@ -653,8 +705,14 @@ impl ContentCache { self.reseed_totals(); } + /// Cache-index key for the `ttype` thumbnail of `uid` — the thumbnail file's + /// name, so eviction can remove the file by joining it onto `thumb_dir`. + fn thumb_key(&self, uid: &NodeUid, ttype: i32) -> String { + format!("{}.t{ttype}", Self::key(uid)) + } + fn thumb_blob(&self, uid: &NodeUid, ttype: i32) -> PathBuf { - self.thumb_dir.join(format!("{}.t{ttype}", Self::key(uid))) + self.thumb_dir.join(self.thumb_key(uid, ttype)) } fn thumb_meta(&self, uid: &NodeUid, ttype: i32) -> PathBuf { @@ -672,7 +730,14 @@ impl ContentCache { if recorded != mtime { return None; } - std::fs::read(self.thumb_blob(uid, ttype)).ok() + let bytes = std::fs::read(self.thumb_blob(uid, ttype)).ok()?; + // Record the access for LRU, as the blob and block readers do. Without + // it the pool evicts in insertion order, so scrolling back up a gallery + // would drop exactly the tiles being looked at. + let _ = self + .db + .cache_accessed(&self.thumb_key(uid, ttype), now_millis()); + Some(bytes) } /// On-disk path of the cached content blob for `uid`, validated against @@ -699,7 +764,13 @@ impl ContentCache { return None; } let blob = self.thumb_blob(uid, ttype); - blob.exists().then_some(blob) + if !blob.exists() { + return None; + } + let _ = self + .db + .cache_accessed(&self.thumb_key(uid, ttype), now_millis()); + Some(blob) } /// On-disk path where `uid`'s `ttype` thumbnail lives once stored. @@ -732,15 +803,33 @@ impl ContentCache { std::fs::write(&tmp, bytes)?; std::fs::rename(&tmp, &blob)?; std::fs::write(self.thumb_meta(uid, ttype), serde_json::to_vec(&mtime)?)?; + self.db.cache_touch( + &self.thumb_key(uid, ttype), + KIND_THUMB, + bytes.len() as u64, + now_millis(), + )?; + self.thumb_bytes + .fetch_add(bytes.len() as u64, Ordering::Relaxed); + self.enforce_thumb_budget(); Ok(()) } + /// Evict least-recently-used thumbnails until the pool fits its share of the + /// budget. Every thumbnail is evictable, including those of pinned nodes: a + /// thumbnail is derived data that the gallery re-fetches or regenerates, and + /// exempting them would mean a large pinned library could hold the pool + /// permanently over its ceiling — the exact unbounded growth C6 was about. + fn enforce_thumb_budget(&self) { + self.enforce_pool(KIND_THUMB, &self.thumb_dir, None); + } + /// Evict least-recently-used *unpinned* blobs until total blob bytes fit the /// configured `max_bytes` cap. No-op when the cap is disabled (`0`) or the /// cache already fits. Pinned blobs are skipped, so a cache held entirely by /// pins can legitimately stay over budget. fn enforce_budget(&self) { - let cap = self.cap(); + let cap = self.pool_cap(KIND_BLOB); 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 @@ -801,33 +890,31 @@ impl ContentCache { .collect() } - /// Total bytes of cached content blobs (pinned + unpinned), matching what - /// [`enforce_budget`](Self::enforce_budget) weighs against the cap. - /// Thumbnails live in a sibling dir and are not counted. Cheap directory - /// scan, safe to call from a front-end for a usage read-out. + /// Total bytes the cache holds on disk — blobs, blocks and thumbnails + /// together — which is what the configured budget caps and therefore the + /// only honest thing to show next to it. + /// + /// This used to scan `content_dir` alone, which excluded both the block dir + /// and the thumb dir: the Settings page could report a few hundred megabytes + /// while the cache held several gigabytes. Now it reads the same running + /// totals the budget enforcers use, so the displayed number and the enforced + /// number cannot disagree — and it costs three atomic loads instead of a + /// directory scan. + /// + /// Excludes the `.meta` sidecars (a few dozen bytes each) and the scratch + /// and staging dirs, which hold unuploaded user writes rather than cache and + /// are deliberately not evictable. pub fn usage(&self) -> u64 { - let Ok(rd) = std::fs::read_dir(&self.content_dir) else { - return 0; - }; - let mut total = 0u64; - for entry in rd.flatten() { - let Some(name) = entry.file_name().to_str().map(str::to_owned) else { - continue; - }; - if name.ends_with(".meta") || name.ends_with(".tmp") { - continue; - } - if let Ok(meta) = entry.metadata() - && meta.is_file() - { - total += meta.len(); - } - } - total + [KIND_BLOB, KIND_BLOCK, KIND_THUMB] + .iter() + .map(|k| self.pool(k).load(Ordering::Relaxed)) + .sum() } /// Configured soft byte cap (`0` = unlimited), for display alongside - /// [`usage`](Self::usage). + /// [`usage`](Self::usage). This is the whole-cache figure; internally it is + /// divided between the pools (see [`SPLIT_BLOB`]), but the user set one + /// number and gets that number. pub fn budget(&self) -> u64 { self.cap() } @@ -846,6 +933,7 @@ impl ContentCache { self.max_bytes.store(bytes, Ordering::Relaxed); self.enforce_budget(); self.enforce_block_budget(); + self.enforce_thumb_budget(); } /// Delete every *unpinned* cached blob plus all on-demand block chunks, @@ -874,13 +962,19 @@ impl ContentCache { freed = freed.saturating_add(size); } } - // Every on-demand block (transient partial reads, re-fetched on demand). - if let Ok(entries) = self.db.cache_entries_by_kind(KIND_BLOCK) { - for (key, size) in entries { - 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); - freed = freed.saturating_add(size); + // Every on-demand block (transient partial reads, re-fetched on demand) + // and every thumbnail (derived, re-fetched or regenerated on demand). + // Both pools are dropped wholesale rather than filtered by pin: neither + // is what "keep this available offline" means, and leaving them behind + // would make "clear cache" free visibly less than it reported. + for (kind, dir) in [(KIND_BLOCK, &self.block_dir), (KIND_THUMB, &self.thumb_dir)] { + if let Ok(entries) = self.db.cache_entries_by_kind(kind) { + for (key, size) in entries { + let _ = std::fs::remove_file(dir.join(&key)); + let _ = std::fs::remove_file(dir.join(format!("{key}.meta"))); + let _ = self.db.cache_remove(&key); + freed = freed.saturating_add(size); + } } } self.reseed_totals(); @@ -925,6 +1019,28 @@ mod tests { cache_capped(0) } + /// A cache whose `kind` pool holds exactly `bytes`. + /// + /// The configured budget is a whole-cache figure divided between the pools + /// (C7), while a test that exercises eviction cares about one pool at a + /// time. So the test states the pool size it means and this inverts the + /// split, rather than every such test hard-coding a number that silently + /// changes meaning if the split is ever retuned. + fn cache_with_pool_cap(kind: &str, bytes: u64) -> (ContentCache, TempDir) { + let share = match kind { + KIND_BLOCK => SPLIT_BLOCK, + KIND_THUMB => SPLIT_THUMB, + _ => SPLIT_BLOB, + }; + // Round up, so the pool holds at least `bytes` rather than one short. + let (c, d) = cache_capped((bytes * 100).div_ceil(share)); + assert!( + c.pool_cap(kind) >= bytes, + "helper must give the {kind} pool at least {bytes} bytes" + ); + (c, d) + } + fn cache_capped(max_bytes: u64) -> (ContentCache, TempDir) { let dir = TempDir::new(); let db = Arc::new(Db::open_in_memory().unwrap()); @@ -1054,23 +1170,33 @@ mod tests { // 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(); - } + // Cost of a store against a near-empty index... let t = Instant::now(); - for i in 2000..2500u64 { + for i in 0..500u64 { c.store_block(&uid("a"), 1, 1 << 30, i, &payload).unwrap(); } - let per_store = t.elapsed() / 500; + let empty = 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. + // ...and against one with a few thousand entries in it. + for i in 500..2500u64 { + c.store_block(&uid("a"), 1, 1 << 30, i, &payload).unwrap(); + } + let t = Instant::now(); + for i in 2500..3000u64 { + c.store_block(&uid("a"), 1, 1 << 30, i, &payload).unwrap(); + } + let full = t.elapsed() / 500; + + println!("B4: store_block under budget — {empty:?} at ~0 entries, {full:?} at 2500"); + // A ratio, not a wall-clock bound: the claim is that the under-budget + // path does not grow with the index, and only a ratio tests that. An + // absolute threshold measures the machine's load instead, and this test + // runs alongside the rest of the suite. assert!( - per_store < std::time::Duration::from_millis(1), - "under-budget store should not scan the index; took {per_store:?}" + full < empty * 3, + "under-budget store should not scale with the index: \ + {empty:?} at ~0 entries vs {full:?} at 2500" ); } @@ -1079,7 +1205,7 @@ mod tests { /// over-budget stores cannot drift them. #[test] fn running_totals_stay_aligned_with_the_index() { - let (c, _d) = cache_capped(8); + let (c, _d) = cache_with_pool_cap(KIND_BLOCK, 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)); @@ -1101,8 +1227,8 @@ mod tests { #[test] fn budget_evicts_lru_unpinned() { - // Cap fits two 4-byte blobs but not three. - let (c, _d) = cache_capped(8); + // Blob pool fits two 4-byte blobs but not three. + let (c, _d) = cache_with_pool_cap(KIND_BLOB, 8); let (a, b, d) = (uid("a"), uid("b"), uid("d")); c.store(&a, 1, 4, b"aaaa").unwrap(); @@ -1123,7 +1249,7 @@ mod tests { #[test] fn budget_never_evicts_pinned() { - let (c, _d) = cache_capped(8); + let (c, _d) = cache_with_pool_cap(KIND_BLOB, 8); let (a, b, d) = (uid("a"), uid("b"), uid("d")); c.store(&a, 1, 4, b"aaaa").unwrap(); @@ -1166,13 +1292,15 @@ mod tests { std::thread::sleep(std::time::Duration::from_millis(10)); c.store(&d, 1, 4, b"dddd").unwrap(); - // Tightening to 8 bytes evicts the LRU unpinned blob(s) now, not on the - // next store. - c.set_budget(8); + // Tightening to a budget whose blob share is 8 bytes evicts the LRU + // unpinned blob(s) now, not on the next store. + c.set_budget((8 * 100u64).div_ceil(SPLIT_BLOB)); assert!(!c.is_cached(&a, 1, 4), "oldest blob evicted on tighten"); assert!(c.is_cached(&b, 1, 4)); assert!(c.is_cached(&d, 1, 4)); - assert_eq!(c.budget(), 8); + // `budget()` reports the whole-cache figure that was set, not one + // pool's share of it. + assert_eq!(c.budget(), (8 * 100u64).div_ceil(SPLIT_BLOB)); } /// **The A2 reproduce.** Thumbnail types 1 and 2 of one node must not share @@ -1242,6 +1370,107 @@ mod tests { assert!(c.read_thumbnail(&u, 2, 100).is_none()); } + /// Total bytes of every real file under `root`, recursively — what `du` + /// would report, and what `usage()` claims to be reporting. + fn bytes_on_disk(root: &Path) -> u64 { + let Ok(rd) = std::fs::read_dir(root) else { + return 0; + }; + let mut total = 0; + for entry in rd.flatten() { + let Ok(meta) = entry.metadata() else { continue }; + if meta.is_dir() { + total += bytes_on_disk(&entry.path()); + } else if meta.is_file() { + total += meta.len(); + } + } + total + } + + /// C6/C7. Two claims the cache makes and does not keep. + /// + /// The configured budget is the whole promise of the Settings page: it is + /// the only thing standing between an on-demand filesystem and a full disk. + /// A thumbnail per photo, on a library of tens of thousands, is not a + /// rounding error. + #[test] + fn every_pool_is_budgeted_and_visible() { + const CAP: u64 = 256 * 1024; + let (c, dir) = cache_capped(CAP); + let payload = vec![0u8; 8 * 1024]; + + // A gallery scroll: one thumbnail per photo, far past the cap. + for i in 0..200u64 { + c.store_thumbnail(&uid(&format!("photo{i}")), 1, 1, &payload) + .unwrap(); + } + // Plus enough blob and block traffic to fill both other pools. + for i in 0..100u64 { + c.store(&uid(&format!("file{i}")), 1, payload.len() as u64, &payload) + .unwrap(); + c.store_block(&uid("streamed"), 1, 1 << 30, i, &payload) + .unwrap(); + } + + let on_disk = bytes_on_disk(dir.path()); + let reported = c.usage(); + + // 1. The cap is a cap on the cache, not on one pool of it. + assert!( + on_disk <= CAP + CAP / 8, + "cache holds {on_disk} bytes against a {CAP}-byte budget" + ); + // 2. And the number shown to the user is the number on disk. Meta + // sidecars are small and untracked, hence the tolerance. + let drift = on_disk.abs_diff(reported); + assert!( + drift <= on_disk / 8, + "usage() reports {reported} against {on_disk} on disk" + ); + } + + /// The thumbnail pool must evict by *use*, not by insertion order. A gallery + /// scrolled down and back up re-reads the tiles it already has; under FIFO + /// those are exactly the ones dropped, so every scroll back would re-fetch. + #[test] + fn thumbnails_evict_least_recently_used() { + // Pool fits two 4-byte thumbnails but not three. + let (c, _d) = cache_with_pool_cap(KIND_THUMB, 8); + let (a, b, d) = (uid("a"), uid("b"), uid("d")); + + c.store_thumbnail(&a, 1, 1, b"aaaa").unwrap(); + std::thread::sleep(std::time::Duration::from_millis(10)); + c.store_thumbnail(&b, 1, 1, b"bbbb").unwrap(); + // Look at `a` again, making `b` the least-recently-used. + std::thread::sleep(std::time::Duration::from_millis(10)); + assert!(c.read_thumbnail(&a, 1, 1).is_some()); + std::thread::sleep(std::time::Duration::from_millis(10)); + + c.store_thumbnail(&d, 1, 1, b"dddd").unwrap(); + assert!(c.read_thumbnail(&a, 1, 1).is_some(), "recently read, kept"); + assert!(c.read_thumbnail(&b, 1, 1).is_none(), "least recently used"); + assert!(c.read_thumbnail(&d, 1, 1).is_some(), "just stored"); + } + + /// Evicting a node drops its thumbnails from the index as well as from disk. + /// A row left behind would keep counting bytes that are gone, and the + /// running total would drift above the truth until the next restart. + #[test] + fn evicting_a_node_clears_its_thumbnail_accounting() { + let (c, _d) = cache(); + let u = uid("a"); + c.store(&u, 1, 4, b"aaaa").unwrap(); + c.store_thumbnail(&u, 1, 1, b"thumb1").unwrap(); + c.store_thumbnail(&u, 2, 1, b"preview2").unwrap(); + assert!(c.thumb_bytes.load(Ordering::Relaxed) > 0); + + c.evict(&u); + assert_eq!(c.thumb_bytes.load(Ordering::Relaxed), 0); + assert!(c.db.cache_entries_by_kind(KIND_THUMB).unwrap().is_empty()); + assert_eq!(c.usage(), 0, "nothing cached, nothing reported"); + } + #[test] fn evict_removes_blob() { let (c, _d) = cache(); @@ -1277,8 +1506,8 @@ mod tests { #[test] fn block_budget_evicts_lru() { - // Cap fits two 4-byte blocks but not three. - let (c, _d) = cache_capped(8); + // Block pool fits two 4-byte blocks but not three. + let (c, _d) = cache_with_pool_cap(KIND_BLOCK, 8); let u = uid("a"); c.store_block(&u, 1, 12, 0, b"aaaa").unwrap(); std::thread::sleep(std::time::Duration::from_millis(10)); diff --git a/crates/pdfs-core/src/db/cache.rs b/crates/pdfs-core/src/db/cache.rs index 817cdd8..2e7cbb0 100644 --- a/crates/pdfs-core/src/db/cache.rs +++ b/crates/pdfs-core/src/db/cache.rs @@ -52,14 +52,17 @@ impl Db { Ok(()) } - /// Drop the whole-file blob row for `key` plus every block row for the same - /// uid (`.b`). Called by `ContentCache::evict`, which removes the - /// blob and all of a uid's cached blocks at once. + /// Drop the whole-file blob row for `key` plus every derived row for the + /// same uid — blocks (`.b`) and thumbnails (`.t`). + /// Called by `ContentCache::evict`, which removes the blob, all of a uid's + /// cached blocks and all of its thumbnails at once; a row left behind here + /// would keep counting bytes that are no longer on disk. pub fn cache_remove_all(&self, key: &str) -> Result<()> { let conn = self.conn.lock(); conn.execute( - "DELETE FROM cache_entries WHERE cache_key = ?1 OR cache_key LIKE ?2", - params![key, format!("{key}.b%")], + "DELETE FROM cache_entries + WHERE cache_key = ?1 OR cache_key LIKE ?2 OR cache_key LIKE ?3", + params![key, format!("{key}.b%"), format!("{key}.t%")], )?; Ok(()) } diff --git a/crates/pdfs-core/src/db/mod.rs b/crates/pdfs-core/src/db/mod.rs index 02cc456..63e6d9f 100644 --- a/crates/pdfs-core/src/db/mod.rs +++ b/crates/pdfs-core/src/db/mod.rs @@ -52,6 +52,13 @@ pub use trash::StoredTrash; /// outliers rather than fighting the normal write path for disk. const WAL_SIZE_LIMIT: i64 = 64 * 1024 * 1024; +/// How long a statement waits for a database lock before giving up. +/// +/// Long enough to outlast any transaction this daemon writes (the largest, a +/// full-index FTS rebuild, is well under a second), short enough that a +/// genuinely stuck holder surfaces as an error rather than an apparent hang. +const BUSY_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(5); + /// Handle to the unified metadata database. /// /// Cheap to wrap in an `Arc`; clone the `Arc`, not this. All access goes through @@ -77,6 +84,17 @@ impl Db { // forever — a multi-GB file next to a database two orders of magnitude // smaller. Checkpointing is unaffected; only the file is truncated back. conn.pragma_update(None, "journal_size_limit", WAL_SIZE_LIMIT)?; + // Wait for a lock rather than failing instantly on SQLITE_BUSY. + // + // This is deliberately explicit and deliberately redundant: rusqlite + // already applies a five-second busy timeout of its own, so this changes + // nothing today. It is here so the value is *ours* — a behaviour the + // daemon depends on should not be an inherited default that a dependency + // bump can silently change. `open_bounds_the_wal_size` pins it. + // + // (Bare SQLite does default to 0, which is where the belief that this + // was unset came from. That default is not what we get.) + conn.busy_timeout(BUSY_TIMEOUT)?; let db = Self { conn: Mutex::new(conn), diff --git a/crates/pdfs-core/src/db/ops.rs b/crates/pdfs-core/src/db/ops.rs index 81498ae..dc25056 100644 --- a/crates/pdfs-core/src/db/ops.rs +++ b/crates/pdfs-core/src/db/ops.rs @@ -188,6 +188,29 @@ impl Db { Ok(Some(AttachedBlob { id, superseded })) } + /// Replace the sidecar of a queued op, for a baseline that has moved under + /// it. + /// + /// Exists for one caller: when *our own* upload seals a new revision while a + /// further write to the same node is already queued, that queued write's + /// [`StagedWrite::based_on`](crate::cache::StagedWrite::based_on) still + /// names the revision we just replaced. Left alone it would read as "another + /// device changed this file" and divert the write into a conflict copy over + /// nothing — a self-conflict. Restamping it is what keeps a chain of writes + /// to one file a chain rather than a pile of copies. + /// + /// Deliberately narrow: only the sidecar changes, so the blob, the attempt + /// count and the backoff all survive. Returns false when no such op is + /// queued, which is the ordinary case. + pub fn update_op_meta(&self, uid: &str, kind: &str, meta_json: &str) -> Result { + let conn = self.conn.lock(); + let n = conn.execute( + "UPDATE pending_op SET meta_json = ?3 WHERE uid = ?1 AND kind = ?2", + params![uid, kind, meta_json], + )?; + Ok(n > 0) + } + /// Point a queued create at a new parent and name, for a node renamed or /// moved before it ever reached the server. /// diff --git a/crates/pdfs-core/src/db/tests.rs b/crates/pdfs-core/src/db/tests.rs index 02b1b36..94f1c15 100644 --- a/crates/pdfs-core/src/db/tests.rs +++ b/crates/pdfs-core/src/db/tests.rs @@ -26,10 +26,60 @@ fn open_bounds_the_wal_size() { .unwrap(); assert_eq!(limit, WAL_SIZE_LIMIT); + // Without a busy timeout the default is 0: a lock held by anyone else fails + // the statement instantly instead of waiting for it to clear. + let busy: i64 = conn + .query_row("PRAGMA busy_timeout", [], |r| r.get(0)) + .unwrap(); + assert_eq!(busy, BUSY_TIMEOUT.as_millis() as i64); + drop(conn); drop(db); let _ = std::fs::remove_file(&path); } + +/// A second connection to the same file — a hand-started `pdfs mount` alongside +/// the systemd service, a backup tool, a stray `sqlite3` session — holds a write +/// lock briefly. Our write must wait for it rather than failing on the spot. +/// +/// **This passed before the busy timeout was set explicitly**, because rusqlite +/// already applies one. It is kept as a characterisation test, not as evidence +/// of a fix: it states the behaviour the daemon relies on, so that losing it +/// (dropping the pragma, or a dependency changing its default) fails here rather +/// than in front of a user. +#[test] +fn a_write_waits_for_a_competing_writer() { + use std::time::Duration; + + let path = std::env::temp_dir().join(format!("pdfs-db-busy-{}.db", std::process::id())); + let _ = std::fs::remove_file(&path); + let db = Db::open(&path).unwrap(); + + // A second process's connection takes an exclusive write lock and holds it + // for a moment, as any real transaction would. + let holder = { + let path = path.clone(); + std::thread::spawn(move || { + let other = rusqlite::Connection::open(&path).unwrap(); + other.busy_timeout(Duration::from_secs(5)).unwrap(); + other.execute_batch("BEGIN IMMEDIATE").unwrap(); + std::thread::sleep(Duration::from_millis(300)); + other.execute_batch("COMMIT").unwrap(); + }) + }; + // Let the holder actually acquire the lock before we contend for it. + std::thread::sleep(Duration::from_millis(50)); + + // This is the assertion: it blocks until the holder commits, rather than + // returning SQLITE_BUSY. + db.set_state_i64("busy_probe", 7).unwrap(); + + holder.join().unwrap(); + assert_eq!(db.state_i64("busy_probe").unwrap(), Some(7)); + + drop(db); + let _ = std::fs::remove_file(&path); +} use proton_drive_rs::proton_sdk::ids::{LinkId, VolumeId}; fn uid(link: &str) -> NodeUid { @@ -541,6 +591,56 @@ fn opens_and_migrates() { assert_eq!(version, SCHEMA_VERSION.to_string()); } +/// A queued write whose baseline is restamped must keep everything else. +/// +/// The restamp happens after our *own* upload seals a new revision under a +/// still-queued write. If it took the blob or the retry state with it, the fix +/// for a spurious conflict copy would cost the bytes that conflict copy was +/// there to protect. +#[test] +fn restamping_a_baseline_leaves_the_blob_and_retry_state_alone() { + let db = Db::open_in_memory().unwrap(); + let op = PendingOp { + id: 0, + kind: OP_REVISION.to_string(), + uid: uid("a").to_string(), + parent_uid: None, + name: None, + blob_path: Some("/staging/blob".to_string()), + meta_json: Some(r#"{"based_on":"old"}"#.to_string()), + created_at: 1, + attempts: 0, + last_error: None, + next_attempt_at: 0, + }; + db.enqueue_op(&op).unwrap(); + db.record_op_failure(db.pending_ops().unwrap()[0].id, "offline", 999) + .unwrap(); + + let updated = db + .update_op_meta(&uid("a").to_string(), OP_REVISION, r#"{"based_on":"new"}"#) + .unwrap(); + assert!(updated, "the queued write is there to restamp"); + + let ops = db.pending_ops().unwrap(); + assert_eq!(ops.len(), 1); + assert_eq!(ops[0].meta_json.as_deref(), Some(r#"{"based_on":"new"}"#)); + assert_eq!( + ops[0].blob_path.as_deref(), + Some("/staging/blob"), + "the only copy of the user's bytes" + ); + assert_eq!(ops[0].attempts, 1, "backoff survives a restamp"); + assert_eq!(ops[0].next_attempt_at, 999); + + // No queued write for that node is the ordinary case, not an error: most + // uploads are the last one for their file. + assert!( + !db.update_op_meta(&uid("b").to_string(), OP_REVISION, "{}") + .unwrap() + ); +} + #[test] fn a_second_write_supersedes_the_first_pending_op() { let db = Db::open_in_memory().unwrap(); @@ -1182,9 +1282,11 @@ fn next_due_op_respects_backoff() { /// 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. +/// Asserts correctness — the aggregate must agree with summing the rows, and +/// must count the kinds apart — and *reports* the timing without asserting on +/// it. See the note at the measurement for why. #[test] -fn cache_total_bytes_does_not_scale_with_cache_size() { +fn cache_total_bytes_agrees_with_summing_the_rows() { use std::time::Instant; let db = Db::open_in_memory().unwrap(); @@ -1223,17 +1325,14 @@ fn cache_total_bytes_does_not_scale_with_cache_size() { } let aggregate = t1.elapsed(); + // Reported, not asserted. The SUM is cheaper than reading every row — no + // materialization, no allocation, no sort — but it is still O(rows), so this + // is a constant factor, and an assertion on a constant factor measured + // alongside the rest of the suite tests the machine's load as much as the + // query. The claim that matters — that the *common* path does not touch the + // database at all — is pinned by `budget_check_is_free_when_under_budget`, + // which compares against itself and so is load-independent. 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 diff --git a/crates/pdfs-fuse/src/drain.rs b/crates/pdfs-fuse/src/drain.rs index 33d2829..4d644b7 100644 --- a/crates/pdfs-fuse/src/drain.rs +++ b/crates/pdfs-fuse/src/drain.rs @@ -24,9 +24,10 @@ use std::fs::File; use std::path::{Path, PathBuf}; use std::sync::atomic::Ordering; -use pdfs_core::cache::StagedWrite; +use pdfs_core::cache::{Baseline, StagedWrite}; use pdfs_core::control::{ActivityKind, TransferDirection}; use pdfs_core::db::{OP_CREATE, OP_MKDIR, OP_RENAME, OP_REVISION, OP_TRASH, PendingOp}; +use proton_drive_rs::Node; use proton_drive_rs::proton_sdk::ids::NodeUid; use tracing::{debug, error, info, warn}; @@ -511,7 +512,7 @@ impl Core { let name = place .as_ref() .map(|(_, n)| n.clone()) - .unwrap_or_else(|| meta.uid.clone()); + .unwrap_or_else(|| self.node_name(uid)); // A node the tree has forgotten still has bytes worth keeping, so the // copy falls back to the root rather than being abandoned. let Some(parent) = place.map(|(p, _)| p).or_else(|| self.root_uid()) else { @@ -625,11 +626,38 @@ impl Core { /// A node's parent uid and name, as the tree currently has them. pub(crate) fn node_place(&self, uid: &NodeUid) -> Option<(NodeUid, String)> { - let st = self.state.lock(); - let ino = st.by_uid.get(uid)?; - let entry = st.entries.get(ino)?; - let parent = entry.node.parent_uid.clone()?; - Some((parent, entry.node.name.clone())) + { + let st = self.state.lock(); + if let Some(entry) = st.by_uid.get(uid).and_then(|ino| st.entries.get(ino)) + && let Some(parent) = entry.node.parent_uid.clone() + { + return Some((parent, entry.node.name.clone())); + } + } + // The in-memory tree only holds what has been walked to since the daemon + // started, and the drain routinely runs before anything has walked to + // this file — after a restart, or for a write that arrived by path. The + // DB knows it anyway. Without this fallback a conflict copy was named + // after the node's *uid* and dumped in the root, which is how a file + // called `G88km…==~c2do…== (sync-conflict 1784429627)` appears at the + // top of someone's Drive. + let node = self.db.node_by_uid(&uid.to_string()).ok().flatten()?; + Some((node.parent_uid.clone()?, node.name.clone())) + } + + /// The name to show a user for `uid`, for a transfer entry or an activity + /// log line. + /// + /// Falls back to something short and human rather than to the uid: a uid is + /// 130 characters of base64 that identifies the file to us and to nobody + /// else, and it has ended up in both the activity log and on real files in + /// the Drive root. The link prefix keeps it traceable without pretending to + /// be a name. + pub(crate) fn node_name(&self, uid: &NodeUid) -> String { + self.node_place(uid).map(|(_, n)| n).unwrap_or_else(|| { + let short: String = uid.link_id.to_string().chars().take(8).collect(); + format!("recovered-{short}") + }) } /// The uid of the My Files root, which every node in the mount descends @@ -681,14 +709,7 @@ impl Core { .map_err(|e| self.errno_error(e, "gap-fill from base failed"))?; } - let name = { - let st = self.state.lock(); - st.by_uid - .get(&uid) - .and_then(|ino| st.entries.get(ino)) - .map(|e| e.node.name.clone()) - .unwrap_or_else(|| meta.uid.clone()) - }; + let name = self.node_name(&uid); let guard = self.transfers .begin(&name, meta.uid.clone(), TransferDirection::Upload, meta.len); @@ -736,15 +757,16 @@ impl Core { /// makes the *next* write to this file look like another device changed it /// underneath us, and diverts it into a conflict copy over nothing. /// + /// A write queued while this upload was on the wire is handled instead by + /// [`Core::rebaseline_pending`]: its optimistic size/mtime must stay on the + /// node (the same reason `apply_event` refuses to overwrite a node with a + /// queued write), but its *baseline* still has to learn about the revision + /// we just sealed. Skipping both, as this used to, is what produced a file + /// that conflicted with itself and left a full-size duplicate behind. + /// /// Best effort: a failure here costs a spurious conflict copy on the next /// write, not this upload, which has already landed. pub(crate) fn refresh_after_upload(&self, uid: &NodeUid) { - // A write queued while this one was on the wire is ahead of the remote - // again, and its optimistic size/mtime must win — the same reason - // `apply_event` refuses to overwrite a node with a queued write. - if self.pending.lock().contains_key(uid) { - return; - } let node = match self.fetch_node_remote(uid) { Ok(Some(node)) => node, // Trashed or deleted under us: the tree will hear it from the event @@ -757,6 +779,12 @@ impl Core { return; } }; + // Ordered so that a write queued *during* the fetch above is still + // caught: it took its baseline from the node's optimistic stamp, and + // this overwrites it with the revision the server actually holds. + if self.rebaseline_pending(uid, &node) { + return; + } let mut st = self.state.lock(); let Some(parent) = st .by_uid @@ -768,4 +796,45 @@ impl Core { }; st.intern(parent, node); } + + /// Point a still-queued write at the revision this upload just sealed, and + /// report whether there was one. + /// + /// The revision we sent *is* the base the queued write will be applied over, + /// but nothing else says so. [`Core::remote_baseline`] carries a baseline + /// across a supersede because the op it inherits from "is the last one that + /// actually observed the remote" — true only until that op drains. Once it + /// has, the inherited baseline names a revision we replaced ourselves, and + /// [`Core::revision_conflict`] reads that as another device having moved the + /// file. It is a self-conflict, and it is expensive: the whole staged blob + /// is re-uploaded as a second file. + /// + /// Both copies of the sidecar are updated — the in-memory one the next + /// `release` inherits from, and the persisted one the drain reloads after a + /// restart. Best effort on the DB half; the in-memory half is what the + /// immediate next write reads. + fn rebaseline_pending(&self, uid: &NodeUid, sealed: &Node) -> bool { + let base = Baseline { + mtime: sealed.modification_time, + size: node_size(sealed), + }; + let mut pending = self.pending.lock(); + let Some(p) = pending.get_mut(uid) else { + return false; + }; + p.meta.based_on = Some(base); + match serde_json::to_string(&p.meta) { + Ok(json) => { + if let Err(e) = self.db.update_op_meta(&uid.to_string(), OP_REVISION, &json) { + warn!(%uid, error = %e, + "restamping a queued write's baseline failed; \ + it may land as a conflict copy of itself"); + } + } + Err(e) => warn!(%uid, error = %e, "serializing a restamped sidecar failed"), + } + debug!(%uid, mtime = base.mtime, size = base.size, + "queued write rebaselined onto the revision just uploaded"); + true + } } diff --git a/crates/pdfs-fuse/src/workers.rs b/crates/pdfs-fuse/src/workers.rs index bad015f..bf9f68d 100644 --- a/crates/pdfs-fuse/src/workers.rs +++ b/crates/pdfs-fuse/src/workers.rs @@ -49,11 +49,26 @@ impl Workers { // `Reply` answers EIO on its own, so the caller of // the failed op is told; the next job is unaffected. // - // This only holds because the shared state is behind - // `parking_lot` mutexes: a `std` mutex held at the - // point of the panic would come back poisoned, and - // the worker we just rescued would die on its next - // acquisition — as would every other thread. + // This only holds while no shared state a job + // touches sits behind a *poisoning* lock. A + // `std::sync::Mutex` held at the point of the panic + // comes back poisoned, and the worker we just + // rescued dies on its next acquisition — as does + // every other thread. The rescue would then convert + // one recoverable EIO into a permanently broken + // daemon: strictly worse than not catching at all. + // + // This crate's state is behind `parking_lot`, which + // does not poison. **The requirement crosses the + // crate boundary**, though: a job runs deep into + // `proton-sdk`, whose entity cache is a `std` + // mutex. It recovers the guard rather than + // unwrapping (see `InMemoryCacheRepository::state`); + // its session and HTTP state are + // `tokio::sync::Mutex`, which has no poisoning at + // all. Anything new reached from a job owes the same + // check — an SDK-side `.lock().unwrap()` silently + // invalidates this comment. Ok(job) => { let _ = std::panic::catch_unwind(std::panic::AssertUnwindSafe(job)); }