From e396c6b4d672489d1ffddb233f0f07d3a63d3c0e Mon Sep 17 00:00:00 2001 From: Nils Pukropp Date: Sat, 15 Aug 2026 18:49:07 +0200 Subject: [PATCH] fix(fs) performance, locks --- crates/pdfs-core/src/cache.rs | 289 ++++++++++++---- crates/pdfs-core/src/db/cache.rs | 24 ++ crates/pdfs-core/src/db/migrations.rs | 52 ++- crates/pdfs-core/src/db/mod.rs | 32 ++ crates/pdfs-core/src/db/pins.rs | 26 +- crates/pdfs-core/src/db/tests.rs | 20 ++ crates/pdfs-core/src/db/utils.rs | 2 +- crates/pdfs-core/src/error.rs | 21 ++ crates/pdfs-fuse/src/background.rs | 77 ++--- crates/pdfs-fuse/src/control.rs | 46 ++- crates/pdfs-fuse/src/devices.rs | 16 +- crates/pdfs-fuse/src/drain.rs | 38 ++- crates/pdfs-fuse/src/filesystem.rs | 464 ++++++++++++++++---------- crates/pdfs-fuse/src/lib.rs | 159 ++++++++- crates/pdfs-fuse/src/mount.rs | 9 + crates/pdfs-fuse/src/sharing.rs | 27 ++ crates/pdfs-fuse/src/state.rs | 20 +- crates/pdfs-fuse/src/sweep.rs | 21 +- 18 files changed, 1000 insertions(+), 343 deletions(-) diff --git a/crates/pdfs-core/src/cache.rs b/crates/pdfs-core/src/cache.rs index 392377e..35d5331 100644 --- a/crates/pdfs-core/src/cache.rs +++ b/crates/pdfs-core/src/cache.rs @@ -16,7 +16,7 @@ //! [`AppDirs::cache_dir`]: crate::config::AppDirs::cache_dir //! [`Node`]: proton_drive_rs::Node -use std::collections::{BTreeMap, HashSet}; +use std::collections::{BTreeMap, HashMap, HashSet}; use std::fmt; use std::io::{Read, Seek, SeekFrom}; use std::path::{Path, PathBuf}; @@ -27,6 +27,9 @@ use std::time::{SystemTime, UNIX_EPOCH}; use proton_drive_rs::proton_sdk::ids::NodeUid; use serde::{Deserialize, Serialize}; use sha2::{Digest, Sha256}; +use tracing::debug; + +use parking_lot::Mutex; use crate::db::{CacheEntryInput, Db}; use crate::error::{Error, Result}; @@ -74,6 +77,18 @@ fn now_millis() -> i64 { /// `download_range` fetch with no straddling. pub const BLOCK_SIZE: u64 = 1 << 22; +/// Buffered LRU touches that trigger a flush. Sized so a sequential read of a +/// large file flushes a few times rather than per block, while an idle cache +/// never holds more than a few hundred pending keys. +const TOUCH_FLUSH_AT: usize = 256; + +/// How long block `idx` of a file of `size` bytes must be: a full +/// [`BLOCK_SIZE`] except for the last one, which holds the remainder. +fn block_len(size: u64, idx: u64) -> u64 { + let start = idx.saturating_mul(BLOCK_SIZE); + size.saturating_sub(start).min(BLOCK_SIZE) +} + /// 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 @@ -263,6 +278,15 @@ pub struct ContentCache { /// store/read/evict updates it, and the budget enforcers query it instead of /// scanning the cache directories (plan.md P4). db: Arc, + /// LRU touches not yet written to `cache_entries`, keyed by cache key. + /// + /// A cache hit only reorders eviction candidates, so it does not have to be + /// durable — but as one write transaction per hit it was 32 transactions per + /// 4 MiB block delivered, taken on the read path, under the connection mutex + /// every `lookup` also needs. Buffered here and flushed in one transaction + /// once [`TOUCH_FLUSH_AT`] have accumulated, and always before an eviction + /// pass reads the ordering these describe. + touches: Mutex>, } impl ContentCache { @@ -325,6 +349,7 @@ impl ContentCache { block_bytes: AtomicU64::new(0), thumb_bytes: AtomicU64::new(0), db, + touches: Mutex::new(HashMap::new()), }; cache.reconcile()?; // After reconcile, so the totals describe the index that now describes @@ -473,7 +498,7 @@ impl ContentCache { let blob = self.valid_blob(uid, mtime, size)?; // Record the access for LRU in the cache index. Best effort — a failed // update only makes eviction order slightly less accurate. - let _ = self.db.cache_accessed(&Self::key(uid), now_millis()); + self.touch(Self::key(uid)); let mut f = std::fs::File::open(&blob).ok()?; if offset >= size { return Some(Vec::new()); @@ -575,10 +600,15 @@ impl ContentCache { return None; } let mut f = std::fs::File::open(self.block_blob(uid, idx)).ok()?; - // LRU touch in the cache index (best effort). - let _ = self - .db - .cache_accessed(&self.block_key(uid, idx), now_millis()); + // The blob is not fsynced on the way in (see `store_block`), so a crash + // can leave a short one behind the sidecar that describes it. Length is + // what makes that detectable, the same check `valid_blob` applies to a + // whole-file blob. + if f.metadata().ok()?.len() != block_len(size, idx) { + return None; + } + // LRU touch, buffered rather than written (see `touch`). + self.touch(self.block_key(uid, idx)); let mut buf = Vec::new(); f.read_to_end(&mut buf).ok()?; Some(buf) @@ -603,7 +633,10 @@ impl ContentCache { let mut f = std::fs::File::create(&tmp)?; use std::io::Write; f.write_all(bytes)?; - f.sync_all()?; + // Deliberately not fsynced. This runs inline on the thread a kernel + // read is waiting on, and the block cache is disposable: the worst a + // crash can do is leave a short blob, which `cached_block` rejects + // on length and refetches. } std::fs::rename(&tmp, &blob)?; std::fs::write( @@ -622,6 +655,40 @@ impl ContentCache { Ok(()) } + /// Record a cache hit for LRU ordering, without touching the database. + /// + /// Later touches of the same key overwrite earlier ones — only the most + /// recent matters — so a hot file costs one map entry however many times it + /// is read. + fn touch(&self, key: String) { + let flush = { + let mut touches = self.touches.lock(); + touches.insert(key, now_millis()); + touches.len() >= TOUCH_FLUSH_AT + }; + if flush { + self.flush_touches(); + } + } + + /// Write the buffered LRU touches out. Best effort: losing them costs + /// eviction accuracy, nothing else. + /// + /// Public because eviction ordering is only as good as what has been + /// flushed, and the daemon flushes on its idle passes as well as here. + pub fn flush_touches(&self) { + let batch: Vec<(String, i64)> = { + let mut touches = self.touches.lock(); + if touches.is_empty() { + return; + } + touches.drain().collect() + }; + if let Err(error) = self.db.cache_accessed_batch(&batch) { + debug!(?error, "flushing cache LRU touches failed"); + } + } + /// Re-read every pool total from the cache index, which is authoritative. fn reseed_totals(&self) { for kind in [KIND_BLOB, KIND_BLOCK, KIND_THUMB] { @@ -689,7 +756,10 @@ impl ContentCache { return; // the common case: one atomic load, no query } - // Over budget by the running count — now do the accurate work. + // Over budget by the running count — now do the accurate work. The + // buffered hits are what say which entries are cold, so they have to + // reach the index before it is asked. + self.flush_touches(); let mut running = match self.db.cache_total_bytes(kind) { Ok(n) => n, Err(_) => return, @@ -727,6 +797,70 @@ impl ContentCache { self.reseed_totals(); } + /// Duplicate `src` onto `dst`, preferring a reflink. + /// + /// The caller is copying a staged revision — a whole file, up to whatever + /// the user stores — so the difference between the three strategies is the + /// difference between instant and minutes: + /// + /// 1. `FICLONE` shares the extents copy-on-write. On btrfs, XFS with + /// `reflink=1`, and bcachefs this is O(1) and consumes no extra space. + /// 2. `copy_file_range` still copies, but entirely inside the kernel — no + /// round trip through a user-space buffer. + /// 3. `std::fs::copy` is the portable fallback, and what this used to + /// always do. + /// + /// `dst` is truncated. Returns the number of bytes the destination ends up + /// holding. + pub fn clone_or_copy(src: &Path, dst: &Path) -> Result { + use std::os::fd::AsRawFd as _; + + // `_IOW(0x94, 9, int)` — the whole-file reflink ioctl. + const FICLONE: libc::c_ulong = 0x4004_9409; + + let source = std::fs::File::open(src)?; + let len = source.metadata()?.len(); + let target = std::fs::OpenOptions::new() + .write(true) + .create(true) + .truncate(true) + .open(dst)?; + + // SAFETY: both descriptors are owned and open; FICLONE takes the source + // fd by value and writes nothing back through the pointer-shaped arg. + if unsafe { libc::ioctl(target.as_raw_fd(), FICLONE, source.as_raw_fd()) } == 0 { + return Ok(len); + } + + let mut copied = 0u64; + while copied < len { + let want = (len - copied) as usize; + // SAFETY: owned descriptors; null offsets advance the file offsets, + // which is what a whole-file copy wants. + let n = unsafe { + libc::copy_file_range( + source.as_raw_fd(), + std::ptr::null_mut(), + target.as_raw_fd(), + std::ptr::null_mut(), + want, + 0, + ) + }; + if n <= 0 { + break; + } + copied += n as u64; + } + if copied == len { + return Ok(len); + } + // A short or refused `copy_file_range` (cross-filesystem on older + // kernels, an unsupported fs) leaves a partial destination; the + // portable copy truncates and rewrites it from scratch. + Ok(std::fs::copy(src, dst)?) + } + /// Create a fresh, empty read-write scratch file for a disk-backed write /// handle. Returns the open file and its path (for cleanup on release). pub fn create_scratch(&self) -> Result<(std::fs::File, PathBuf)> { @@ -806,8 +940,9 @@ impl ContentCache { let _ = std::fs::remove_file(Self::scratch_sidecar(scratch)); } - /// Move every scratch file that carries a readable sidecar into `recovery`, - /// before the caller empties the scratch directory. + /// Move every scratch file that carries a sidecar into `recovery`, before the + /// caller empties the scratch directory. A sidecar that does not parse is + /// dropped and its blob moved anyway, unaccompanied — see below. /// /// Blob first, then sidecar: a blob in recovery without its sidecar is /// dropped on the next open (indistinguishable from rubbish), whereas a @@ -828,43 +963,28 @@ impl ContentCache { // Not marked durable, or durability marker was explicitly cleared on release continue; } - // Parsed if valid; if sidecar exists but is corrupted, build synthetic StagedWrite + // A sidecar that does not parse describes nothing, and inventing one + // is worse than having none: a fabricated uid parses, so recovery + // would queue an op against a node that does not exist and retry it + // forever, while a fabricated `complete: true` claims the blob is the + // whole file and stops any gap-fill. The blob still moves to + // recovery — sidecar-less, which `recovered_writes` skips and so + // leaves on disk for a human, the right outcome for bytes we cannot + // describe. let has_valid_sidecar = std::fs::read(&side) .ok() .and_then(|b| serde_json::from_slice::(&b).ok()) .is_some(); - if !has_valid_sidecar { - if let Ok(meta) = std::fs::metadata(&blob) - && meta.len() > 0 - { - let blob_name = blob - .file_name() - .map(|n| n.to_string_lossy().to_string()) - .unwrap_or_default(); - let synthetic = StagedWrite { - uid: format!("recovered~{blob_name}"), - len: meta.len(), - base_size: meta.len(), - base_mtime: 0, - authored: vec![(0, meta.len())], - complete: true, - based_on: None, - }; - if let Ok(data) = serde_json::to_vec(&synthetic) { - let _ = std::fs::write(&side, data); - } else { - continue; - } - } else { - continue; - } - } let Some(name) = blob.file_name() else { continue; }; if std::fs::rename(&blob, recovery_dir.join(name)).is_err() { continue; } + if !has_valid_sidecar { + let _ = std::fs::remove_file(&side); + continue; + } let Some(side_name) = side.file_name() else { continue; }; @@ -1210,9 +1330,7 @@ impl ContentCache { // 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()); + self.touch(self.thumb_key(uid, ttype)); Some(bytes) } @@ -1243,9 +1361,7 @@ impl ContentCache { if !blob.exists() { return None; } - let _ = self - .db - .cache_accessed(&self.thumb_key(uid, ttype), now_millis()); + self.touch(self.thumb_key(uid, ttype)); Some(blob) } @@ -2004,45 +2120,63 @@ mod tests { fn block_store_then_read() { let (c, _d) = cache(); let u = uid("a"); - c.store_block(&u, 100, 4096, 0, b"block-zero").unwrap(); - assert_eq!(c.cached_block(&u, 100, 4096, 0).unwrap(), b"block-zero"); + // A 10-byte file is one block of 10 bytes: the length a reader expects + // is derived from the file size, so the fixture has to be geometrically + // real or the torn-write check rejects it. + c.store_block(&u, 100, 10, 0, b"block-zero").unwrap(); + assert_eq!(c.cached_block(&u, 100, 10, 0).unwrap(), b"block-zero"); // A different index is a separate cache entry, absent here. - assert!(c.cached_block(&u, 100, 4096, 1).is_none()); + assert!(c.cached_block(&u, 100, 10, 1).is_none()); // A new revision (mtime/size bump) invalidates the block. - assert!(c.cached_block(&u, 101, 4096, 0).is_none()); + assert!(c.cached_block(&u, 101, 10, 0).is_none()); assert!(c.cached_block(&u, 100, 5000, 0).is_none()); } + /// The block cache is written without an fsync, so a crash can leave a blob + /// shorter than the sidecar beside it claims. Serving that as content would + /// hand the reader a hole; it has to read as a miss. + #[test] + fn a_truncated_block_reads_as_a_miss() { + let (c, _d) = cache(); + let u = uid("a"); + c.store_block(&u, 100, 10, 0, b"block-zero").unwrap(); + std::fs::write(c.block_blob(&u, 0), b"block").unwrap(); + assert!(c.cached_block(&u, 100, 10, 0).is_none()); + } + #[test] fn evict_drops_blocks() { let (c, _d) = cache(); let u = uid("a"); - c.store_block(&u, 1, 8, 0, b"aaaa").unwrap(); - c.store_block(&u, 1, 8, 1, b"bbbb").unwrap(); + // Both blocks are stored at their real geometry (a 4-byte file is one + // block), so this fails if `evict` misses them rather than because the + // fixture was unreadable to begin with. + c.store_block(&u, 1, 4, 0, b"aaaa").unwrap(); + assert!(c.cached_block(&u, 1, 4, 0).is_some()); c.evict(&u); - assert!(c.cached_block(&u, 1, 8, 0).is_none()); - assert!(c.cached_block(&u, 1, 8, 1).is_none()); + assert!(c.cached_block(&u, 1, 4, 0).is_none()); } #[test] fn block_budget_evicts_lru() { - // Block pool fits two 4-byte blocks but not three. + // Block pool fits two 4-byte blocks but not three. One block per file, + // so each fixture is a geometrically real single-block file. let (c, _d) = cache_with_pool_cap(KIND_BLOCK, 8); - let u = uid("a"); - c.store_block(&u, 1, 12, 0, b"aaaa").unwrap(); + let (a, b, d) = (uid("a"), uid("b"), uid("d")); + c.store_block(&a, 1, 4, 0, b"aaaa").unwrap(); std::thread::sleep(std::time::Duration::from_millis(10)); - c.store_block(&u, 1, 12, 1, b"bbbb").unwrap(); - // Touch block 0 so block 1 is the least-recently-used. + c.store_block(&b, 1, 4, 0, b"bbbb").unwrap(); + // Touch a's block so b's is the least-recently-used. std::thread::sleep(std::time::Duration::from_millis(10)); - assert!(c.cached_block(&u, 1, 12, 0).is_some()); + assert!(c.cached_block(&a, 1, 4, 0).is_some()); std::thread::sleep(std::time::Duration::from_millis(10)); - c.store_block(&u, 1, 12, 2, b"cccc").unwrap(); + c.store_block(&d, 1, 4, 0, b"cccc").unwrap(); assert!( - c.cached_block(&u, 1, 12, 0).is_some(), + c.cached_block(&a, 1, 4, 0).is_some(), "recently-read survives" ); - assert!(c.cached_block(&u, 1, 12, 1).is_none(), "LRU block evicted"); - assert!(c.cached_block(&u, 1, 12, 2).is_some(), "newest survives"); + assert!(c.cached_block(&b, 1, 4, 0).is_none(), "LRU block evicted"); + assert!(c.cached_block(&d, 1, 4, 0).is_some(), "newest survives"); } /// A failed upload must never cost the user their bytes (offline.md Phase 2): @@ -2185,6 +2319,39 @@ mod tests { drop(reopened); } + /// `clone_or_copy` has three strategies and only the last one is portable, + /// so what matters is that the destination is byte-identical whichever the + /// filesystem under `/tmp` happens to take — including when it already held + /// longer content, which a reflink or a short `copy_file_range` must not + /// leave a tail of. + #[test] + fn clone_or_copy_reproduces_the_source_over_any_existing_destination() { + let dir = TempDir::new(); + let src = dir.path().join("src"); + let dst = dir.path().join("dst"); + std::fs::write(&src, b"the new content").unwrap(); + std::fs::write(&dst, vec![b'x'; 4096]).unwrap(); + + let len = ContentCache::clone_or_copy(&src, &dst).unwrap(); + + assert_eq!(len, 15); + assert_eq!(std::fs::read(&dst).unwrap(), b"the new content"); + } + + /// An empty staged blob is a real case — a file truncated to zero and + /// closed — and `FICLONE` of a zero-length source is a documented `EINVAL`. + #[test] + fn clone_or_copy_handles_an_empty_source() { + let dir = TempDir::new(); + let src = dir.path().join("empty"); + let dst = dir.path().join("dst"); + std::fs::write(&src, b"").unwrap(); + std::fs::write(&dst, b"stale").unwrap(); + + assert_eq!(ContentCache::clone_or_copy(&src, &dst).unwrap(), 0); + assert!(std::fs::read(&dst).unwrap().is_empty()); + } + /// A2: `fsync(2)` promises the bytes survive a crash, but queueing happens /// at `release`. A scratch file marked durable must therefore outlive the /// open that clears the scratch directory — and an unmarked one must not, diff --git a/crates/pdfs-core/src/db/cache.rs b/crates/pdfs-core/src/db/cache.rs index 7d20296..8c585bf 100644 --- a/crates/pdfs-core/src/db/cache.rs +++ b/crates/pdfs-core/src/db/cache.rs @@ -42,6 +42,30 @@ impl Db { Ok(()) } + /// Bump `last_accessed` for many keys in one transaction. + /// + /// The read path touches a key per 128 KiB the kernel asks for — 32 times + /// per 4 MiB block — and each of those was its own write transaction under + /// the single connection mutex, contending with every `lookup` and + /// `readdir`. Callers accumulate touches in memory and flush them here. + pub fn cache_accessed_batch(&self, touches: &[(String, i64)]) -> Result<()> { + if touches.is_empty() { + return Ok(()); + } + let mut conn = self.conn.lock(); + let tx = conn.transaction()?; + { + let mut stmt = tx.prepare_cached( + "UPDATE cache_entries SET last_accessed = ?2 WHERE cache_key = ?1", + )?; + for (key, at) in touches { + stmt.execute(params![key, at])?; + } + } + tx.commit()?; + Ok(()) + } + /// Drop a single cache-entry row (one evicted blob or block). pub fn cache_remove(&self, key: &str) -> Result<()> { let conn = self.conn.lock(); diff --git a/crates/pdfs-core/src/db/migrations.rs b/crates/pdfs-core/src/db/migrations.rs index de7ec96..5ec5f3e 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 = 21; +pub(super) const SCHEMA_VERSION: i64 = 22; impl Db { pub(super) fn migrate(&self) -> Result<()> { @@ -113,6 +113,24 @@ impl Db { tx.execute_batch(MIGRATION_V21)?; } } + if current < 22 { + // Older fixtures (and a database that never grew a table because + // its feature was never used) may be missing any of these, and an + // index on an absent table is a hard error rather than a no-op. + for (table, sql) in MIGRATION_V22 { + let present: bool = tx.query_row( + "SELECT COUNT(*) FROM sqlite_master WHERE type = 'table' AND name = ?1", + [table], + |row| row.get::<_, i64>(0), + )? > 0; + if present { + tx.execute_batch(sql)?; + } + } + // `activity_time` indexes `time DESC` while both activity queries + // order by `id`: it has only ever cost writes. + tx.execute_batch("DROP INDEX IF EXISTS activity_time;")?; + } tx.execute( "INSERT INTO sync_state (key, value) VALUES ('schema_version', ?1) ON CONFLICT(key) DO UPDATE SET value = excluded.value", @@ -583,3 +601,35 @@ SELECT n.rowid, n.name, paths.path FROM nodes n JOIN paths ON paths.rowid = n.rowid WHERE paths.trashed = 0; "; + +/// V22: indexes for the lookups that were scanning, each paired with the table +/// it needs to exist for. +/// +/// Each backs a query on a hot path that had no index to use: +/// +/// - `sync_entry(remote_uid)` — the reconcile resolves a remote uid to its +/// local entry once per node, per pass. +/// - `local_files(scan_gen)` — the sweep that retires the previous generation +/// after a local index scan. +/// - `nodes(parent_uid) WHERE trashed = 0` — every listing, with the partial +/// index keeping it to the live rows a listing actually wants. +/// +/// Deliberately *not* here: an index on `pending_op(next_attempt_at, id)`. +/// `next_due_op` orders by `id`, which is the rowid, so it already stops at the +/// first due row; offering the planner a `next_attempt_at` index makes it scan +/// that index and then sort the matches by `id` instead — +/// `next_due_op_does_not_scale_with_queue_length` measures the regression. +const MIGRATION_V22: [(&str, &str); 3] = [ + ( + "sync_entry", + "CREATE INDEX IF NOT EXISTS idx_sync_entry_remote_uid ON sync_entry(remote_uid);", + ), + ( + "local_files", + "CREATE INDEX IF NOT EXISTS idx_local_files_scan_gen ON local_files(scan_gen);", + ), + ( + "nodes", + "CREATE INDEX IF NOT EXISTS idx_nodes_parent_live ON nodes(parent_uid) WHERE trashed = 0;", + ), +]; diff --git a/crates/pdfs-core/src/db/mod.rs b/crates/pdfs-core/src/db/mod.rs index 7ac02fb..b16f743 100644 --- a/crates/pdfs-core/src/db/mod.rs +++ b/crates/pdfs-core/src/db/mod.rs @@ -66,6 +66,20 @@ const WAL_SIZE_LIMIT: i64 = 64 * 1024 * 1024; /// genuinely stuck holder surfaces as an error rather than an apparent hang. const BUSY_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(5); +/// Page cache, in KiB (applied negated, which is how SQLite reads a size in KiB +/// rather than in pages). 32 MiB against a database that is typically tens of +/// MiB: large enough to hold the working set of a listing-heavy session, small +/// enough to be unremarkable in a desktop daemon's RSS. +const CACHE_SIZE_KIB: i64 = 32 * 1024; + +/// Ceiling on the memory-mapped window over the database file. 256 MiB covers +/// any realistic index while keeping the mapping bounded. +const MMAP_SIZE: i64 = 256 * 1024 * 1024; + +/// WAL pages between automatic checkpoints (default 1000, i.e. ~4 MiB). Halved +/// so the synchronous checkpoint a committing thread inherits stays short. +const WAL_AUTOCHECKPOINT_PAGES: i64 = 500; + /// Handle to the unified metadata database. /// /// Cheap to wrap in an `Arc`; clone the `Arc`, not this. All access goes through @@ -102,6 +116,24 @@ impl Db { // (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)?; + // Page cache. The default is 2 MiB, against a schema whose hot table + // stores a JSON blob per node — a listing of a few thousand files + // evicts itself while it is being read. + conn.pragma_update(None, "cache_size", -CACHE_SIZE_KIB)?; + // Sorts and recursive CTEs materialise their intermediate results. + // Every `ORDER BY … COLLATE NOCASE` listing and every ancestor walk + // spills those to a file in `/tmp` by default; they are small and + // short-lived, so memory is both faster and less surprising. + conn.pragma_update(None, "temp_store", "MEMORY")?; + // Read the database through the page cache of the OS rather than + // `pread` per page. Bounded rather than unlimited so the daemon's RSS + // does not track the database size. + conn.pragma_update(None, "mmap_size", MMAP_SIZE)?; + // Checkpointing is synchronous in whichever thread happens to commit + // the transaction that crosses the threshold — often a FUSE callback. + // A smaller autocheckpoint makes that debt smaller and more frequent + // rather than rare and multi-second. + conn.pragma_update(None, "wal_autocheckpoint", WAL_AUTOCHECKPOINT_PAGES)?; let db = Self { conn: Mutex::new(conn), diff --git a/crates/pdfs-core/src/db/pins.rs b/crates/pdfs-core/src/db/pins.rs index 2631af9..a89f046 100644 --- a/crates/pdfs-core/src/db/pins.rs +++ b/crates/pdfs-core/src/db/pins.rs @@ -4,6 +4,7 @@ use rusqlite::{OptionalExtension, params}; use super::Db; +use super::utils::MAX_PATH_DEPTH; use crate::Result; /// One row of [`Db::pin_list`]. `recursive` is pin *policy* — whether the whole @@ -66,6 +67,11 @@ impl Db { /// is pinned recursively. The ancestor check walks `parent_uid` to the root /// via a CTE; a direct pin is honoured even when the node has no `nodes` row /// yet (e.g. a CLI that never hydrates the node cache). + /// + /// The walk is depth-capped like [`super::utils::path_of`]: a `parent_uid` + /// cycle is corrupt data the API can hand us, and `UNION ALL` over one never + /// terminates — while holding the daemon's only SQLite connection, on a path + /// that runs per cached read. pub fn is_pinned(&self, uid: &str) -> Result { let conn = self.conn.lock(); let direct: Option = conn @@ -78,14 +84,18 @@ impl Db { } let anc: Option = conn .query_row( - "WITH RECURSIVE anc(uid, parent_uid) AS ( - SELECT uid, parent_uid FROM nodes WHERE uid = ?1 - UNION ALL - SELECT n.uid, n.parent_uid FROM nodes n JOIN anc ON n.uid = anc.parent_uid - ) - SELECT 1 FROM anc a JOIN pins p ON p.uid = a.uid - WHERE a.uid != ?1 AND p.recursive = 1 - LIMIT 1", + &format!( + "WITH RECURSIVE anc(uid, parent_uid, depth) AS ( + SELECT uid, parent_uid, 0 FROM nodes WHERE uid = ?1 + UNION ALL + SELECT n.uid, n.parent_uid, anc.depth + 1 + FROM nodes n JOIN anc ON n.uid = anc.parent_uid + WHERE anc.depth < {MAX_PATH_DEPTH} + ) + SELECT 1 FROM anc a JOIN pins p ON p.uid = a.uid + WHERE a.uid != ?1 AND p.recursive = 1 + LIMIT 1" + ), params![uid], |r| r.get(0), ) diff --git a/crates/pdfs-core/src/db/tests.rs b/crates/pdfs-core/src/db/tests.rs index de25d79..23c85e5 100644 --- a/crates/pdfs-core/src/db/tests.rs +++ b/crates/pdfs-core/src/db/tests.rs @@ -2628,6 +2628,26 @@ fn non_recursive_folder_pin_does_not_cover_children() { assert!(!db.is_pinned(&du("rep")).unwrap()); } +/// A `parent_uid` cycle is corrupt data the API can hand us, and `is_pinned`'s +/// ancestor walk is `UNION ALL` — uncapped it never terminates, while holding +/// the daemon's only SQLite connection, on a path that runs per cached read. +/// The answer here matters less than the fact that there is one. +#[test] +fn is_pinned_terminates_on_a_parent_cycle() { + let db = Db::open_in_memory().unwrap(); + let du = |l: &str| uid(l).to_string(); + // a -> b -> a, with nothing pinned above either. + db.upsert_node(&folder("a", Some("b"), "A")).unwrap(); + db.upsert_node(&folder("b", Some("a"), "B")).unwrap(); + db.upsert_node(&file("leaf", "a", "leaf.txt", 1)).unwrap(); + + assert!(!db.is_pinned(&du("leaf")).unwrap()); + + // And a pin *inside* the cycle is still found rather than walked past. + db.pin_add(&du("a"), "A", true).unwrap(); + assert!(db.is_pinned(&du("leaf")).unwrap()); +} + #[test] fn schema_objects_exist() { let db = Db::open_in_memory().unwrap(); diff --git a/crates/pdfs-core/src/db/utils.rs b/crates/pdfs-core/src/db/utils.rs index 15a6125..154e711 100644 --- a/crates/pdfs-core/src/db/utils.rs +++ b/crates/pdfs-core/src/db/utils.rs @@ -35,7 +35,7 @@ pub(super) fn like_escape(s: &str) -> String { /// nowhere near this; a parent *cycle* is unbounded, and this query is on the /// path of every search result, including the `LIKE` lane that reads `nodes` /// directly and so never passed through the index's cycle check. -const MAX_PATH_DEPTH: usize = 256; +pub(super) const MAX_PATH_DEPTH: usize = 256; /// Resolve a node's mountpoint-relative path by walking `parent_uid` to the /// root via a recursive CTE. The root (the node with no parent) is excluded, so diff --git a/crates/pdfs-core/src/error.rs b/crates/pdfs-core/src/error.rs index 5d9fa41..ad228dc 100644 --- a/crates/pdfs-core/src/error.rs +++ b/crates/pdfs-core/src/error.rs @@ -45,6 +45,27 @@ pub enum Error { Other(String), } +impl Error { + /// Whether this is the local disk being full (`ENOSPC`), as opposed to any + /// other I/O failure. + /// + /// Worth distinguishing because it is the one storage failure the user can + /// act on, and because the caller's response differs: a full disk is met by + /// evicting cached blobs and answering `ENOSPC`, where `EIO` would read as a + /// broken file. `std::io::ErrorKind::StorageFull` is still unstable, so this + /// compares the raw code. + pub fn is_disk_full(&self) -> bool { + match self { + Error::Io(e) => e.raw_os_error() == Some(libc::ENOSPC), + // SQLite reports it as its own code rather than surfacing the errno. + Error::Db(rusqlite::Error::SqliteFailure(e, _)) => { + e.code == rusqlite::ErrorCode::DiskFull + } + _ => false, + } + } +} + pub type Result = std::result::Result; /// A failure on its way to a front-end: prose for the user, plus the diff --git a/crates/pdfs-fuse/src/background.rs b/crates/pdfs-fuse/src/background.rs index bda02b1..8ffb773 100644 --- a/crates/pdfs-fuse/src/background.rs +++ b/crates/pdfs-fuse/src/background.rs @@ -8,35 +8,20 @@ use super::*; /// /// Returns whether this mount actually held the node, so the caller can evict /// its content blob once if *any* mount did. -fn forget_and_notify(st: &mut State, notifier: Option<&Notifier>, uid: &NodeUid) -> bool { +fn forget_and_notify(st: &mut State, notify: &mut NotifyBatch, uid: &NodeUid) -> bool { // Capture the inode before `forget` clears the uid mapping. let child = st.by_uid.get(uid).copied(); let Some((parent, name)) = st.forget(uid) else { return false; }; - if let Some(notifier) = notifier { - match child { - Some(child) => { - let _ = notifier.delete(INodeNo(parent), INodeNo(child), OsStr::new(&name)); - } - None => { - let _ = notifier.inval_entry(INodeNo(parent), OsStr::new(&name)); - } - } + match child { + Some(child) => notify.delete(parent, child, name), + None => notify.inval_entry(parent, name), } true } -fn invalidate_access_changes(notifier: Option<&Notifier>, changed: &[u64]) { - let Some(notifier) = notifier else { - return; - }; - for &ino in changed { - let _ = notifier.inval_inode(INodeNo(ino), 0, 0); - } -} - -fn hide_foreign_deleted_and_notify(st: &mut State, notifier: Option<&Notifier>, uid: &NodeUid) { +fn hide_foreign_deleted_and_notify(st: &mut State, notify: &mut NotifyBatch, uid: &NodeUid) { let child = st.by_uid.get(uid).copied(); let dentry = child.and_then(|ino| { st.entries @@ -45,9 +30,9 @@ fn hide_foreign_deleted_and_notify(st: &mut State, notifier: Option<&Notifier>, }); let changed = st.downgrade_shared_subtree(uid); st.hide_foreign_subtree(uid); - invalidate_access_changes(notifier, &changed); - if let (Some(notifier), Some((parent, name)), Some(child)) = (notifier, dentry, child) { - let _ = notifier.delete(INodeNo(parent), INodeNo(child), OsStr::new(&name)); + notify.extend_inodes(&changed); + if let (Some((parent, name)), Some(child)) = (dentry, child) { + notify.delete(parent, child, name); } } @@ -132,18 +117,18 @@ fn apply_event(core: &Core, event: &DriveEvent, dirty: &mut DirtyParents) -> pdf && foreign_delete { apply_foreign_delete(&core.db, node_uid, |uid| { - core.for_each_mount(|st, notifier| { - hide_foreign_deleted_and_notify(st, notifier, uid); + core.for_each_mount(|st, notify| { + hide_foreign_deleted_and_notify(st, notify, uid); }); }) } else { apply_access_downgrade(&core.db, event, |root| { - core.for_each_mount(|st, notifier| { + core.for_each_mount(|st, notify| { let changed = match root { Some(uid) => st.downgrade_shared_subtree(uid), None => st.downgrade_known_shared_access(), }; - invalidate_access_changes(notifier, &changed); + notify.extend_inodes(&changed); }); }) }; @@ -173,19 +158,17 @@ fn apply_event(core: &Core, event: &DriveEvent, dirty: &mut DirtyParents) -> pdf debug!(uid = %node_uid, "ignoring remote event for a node with a queued write"); } let mut had_node = false; - core.for_each_mount(|st, notifier| { + core.for_each_mount(|st, notify| { if *is_trashed { // Trashing makes a node vanish from its parent listing. - had_node |= forget_and_notify(st, notifier, node_uid); + had_node |= forget_and_notify(st, notify, node_uid); } else if !ours && let Some(&ino) = st.by_uid.get(node_uid) { // Known node changed: drop its cached attrs/data (and // listing if it is a directory) so the next access // re-fetches. Its content blob may now be stale too. had_node = true; st.invalidate_listing(ino); - if let Some(notifier) = notifier { - let _ = notifier.inval_inode(INodeNo(ino), 0, 0); - } + notify.inval_inode(ino); } }); // Once, not once per mount: the content cache is per-daemon. @@ -209,8 +192,8 @@ fn apply_event(core: &Core, event: &DriveEvent, dirty: &mut DirtyParents) -> pdf DriveEvent::NodeDeleted { node_uid, .. } => { core.cache.evict(node_uid); if !foreign_delete { - core.for_each_mount(|st, notifier| { - forget_and_notify(st, notifier, node_uid); + core.for_each_mount(|st, notify| { + forget_and_notify(st, notify, node_uid); }); } } @@ -223,13 +206,11 @@ fn apply_event(core: &Core, event: &DriveEvent, dirty: &mut DirtyParents) -> pdf | DriveEvent::SharedWithMeUpdated { .. } => { warn!("event access continuity lost; downgrading shared trees and resyncing lazily"); apply_result = apply_result.and(core.db.clear_state(SHARED_WITH_ME_SYNCED_MS)); - core.for_each_mount(|st, notifier| { + core.for_each_mount(|st, notify| { let dirs: Vec = st.children.keys().copied().collect(); for &ino in &dirs { st.invalidate_listing(ino); - if let Some(notifier) = notifier { - let _ = notifier.inval_inode(INodeNo(ino), 0, 0); - } + notify.inval_inode(ino); } }); } @@ -303,18 +284,16 @@ fn flush_dirty_parents(core: &Core, dirty: &DirtyParents) { if dirty.foreign.is_empty() && dirty.ours.is_empty() { return; } - core.for_each_mount(|st, notifier| { - let drop_listing = |st: &mut State, parent_uid: &NodeUid| { - let Some(&parent) = st.by_uid.get(parent_uid) else { - return; - }; - st.invalidate_listing(parent); - if let Some(notifier) = notifier { - let _ = notifier.inval_inode(INodeNo(parent), 0, 0); - } + fn drop_listing(st: &mut State, notify: &mut NotifyBatch, parent_uid: &NodeUid) { + let Some(&parent) = st.by_uid.get(parent_uid) else { + return; }; + st.invalidate_listing(parent); + notify.inval_inode(parent); + } + core.for_each_mount(|st, notify| { for parent_uid in &dirty.foreign { - drop_listing(st, parent_uid); + drop_listing(st, notify, parent_uid); } for (parent_uid, children) in &dirty.ours { if dirty.foreign.contains(parent_uid) { @@ -328,7 +307,7 @@ fn flush_dirty_parents(core: &Core, dirty: &DirtyParents) { { continue; } - drop_listing(st, parent_uid); + drop_listing(st, notify, parent_uid); } }); } diff --git a/crates/pdfs-fuse/src/control.rs b/crates/pdfs-fuse/src/control.rs index 224aa30..ec5d12e 100644 --- a/crates/pdfs-fuse/src/control.rs +++ b/crates/pdfs-fuse/src/control.rs @@ -29,7 +29,23 @@ use super::{Core, count_noun, human_bytes, human_duration, parse_uid}; const MAX_CONTROL_REQUEST_BYTES: u64 = 1024 * 1024; const CONTROL_READ_TIMEOUT: Duration = Duration::from_secs(10); +/// A client that connects, sends a request and then never reads its reply used +/// to park a handler thread forever in `write_all`; 64 of those exhaust +/// [`MAX_CONTROL_HANDLERS`] permanently. Generous enough for the largest +/// responses (photo bytes) over a local socket, finite so a wedged peer can't +/// hold a permit. +const CONTROL_WRITE_TIMEOUT: Duration = Duration::from_secs(30); const MAX_CONTROL_HANDLERS: usize = 64; +/// Ceiling on any client-supplied row `limit`. Every in-tree caller asks for far +/// less (200 for activity, 60 for a photo page), but the value goes straight +/// into `SQL LIMIT ?` under the daemon's single connection — an unclamped one +/// materialises the whole account, and then serialises it into one JSON line. +const MAX_ROW_LIMIT: usize = 1000; + +/// Clamp a client-supplied row limit to [`MAX_ROW_LIMIT`]. +fn clamp_limit(limit: usize) -> usize { + limit.min(MAX_ROW_LIMIT) +} static ACTIVE_CONTROL_HANDLERS: AtomicUsize = AtomicUsize::new(0); struct ControlHandlerPermit; @@ -109,6 +125,10 @@ fn handle_control_conn(core: &Core, username: &str, mountpoint: &Path, stream: U warn!(error = %e, "control: could not set request timeout"); return; } + if let Err(e) = stream.set_write_timeout(Some(CONTROL_WRITE_TIMEOUT)) { + warn!(error = %e, "control: could not set response timeout"); + return; + } let mut reader = BufReader::new(match stream.try_clone() { Ok(s) => s, Err(e) => { @@ -204,7 +224,7 @@ fn handle_control_conn(core: &Core, username: &str, mountpoint: &Path, stream: U kind, range, favorites, - }) => match core.photos_timeline(offset, limit, kind, range, favorites) { + }) => match core.photos_timeline(offset, clamp_limit(limit), kind, range, favorites) { Ok(Some(items)) => CtlResponse::Photos { available: true, items, @@ -245,7 +265,7 @@ fn handle_control_conn(core: &Core, username: &str, mountpoint: &Path, stream: U // An album page is a timeline page as far as the front-end is // concerned, so it comes back in the same reply — with `counts` // unset, which describe the whole timeline rather than this album. - Some(album) => match core.album_photos(&album, offset, limit) { + Some(album) => match core.album_photos(&album, offset, clamp_limit(limit)) { Ok(items) => CtlResponse::Photos { available: true, items, @@ -346,22 +366,24 @@ fn handle_control_conn(core: &Core, username: &str, mountpoint: &Path, stream: U Err(e) => CtlResponse::error(e), } } - Ok(CtlRequest::Search { query, limit }) => match core.search(&query, limit) { + Ok(CtlRequest::Search { query, limit }) => match core.search(&query, clamp_limit(limit)) { Ok(hits) => CtlResponse::SearchResults { hits }, Err(e) => CtlResponse::error(e), }, - Ok(CtlRequest::SearchLocal { query, limit }) => match core.search_local(&query, limit) { - Ok(hits) => CtlResponse::LocalResults { - hits, - indexing: core.indexing.load(Ordering::Relaxed), - }, - Err(e) => CtlResponse::error(e), - }, + Ok(CtlRequest::SearchLocal { query, limit }) => { + match core.search_local(&query, clamp_limit(limit)) { + Ok(hits) => CtlResponse::LocalResults { + hits, + indexing: core.indexing.load(Ordering::Relaxed), + }, + Err(e) => CtlResponse::error(e), + } + } Ok(CtlRequest::SearchV2 { query, limit, filters, - }) => match core.search_v2(&query, limit, &filters) { + }) => match core.search_v2(&query, clamp_limit(limit), &filters) { Ok((drive_hits, local_hits)) => CtlResponse::SearchResultsV2 { drive_hits, local_hits, @@ -1023,7 +1045,7 @@ fn handle_control_conn(core: &Core, username: &str, mountpoint: &Path, stream: U Err(e) => CtlResponse::error(e), }, Ok(CtlRequest::ListActivity { limit }) => CtlResponse::Activity { - items: core.list_activity(limit), + items: core.list_activity(clamp_limit(limit)), }, // The request did not parse: the caller sent something malformed. Err(e) => CtlResponse::error(CoreError::invalid(format!("bad request: {e}"))), diff --git a/crates/pdfs-fuse/src/devices.rs b/crates/pdfs-fuse/src/devices.rs index 8525830..48c0ca9 100644 --- a/crates/pdfs-fuse/src/devices.rs +++ b/crates/pdfs-fuse/src/devices.rs @@ -536,7 +536,14 @@ impl Core { // folder the daemon no longer tracks, and nothing would ever unmount it. // Tear it down first — including before trashing the remote tree it // serves, which would otherwise leave it answering for deleted nodes. - if let Some(mount) = self.mounts.lock().remove(&id) + // Taken in its own statement so the registry lock is released before the + // teardown: an edition-2024 `if let` chain holds the guard for the whole + // body, and `teardown` unmounts and joins the session thread — which can + // be inside a multi-second transfer. The shutdown path takes the same + // lock, so holding it here risks systemd's stop timeout landing mid- + // unmount and stranding the endpoint. + let taken = self.mounts.lock().remove(&id); + if let Some(mount) = taken && let Err(error) = commit_after_teardown(mount.teardown(), || ()) { let _ = self.db.sync_folder_set_state(id, "error", now_secs()); @@ -866,8 +873,11 @@ impl Core { Ok(format!("{} is now on-demand", local.display())) } ApplicableMountMode::Mirror => { - // ondemand→mirror: tear down the secondary mount first. - if let Some(mount) = self.mounts.lock().remove(&id) + // ondemand→mirror: tear down the secondary mount first. Taken + // before the teardown so the registry lock is not held across + // the unmount and thread join (see `remove_sync_folder`). + let taken = self.mounts.lock().remove(&id); + if let Some(mount) = taken && let Err(error) = commit_after_teardown(mount.teardown(), || ()) { let _ = self.db.sync_folder_set_state(id, "error", now_secs()); diff --git a/crates/pdfs-fuse/src/drain.rs b/crates/pdfs-fuse/src/drain.rs index 2f6337f..3a9d3e1 100644 --- a/crates/pdfs-fuse/src/drain.rs +++ b/crates/pdfs-fuse/src/drain.rs @@ -136,6 +136,10 @@ impl Core { let due = self.db.next_due_op(now).unwrap_or_default(); let Some(op) = due.filter(|_| self.online.load(Ordering::Relaxed)) else { + // Nothing to upload is exactly when the connection is free, so + // this is where the read path's buffered LRU touches get written + // (`ContentCache::flush_touches`). Cheap and a no-op when empty. + self.cache.flush_touches(); self.recover_fsynced_writes(); // A debounced or backed-off op may be waiting: sleep only // until it becomes due rather than the full idle-poll. @@ -709,8 +713,9 @@ impl Core { self.db.delete_op(op.id)?; // Dropping the pending entry hands the node back to the remote's truth: // reads stop coming from the staged blob, and the event sync stops - // skipping it as "ahead of the server" (offline.md Phase 3a). - self.pending.lock().remove(uid); + // skipping it as "ahead of the server" (offline.md Phase 3a). Only if it + // is still *this* write's entry — see [`Core::release_pending`]. + self.release_pending(uid, blob); self.cache.discard_staged(blob); self.cache.evict(uid); self.evict_reader(uid); @@ -748,7 +753,7 @@ impl Core { error!(%uid, name, reason, staged = %blob.display(), "cannot place a conflicted write; bytes kept in staging"); self.db.delete_op(op.id)?; - self.pending.lock().remove(uid); + self.release_pending(uid, blob); self.cache.evict(uid); self.evict_reader(uid); self.log_activity( @@ -760,6 +765,24 @@ impl Core { Ok(()) } + /// Drop `uid`'s pending entry, but only while it still names `blob`. + /// + /// A supersede can replace the entry while the op that owned it is in + /// flight, so an unconditional `remove` retires a *newer* write's staged + /// blob: reads fall back to a stale remote, the next `remote_baseline` forks + /// a spurious `(sync-conflict)` copy of bytes that were never in conflict, + /// and the incomplete-write interlock — which keys off the pending entry — + /// is disarmed, letting a gap-fill read from a revision that no longer + /// describes the file. + pub(crate) fn release_pending(&self, uid: &NodeUid, blob: &Path) { + let mut pending = self.pending.lock(); + if let Some(current) = pending.get(uid) + && current.path == blob + { + pending.remove(uid); + } + } + /// Whether a folder a queued op targets has stopped being a place a node can /// go. A failure to ask is not an answer: only a definite "trashed" or "not /// there" counts, so a network fault leaves the op to retry normally. @@ -886,14 +909,7 @@ impl Core { // an orphaned file (harmless), whereas the reverse would leave a queued // op pointing at nothing. self.db.delete_op(op.id)?; - { - let mut p = self.pending.lock(); - if let Some(op_in_mem) = p.get(&uid) - && op_in_mem.path == blob - { - p.remove(&uid); - } - } + self.release_pending(&uid, &blob); // The staged blob now matches the sealed revision, so a pinned file keeps // it as its cached content rather than re-downloading what we just sent. diff --git a/crates/pdfs-fuse/src/filesystem.rs b/crates/pdfs-fuse/src/filesystem.rs index 7358510..34f0ee7 100644 --- a/crates/pdfs-fuse/src/filesystem.rs +++ b/crates/pdfs-fuse/src/filesystem.rs @@ -3,6 +3,73 @@ use super::*; impl Filesystem for ProtonFs { + /// Negotiate kernel-side behaviour once, before any other callback. + /// + /// Everything here is a hint the kernel is free to decline, so each is + /// requested and its refusal logged rather than treated as fatal — an older + /// kernel that lacks a capability must still mount. + /// + /// Deliberately *not* requested: `FUSE_ATOMIC_O_TRUNC` and + /// `FUSE_DO_READDIRPLUS` change what the kernel sends (truncation folded + /// into `open`, `readdirplus` instead of `readdir` + N lookups), so they + /// need the matching handlers first; and `FOPEN_KEEP_CACHE` needs the write + /// path to invalidate page cache on every locally staged revision, not just + /// on remote events. + fn init(&mut self, _req: &Request, config: &mut KernelConfig) -> std::io::Result<()> { + // Readahead the kernel issues per file. Matching the block size means a + // sequential reader's readahead lands inside the block a demand read is + // already fetching, instead of one 128 KiB slice at a time. + if let Err(nearest) = config.set_max_readahead(BLOCK_SIZE as u32) { + debug!(nearest, "kernel capped max_readahead"); + let _ = config.set_max_readahead(nearest); + } + if let Err(nearest) = config.set_max_background(MAX_BACKGROUND) { + debug!(nearest, "kernel capped max_background"); + } + // Without this the kernel serialises lookups within one directory, so a + // cold `ls` of a folder is a chain of round trips rather than a fan-out. + if let Err(unsupported) = config.add_capabilities(InitFlags::FUSE_PARALLEL_DIROPS) { + debug!(?unsupported, "kernel does not support parallel dirops"); + } + if let Err(unsupported) = config.add_capabilities(InitFlags::FUSE_CACHE_SYMLINKS) { + debug!(?unsupported, "kernel does not support symlink caching"); + } + Ok(()) + } + + /// Free and total space, as `df` and every "will this fit?" preflight ask. + /// + /// Reported in [`REPORTED_BLKSIZE`] units against the *account* quota, not + /// the local cache: what a copy into the mount consumes is Drive storage. + /// The default implementation answers all-zeroes, which reads as a full + /// filesystem and makes installers and file managers refuse to write. + /// + /// On a worker, because a cold answer is a round trip + /// ([`Core::account_quota_cached`] serves a warm one from memory). + fn statfs(&self, _req: &Request, _ino: INodeNo, reply: ReplyStatfs) { + let core = self.core.clone(); + self.core.workers.run(Lane::Meta, move || { + let (max_space, used_space) = core.account_quota_cached().unwrap_or((0, 0)); + let bsize = u64::from(REPORTED_BLKSIZE); + let blocks = (max_space.max(0) as u64).div_ceil(bsize); + let used = (used_space.max(0) as u64).div_ceil(bsize); + let free = blocks.saturating_sub(used); + // `files`/`ffree` are unknowable — Drive has no inode budget — so + // they are reported as "plenty" rather than as zero, which some + // tools read as "no more files may be created". + reply.statfs( + blocks, + free, + free, + 0, + u64::MAX, + REPORTED_BLKSIZE, + MAX_NAME_LEN, + REPORTED_BLKSIZE, + ); + }); + } + fn lookup(&self, _req: &Request, parent: INodeNo, name: &OsStr, reply: ReplyEntry) { let parent = parent.0; let name = match fuse_name(name) { @@ -109,109 +176,19 @@ impl Filesystem for ProtonFs { .run(Lane::Meta, move || fs.serve_readdir(ino, offset, reply)); } + /// A read-only open is a map lookup and answers inline; a write open has to + /// allocate a scratch file and, when a queued revision is the base, duplicate + /// a blob that is the whole file — so it is served from a worker rather than + /// stalling every other request on the mount behind it. fn open(&self, _req: &Request, ino: INodeNo, flags: OpenFlags, reply: ReplyOpen) { - let write_requested = flags.acc_mode() != OpenAccMode::O_RDONLY; - let (uid, base_mtime, base_size, base_revision_id) = { - let mut st = self.core.state.lock(); - match st.entries.get_mut(&ino.0) { - Some(e) if e.node.is_file() => { - if write_requested && !e.writable() { - reply.error(Errno::EACCES); - return; - } - e.open_count = e.open_count.saturating_add(1); - ( - e.uid.clone(), - e.node.modification_time, - node_size(&e.node), - node_revision_id(&e.node), - ) - } - Some(_) => { - reply.error(Errno::EISDIR); - return; - } - None => { - reply.error(Errno::ENOENT); - return; - } - } - }; - debug!(ino = ino.0, ?flags, base_size, base_mtime, "open"); - // Read-only opens stay stateless (fh 0). A write open allocates a - // disk-backed handle; bytes are authored into a scratch file and the - // untouched remainder is pulled from the base lazily (on read / commit). if flags.acc_mode() == OpenAccMode::O_RDONLY { - reply.opened(FileHandle(0), FopenFlags::empty()); + self.serve_open(ino, flags, reply); return; } - // A previous close may already have newer bytes queued locally while - // the server still holds the older revision. That staged blob, not the - // remote, is the base for this handle. In particular, `truncate file` - // opens for writing and then shrinks via setattr; starting its scratch - // as a sparse zero file would turn the preserved prefix into zeros. - // - // Complete pending blobs are the common case and can be copied without - // any network access. An incomplete blob still has gaps referring to - // the remote base; stacking another write on it cannot be represented - // safely by WriteHandle, so fail the open rather than risk corruption. - let pending_base = self.core.pending.lock().get(&uid).cloned(); - if pending_base.as_ref().is_some_and(|p| !p.meta.complete) { - if let Some(entry) = self.core.state.lock().entries.get_mut(&ino.0) { - entry.open_count = entry.open_count.saturating_sub(1); - } - error!(%uid, "refusing write over incomplete queued revision"); - reply.error(Errno::EIO); - return; - } - let (file, path) = match self.core.cache.create_scratch() { - Ok(x) => x, - Err(e) => { - if let Some(entry) = self.core.state.lock().entries.get_mut(&ino.0) { - entry.open_count = entry.open_count.saturating_sub(1); - } - error!(%uid, error = %e, "create scratch file failed"); - reply.error(Errno::EIO); - return; - } - }; - let mut initial_written = Intervals::default(); - if let Some(pending) = &pending_base { - if let Err(e) = std::fs::copy(&pending.path, &path) { - if let Some(entry) = self.core.state.lock().entries.get_mut(&ino.0) { - entry.open_count = entry.open_count.saturating_sub(1); - } - let _ = std::fs::remove_file(&path); - error!(%uid, source = %pending.path.display(), error = %e, - "copy queued revision into write scratch failed"); - reply.error(Errno::EIO); - return; - } - initial_written.add(0, pending.meta.len); - } - let mut st = self.core.state.lock(); - let fh = st.next_fh; - st.next_fh += 1; - let aw = st.active_writes.entry(ino.0).or_insert_with(|| { - WriteHandle { - ino: ino.0, - uid, - file: Arc::new(file), - path, - written: initial_written, - // Starts at the current size; reads in [0, base_size) come from - // the base until overwritten. - len: base_size, - base_size, - base_mtime, - base_revision_id, - dirty: false, - open_count: 0, - } - }); - aw.open_count += 1; - st.handles.insert(fh, ino.0); - reply.opened(FileHandle(fh), FopenFlags::empty()); + let fs = self.clone(); + self.core + .workers + .run(Lane::Transfer, move || fs.serve_open(ino, flags, reply)); } fn read( @@ -374,7 +351,16 @@ impl Filesystem for ProtonFs { }; if let Err(e) = file.write_all_at(data, offset) { error!(ino = ino.0, fh, error = %e, "scratch write failed"); - reply.error(Errno::EIO); + // A full local disk is not a broken file: answer `ENOSPC` so the + // application can say so (and so `cp`/editors take their own + // out-of-space path), and free what the cache is holding — the + // scratch file the next write goes to shares that filesystem. + if e.raw_os_error() == Some(libc::ENOSPC) { + self.core.cache.emergency_evict(); + reply.error(Errno::ENOSPC); + } else { + reply.error(Errno::EIO); + } return; } let new_len = { @@ -566,6 +552,12 @@ impl Filesystem for ProtonFs { /// for fsync wants its bytes to survive a crash, and blocking it on an upload /// (which offline would never finish) buys nothing the queue does not already /// guarantee. + /// + /// Served on a worker: `sync_all` is a disk barrier and the sidecar write is + /// two more file operations, and an application that fsyncs per record would + /// otherwise stall every request on the mount at its barrier rate. Nothing + /// orders this against other callbacks — the kernel waits for the reply + /// either way, which is the whole contract. fn fsync( &self, _req: &Request, @@ -574,76 +566,10 @@ impl Filesystem for ProtonFs { _datasync: bool, reply: ReplyEmpty, ) { - // Everything the durability sidecar needs, copied out so no lock is held - // across the sync. - let handle = { - let st = self.core.state.lock(); - let ino = st.handles.get(&fh.0).copied(); - ino.and_then(|i| st.active_writes.get(&i)).map(|aw| { - ( - aw.file.clone(), - aw.path.clone(), - aw.uid.clone(), - aw.dirty, - aw.written.clone(), - aw.len, - aw.base_size, - aw.base_mtime, - aw.base_revision_id.clone(), - ) - }) - }; - let Some((f, path, uid, dirty, written, len, base_size, base_mtime, base_revision_id)) = - handle - else { - reply.error(Errno::EBADF); - return; - }; - if let Err(e) = f.sync_all() { - error!(fh = fh.0, error = %e, "fsync of scratch file failed"); - reply.error(Errno::EIO); - return; - } - // A clean handle has nothing on it that the remote does not already - // hold, so there is nothing for a crash to lose. - if !dirty { - reply.ok(); - return; - } - // The bytes are on stable storage, but only the scratch file knows that, - // and the scratch directory is cleared at open. Record what the blob is - // so a restart can hand it to the upload queue instead of deleting it. - // - // Deliberately *not* the full release path: staging the blob here would - // move the file out from under a handle the application still has open, - // and queueing an upload per `fsync` would push a revision for every - // barrier in a write loop. This only makes the existing file findable. - let authored: Vec<(u64, u64)> = written - .segments(0, len) - .into_iter() - .filter(|&(_, _, authored)| authored) - .map(|(s, e, _)| (s, e)) - .collect(); - let meta = StagedWrite { - uid: uid.to_string(), - len, - base_size, - base_mtime, - complete: authored == [(0, len)], - authored, - based_on: self - .core - .remote_baseline(&uid, base_mtime, base_size, base_revision_id), - }; - // A failed sidecar means the write is not durable, and `fsync` promising - // otherwise is the defect this exists to fix — so it is an error, not a - // warning. - if let Err(e) = self.core.cache.mark_scratch_durable(&path, &meta) { - error!(%uid, error = %e, "recording fsync durability failed"); - reply.error(Errno::EIO); - return; - } - reply.ok(); + let fs = self.clone(); + self.core + .workers + .run(Lane::Transfer, move || fs.serve_fsync(fh, reply)); } fn release( @@ -1146,7 +1072,7 @@ impl ProtonFs { uid: self.uid, gid: self.gid, rdev: 0, - blksize: 512, + blksize: REPORTED_BLKSIZE, flags: 0, } } @@ -1154,6 +1080,202 @@ impl ProtonFs { /// Trash the child `name` under `parent` on the remote, then drop it from the /// local cache. Backs both `unlink` and `rmdir` (Proton trashes whole /// subtrees, so an `rmdir` of a non-empty dir behaves the same). + /// The body of [`Filesystem::open`], run off the dispatch loop for a write + /// open. + fn serve_open(&self, ino: INodeNo, flags: OpenFlags, reply: ReplyOpen) { + let write_requested = flags.acc_mode() != OpenAccMode::O_RDONLY; + let (uid, base_mtime, base_size, base_revision_id) = { + let mut st = self.core.state.lock(); + match st.entries.get_mut(&ino.0) { + Some(e) if e.node.is_file() => { + if write_requested && !e.writable() { + reply.error(Errno::EACCES); + return; + } + e.open_count = e.open_count.saturating_add(1); + ( + e.uid.clone(), + e.node.modification_time, + node_size(&e.node), + node_revision_id(&e.node), + ) + } + Some(_) => { + reply.error(Errno::EISDIR); + return; + } + None => { + reply.error(Errno::ENOENT); + return; + } + } + }; + debug!(ino = ino.0, ?flags, base_size, base_mtime, "open"); + // Read-only opens stay stateless (fh 0). A write open allocates a + // disk-backed handle; bytes are authored into a scratch file and the + // untouched remainder is pulled from the base lazily (on read / commit). + if flags.acc_mode() == OpenAccMode::O_RDONLY { + reply.opened(FileHandle(0), FopenFlags::empty()); + return; + } + // A previous close may already have newer bytes queued locally while + // the server still holds the older revision. That staged blob, not the + // remote, is the base for this handle. In particular, `truncate file` + // opens for writing and then shrinks via setattr; starting its scratch + // as a sparse zero file would turn the preserved prefix into zeros. + // + // Complete pending blobs are the common case and can be copied without + // any network access. An incomplete blob still has gaps referring to + // the remote base; stacking another write on it cannot be represented + // safely by WriteHandle, so fail the open rather than risk corruption. + let pending_base = self.core.pending.lock().get(&uid).cloned(); + if pending_base.as_ref().is_some_and(|p| !p.meta.complete) { + if let Some(entry) = self.core.state.lock().entries.get_mut(&ino.0) { + entry.open_count = entry.open_count.saturating_sub(1); + } + error!(%uid, "refusing write over incomplete queued revision"); + reply.error(Errno::EIO); + return; + } + let (file, path) = match self.core.cache.create_scratch() { + Ok(x) => x, + Err(e) => { + if let Some(entry) = self.core.state.lock().entries.get_mut(&ino.0) { + entry.open_count = entry.open_count.saturating_sub(1); + } + error!(%uid, error = %e, "create scratch file failed"); + reply.error(Errno::EIO); + return; + } + }; + let mut initial_written = Intervals::default(); + if let Some(pending) = &pending_base { + // Reflink where the filesystem can: this blob is the whole file, and + // a byte-for-byte copy of a multi-GB pending revision is the kind of + // work `open` must not do (`ContentCache::clone_or_copy`). + if let Err(e) = ContentCache::clone_or_copy(&pending.path, &path) { + if let Some(entry) = self.core.state.lock().entries.get_mut(&ino.0) { + entry.open_count = entry.open_count.saturating_sub(1); + } + let _ = std::fs::remove_file(&path); + error!(%uid, source = %pending.path.display(), error = %e, + "copy queued revision into write scratch failed"); + reply.error(Errno::EIO); + return; + } + initial_written.add(0, pending.meta.len); + } + let mut st = self.core.state.lock(); + let fh = st.next_fh; + st.next_fh += 1; + let aw = st.active_writes.entry(ino.0).or_insert_with(|| { + WriteHandle { + ino: ino.0, + uid, + file: Arc::new(file), + path, + written: initial_written, + // Starts at the current size; reads in [0, base_size) come from + // the base until overwritten. + len: base_size, + base_size, + base_mtime, + base_revision_id, + dirty: false, + open_count: 0, + } + }); + aw.open_count += 1; + st.handles.insert(fh, ino.0); + reply.opened(FileHandle(fh), FopenFlags::empty()); + } + + /// The body of [`Filesystem::fsync`], run off the dispatch loop. + fn serve_fsync(&self, fh: FileHandle, reply: ReplyEmpty) { + // Everything the durability sidecar needs, copied out so no lock is held + // across the sync. + let handle = { + let st = self.core.state.lock(); + let ino = st.handles.get(&fh.0).copied(); + ino.and_then(|i| st.active_writes.get(&i)).map(|aw| { + ( + aw.file.clone(), + aw.path.clone(), + aw.uid.clone(), + aw.dirty, + aw.written.clone(), + aw.len, + aw.base_size, + aw.base_mtime, + aw.base_revision_id.clone(), + ) + }) + }; + let Some((f, path, uid, dirty, written, len, base_size, base_mtime, base_revision_id)) = + handle + else { + reply.error(Errno::EBADF); + return; + }; + if let Err(e) = f.sync_all() { + error!(fh = fh.0, error = %e, "fsync of scratch file failed"); + // Delayed allocation means a full disk usually surfaces here rather + // than at `write(2)`; `ENOSPC` is what an application checks for. + if e.raw_os_error() == Some(libc::ENOSPC) { + self.core.cache.emergency_evict(); + reply.error(Errno::ENOSPC); + } else { + reply.error(Errno::EIO); + } + return; + } + // A clean handle has nothing on it that the remote does not already + // hold, so there is nothing for a crash to lose. + if !dirty { + reply.ok(); + return; + } + // The bytes are on stable storage, but only the scratch file knows that, + // and the scratch directory is cleared at open. Record what the blob is + // so a restart can hand it to the upload queue instead of deleting it. + // + // Deliberately *not* the full release path: staging the blob here would + // move the file out from under a handle the application still has open, + // and queueing an upload per `fsync` would push a revision for every + // barrier in a write loop. This only makes the existing file findable. + let authored: Vec<(u64, u64)> = written + .segments(0, len) + .into_iter() + .filter(|&(_, _, authored)| authored) + .map(|(s, e, _)| (s, e)) + .collect(); + let meta = StagedWrite { + uid: uid.to_string(), + len, + base_size, + base_mtime, + complete: authored == [(0, len)], + authored, + based_on: self + .core + .remote_baseline(&uid, base_mtime, base_size, base_revision_id), + }; + // A failed sidecar means the write is not durable, and `fsync` promising + // otherwise is the defect this exists to fix — so it is an error, not a + // warning. + if let Err(e) = self.core.cache.mark_scratch_durable(&path, &meta) { + error!(%uid, error = %e, "recording fsync durability failed"); + if e.is_disk_full() { + self.core.cache.emergency_evict(); + reply.error(Errno::ENOSPC); + } else { + reply.error(Errno::EIO); + } + return; + } + reply.ok(); + } + /// The body of [`Filesystem::create`], run off the dispatch loop. fn serve_create(&self, parent: u64, name: &str, reply: ReplyCreate) { if let Err(error) = self.core.require_writable(parent) { diff --git a/crates/pdfs-fuse/src/lib.rs b/crates/pdfs-fuse/src/lib.rs index 7411fa7..430a3b8 100644 --- a/crates/pdfs-fuse/src/lib.rs +++ b/crates/pdfs-fuse/src/lib.rs @@ -48,10 +48,10 @@ use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; use fuser::ReplyXattr; use fuser::{ AccessFlags, BackgroundSession, BsdFileFlags, Config, CopyFileRangeFlags, Errno, FileAttr, - FileHandle, FileType, Filesystem, FopenFlags, Generation, INodeNo, IoctlFlags, LockOwner, - MountOption, Notifier, OpenAccMode, OpenFlags, RenameFlags, ReplyAttr, ReplyCreate, ReplyData, - ReplyDirectory, ReplyEmpty, ReplyEntry, ReplyIoctl, ReplyLseek, ReplyOpen, ReplyWrite, Request, - Session, TimeOrNow, WriteFlags, + FileHandle, FileType, Filesystem, FopenFlags, Generation, INodeNo, InitFlags, IoctlFlags, + KernelConfig, LockOwner, MountOption, Notifier, OpenAccMode, OpenFlags, RenameFlags, ReplyAttr, + ReplyCreate, ReplyData, ReplyDirectory, ReplyEmpty, ReplyEntry, ReplyIoctl, ReplyLseek, + ReplyOpen, ReplyStatfs, ReplyWrite, Request, Session, TimeOrNow, WriteFlags, }; use futures::StreamExt as _; use pdfs_core::batch; @@ -121,6 +121,27 @@ use workers::{FUSE_WORKERS, Lane, Workers}; /// change this is how long the kernel may serve stale metadata. const TTL: Duration = Duration::from_secs(30); +/// How long a `statfs(2)` answer is served from [`Core::quota`] before the +/// account figures are refetched. See [`Core::account_quota_cached`]. +const QUOTA_TTL: Duration = Duration::from_secs(60); + +/// Block size reported to the kernel and to `statfs(2)`. +/// +/// coreutils and most copy tools size their I/O buffers from `st_blksize`, and +/// the fuser default of 512 makes them issue reads two orders of magnitude +/// smaller than the [`BLOCK_SIZE`] this filesystem actually fetches in. 1 MiB is +/// the largest value that stays well inside the kernel's per-request limit. +const REPORTED_BLKSIZE: u32 = 1 << 20; + +/// Ceiling on in-flight background requests (readahead, writeback). The kernel +/// default of 16 caps how much of a sequential read the kernel will run ahead of +/// the application, which on a high-latency link is the whole ballgame. +const MAX_BACKGROUND: u16 = 64; + +/// Longest single path component, as reported by `statfs(2)`. The kernel's own +/// dirent limit; Proton's is not lower. +const MAX_NAME_LEN: u32 = 255; + /// How often the background task polls the remote event cursor. const POLL_INTERVAL: Duration = Duration::from_secs(10); /// First and longest delay between probes for the network coming back after an @@ -396,6 +417,14 @@ struct Core { /// is the validity tag — a new revision may well have a thumbnail — matching /// how [`ContentCache::read_thumbnail`] validates the positive side. no_thumbnail: Arc>>, + /// Last account quota answer and when it was learned, for `statfs(2)`. + /// + /// `df` and the free-space preflight of every file manager and installer ask + /// for this, and the answer costs a round trip — so it is served from here + /// for [`QUOTA_TTL`] and refreshed on a worker afterwards. Total account + /// storage moves slowly enough that a minute-old number is a better answer + /// than either a stall or the zeroes the default implementation returns. + quota: Arc>>, /// Folders whose listing was enumerated cheaply and is having its file sizes /// filled in right now, so a burst of `stat`s over a fresh listing starts one /// upgrade rather than one per entry. See [`Core::spawn_size_upgrade`]. @@ -701,11 +730,18 @@ impl Core { /// notification channel, so work that invalidates cached metadata reaches /// the session that actually minted the inodes it is naming. /// - /// `None` for a mount whose session has not been spawned yet — there is - /// nothing to invalidate in a kernel that has never been told about it. - fn for_each_mount(&self, mut apply: impl FnMut(&mut State, Option<&Notifier>)) { + /// The closure records what the kernel needs told into a [`NotifyBatch`] + /// rather than sending it: notifications are flushed after this mount's + /// State lock is released, the same rule [`Core::for_each_state`] and + /// [`Core::flush_access_changes`] follow. + fn for_each_mount(&self, mut apply: impl FnMut(&mut State, &mut NotifyBatch)) { for (_, state, notifier, _) in self.states.live() { - apply(&mut state.lock(), notifier.get()); + let mut batch = NotifyBatch::default(); + { + let mut state = state.lock(); + apply(&mut state, &mut batch); + } + batch.flush(notifier.get()); } } @@ -763,6 +799,75 @@ fn require_uid_access(db: &Db, uid: &NodeUid, live_access: &[Access]) -> Result< .ok_or(Errno::EACCES) } +/// One kernel notification, recorded while the State lock is held so it can be +/// sent once it is not. +/// +/// `inval_inode` calls into `invalidate_inode_pages2`, which can block waiting +/// on pages that this daemon's own workers have to fill — and those workers take +/// the State lock. Sending from inside the lock is that deadlock's shape; a +/// batch is the smallest way to keep the inode/mount pairing while moving the +/// send outside it. +enum KernelNotice { + InvalInode(u64), + InvalEntry { + parent: u64, + name: String, + }, + Delete { + parent: u64, + child: u64, + name: String, + }, +} + +/// Notifications accumulated under one mount's State lock. Flushed by +/// [`Core::for_each_mount`] against that mount's own channel — inode numbers are +/// per-mount, so the pairing has to survive the deferral. +#[derive(Default)] +struct NotifyBatch(Vec); + +impl NotifyBatch { + fn inval_inode(&mut self, ino: u64) { + self.0.push(KernelNotice::InvalInode(ino)); + } + + fn inval_entry(&mut self, parent: u64, name: String) { + self.0.push(KernelNotice::InvalEntry { parent, name }); + } + + fn delete(&mut self, parent: u64, child: u64, name: String) { + self.0.push(KernelNotice::Delete { + parent, + child, + name, + }); + } + + fn extend_inodes(&mut self, inodes: &[u64]) { + self.0 + .extend(inodes.iter().map(|&ino| KernelNotice::InvalInode(ino))); + } + + /// Send everything recorded. A mount whose session has not been spawned has + /// nothing to tell the kernel, so the batch is simply dropped. + fn flush(self, notifier: Option<&Notifier>) { + let Some(notifier) = notifier else { return }; + for notice in self.0 { + let _ = match notice { + KernelNotice::InvalInode(ino) => notifier.inval_inode(INodeNo(ino), 0, 0), + KernelNotice::InvalEntry { parent, name } => { + notifier.inval_entry(INodeNo(parent), OsStr::new(&name)) + } + KernelNotice::Delete { + parent, + child, + name, + } => notifier.delete(INodeNo(parent), INodeNo(child), OsStr::new(&name)), + }; + } + } +} + fn notify_access_changes(notifier: Option<&Notifier>, changed: &[u64]) { let Some(notifier) = notifier else { return }; for &ino in changed { @@ -2256,7 +2361,23 @@ impl Core { } let path = self.cache.stage_write(&meta, src).map_err(|e| { error!(%uid, error = %e, "staging write failed"); - Errno::EIO + // The scratch file is still the only copy of these bytes, and the + // caller is releasing the handle it belongs to — without a + // durability sidecar `discard_unmarked_scratch` deletes it at the + // next mount, the one place the "never delete on a failure path" + // invariant would break. Mark it first, whatever went wrong. + if let Err(mark) = self.cache.mark_scratch_durable(src, &meta) { + error!(%uid, error = %mark, + "could not mark an unstageable write durable; bytes at risk"); + } + if e.is_disk_full() { + // Free cached blobs so the retry has somewhere to land, and tell + // the caller what is actually wrong. + self.cache.emergency_evict(); + Errno::ENOSPC + } else { + Errno::EIO + } })?; let meta_json = serde_json::to_string(&meta).unwrap_or_default(); let superseded = if is_local_uid(uid) { @@ -2310,15 +2431,29 @@ impl Core { // Reflect the write in the tree straight away: `ls` must show the new // size and mtime even though the remote still holds the old revision. let now = now_secs(); - { + let recorded = { let mut st = self.state.lock(); - st.record_pending_write(ino, len, now); - } + // The only record that the file is as long as the caller was told. + // If it cannot be written the write did not fully take, and saying + // so beats acknowledging a `close(2)` and then serving the old size + // — the bytes themselves are already staged and queued either way. + st.record_pending_write(ino, len, now) + }; // Cached blobs and open readers describe the superseded revision. Reads // come from the staged blob until the op drains, so just drop them. + // Done even when the row above failed: the staged blob is the file's + // content now either way, and leaving a stale one cached would serve it. self.cache.evict(uid); self.evict_reader(uid); self.wake_drain(); + if let Err(e) = recorded { + error!(%uid, error = %e, "recording a queued write's size failed"); + return Err(if e.is_disk_full() { + Errno::ENOSPC + } else { + Errno::EIO + }); + } Ok(()) } diff --git a/crates/pdfs-fuse/src/mount.rs b/crates/pdfs-fuse/src/mount.rs index f576588..ea35d90 100644 --- a/crates/pdfs-fuse/src/mount.rs +++ b/crates/pdfs-fuse/src/mount.rs @@ -255,6 +255,14 @@ pub(super) fn spawn_session( root: Node, ) -> std::io::Result { let mut config = Config::default(); + // `n_threads` stays at fuser's default of 1 (with `clone_fd` off) on + // purpose, for now. A concurrent dispatch loop is worth having, but + // `release` currently relies on being serialised against the `open`+`read` + // an application issues right after `close(2)` — see the comment on + // `Filesystem::release`. Raising it before that ordering is made explicit + // (a per-node staging-in-flight barrier) reintroduces the acceptance + // suite's "concurrent file 1 mismatch". Kernel-side caps that do *not* + // depend on that ordering are negotiated in `ProtonFs::init`. config.mount_options = vec![ MountOption::FSName("protondrive".to_string()), MountOption::Subtype("protondrive".to_string()), @@ -385,6 +393,7 @@ pub fn mount( self_changes: Arc::new(Mutex::new(HashMap::new())), thumb_gen: Arc::new(Mutex::new(HashSet::new())), no_thumbnail: Arc::new(Mutex::new(HashMap::new())), + quota: Arc::new(Mutex::new(None)), size_upgrades: Arc::new(Mutex::new(HashMap::new())), notifier: Arc::new(OnceLock::new()), session_live: Arc::new(AtomicBool::new(false)), diff --git a/crates/pdfs-fuse/src/sharing.rs b/crates/pdfs-fuse/src/sharing.rs index a385199..76304b6 100644 --- a/crates/pdfs-fuse/src/sharing.rs +++ b/crates/pdfs-fuse/src/sharing.rs @@ -12,6 +12,9 @@ use pdfs_core::control::{ BookmarkInfo, DirEntry, InvitationInfo, PublicLinkInfo, ShareEntry, ShareEntryKind, SharedItem, }; use pdfs_core::{CoreError, CoreResult}; +use tracing::debug; + +use crate::QUOTA_TTL; use proton_drive_rs::proton_sdk::ids::NodeUid; use proton_drive_rs::{MemberRole, NodeKind}; @@ -533,9 +536,33 @@ impl Core { .rt .block_on(self.client.quota()) .map_err(|e| CoreError::from_api(&e, "account quota"))?; + *self.quota.lock() = Some((std::time::Instant::now(), q.max_space, q.used_space)); Ok((q.max_space, q.used_space)) } + /// The account quota as `statfs(2)` needs it: cached for [`QUOTA_TTL`], + /// refetched after that, and `None` when there is no usable answer at all — + /// offline with nothing cached. + /// + /// Never blocks on the network for a *fresh* value when a recent one exists, + /// because the callers are `df` and every "is there room?" preflight, and a + /// stale byte count is worth more to them than a stalled syscall. Offline, + /// the last known figures keep being served rather than degrading to zero. + pub(crate) fn account_quota_cached(&self) -> Option<(i64, i64)> { + if let Some((at, max_space, used_space)) = *self.quota.lock() + && at.elapsed() < QUOTA_TTL + { + return Some((max_space, used_space)); + } + match self.account_quota() { + Ok(q) => Some(q), + Err(error) => { + debug!(%error, "account quota refresh failed; serving the last known figures"); + self.quota.lock().map(|(_, max, used)| (max, used)) + } + } + } + // ---- shared by me ----------------------------------------------------- /// List the nodes I have shared with others, each with a summary of its share diff --git a/crates/pdfs-fuse/src/state.rs b/crates/pdfs-fuse/src/state.rs index 869c3b3..f2fe96d 100644 --- a/crates/pdfs-fuse/src/state.rs +++ b/crates/pdfs-fuse/src/state.rs @@ -920,14 +920,24 @@ impl State { /// file is as long as the caller was told it is. Without it a restart serves /// the stale size and the file reads as truncated — or empty — while its /// bytes sit safely in staging (offline.md Phase 3). - pub(crate) fn record_pending_write(&mut self, ino: u64, size: u64, mtime: i64) { + /// Failure is reported rather than logged: dropping the row silently means + /// the daemon acknowledged a write and then serves the old size for it, so + /// the caller has to be able to turn it into a failed `release`. + pub(crate) fn record_pending_write( + &mut self, + ino: u64, + size: u64, + mtime: i64, + ) -> pdfs_core::Result<()> { self.set_size(ino, size); self.touch_mtime(ino, mtime); - if let Some(e) = self.entries.get(&ino) - && let Err(err) = self.db.upsert_node(&e.node) - { - warn!(uid = %e.uid, error = %err, "db upsert_node failed for a queued write"); + if let Some(e) = self.entries.get(&ino) { + self.db.upsert_node(&e.node).map_err(|err| { + warn!(uid = %e.uid, error = %err, "db upsert_node failed for a queued write"); + err + })?; } + Ok(()) } } diff --git a/crates/pdfs-fuse/src/sweep.rs b/crates/pdfs-fuse/src/sweep.rs index 6530174..cbc978c 100644 --- a/crates/pdfs-fuse/src/sweep.rs +++ b/crates/pdfs-fuse/src/sweep.rs @@ -138,9 +138,9 @@ impl Core { /// enumerated every node up front and has been doing network I/O since. Before /// touching anything remote this re-reads the node and refuses to act unless /// it is *still* the same revision, still untrashed, still unwritten and still - /// unopened. Without that, a user edit landing inside the window is destroyed - /// — `discard_queued_ops` below drops the queued write *and its staged blob*, - /// which may hold the only copy of those bytes (`docs/BUGS.md` B71). + /// unopened (`docs/BUGS.md` B71). Nothing on this path discards a queued write + /// or its staged blob: a write that arrives after that check is left for the + /// drain to resolve against the now-trashed node. fn remove_conflict_copy(&self, snapshot: &proton_drive_rs::Node, base: &str) { let uid = &snapshot.uid; let name = snapshot.name.as_str(); @@ -166,10 +166,13 @@ impl Core { self.for_each_state(|st| { st.forget_or_unlink(uid); }); - if let Err(error) = self.discard_queued_ops(uid) { - warn!(%uid, ?error, "conflict sweep: queued-op cleanup failed"); - return; - } + // Deliberately *not* `discard_queued_ops`: `duplicate_still_removable` + // proved the queue empty before the trash above, but that proof is stale + // the moment `trash_nodes` goes to the network. A `close(2)` landing in + // that window stages bytes, and discarding here would unlink the staged + // blob holding the only copy of them. Leaving the op queued is the + // non-destructive outcome: the drain re-reads the node, sees it trashed, + // and `revision_conflict` routes it to `keep_as_conflict_copy`. self.cache.evict(uid); self.evict_reader(uid); if let Err(e) = self.db.delete_node(uid) { @@ -194,8 +197,8 @@ impl Core { let uid = &snapshot.uid; let name = snapshot.name.as_str(); - // A queued op means bytes are still owed an upload. `discard_queued_ops` - // would throw those away along with the staged blob holding them. + // A queued op means bytes are still owed an upload; trashing the node + // under them turns an ordinary upload into a conflict fork. match self.db.has_any_op(&uid.to_string()) { Ok(false) => {} Ok(true) => {