diff --git a/crates/pdfs-fuse/src/filesystem.rs b/crates/pdfs-fuse/src/filesystem.rs index 15ac739..a912f82 100644 --- a/crates/pdfs-fuse/src/filesystem.rs +++ b/crates/pdfs-fuse/src/filesystem.rs @@ -290,7 +290,7 @@ impl Filesystem for ProtonFs { } return; } - let (uid, mtime, fsize, is_video) = { + let (uid, mtime, fsize, provisional, is_video) = { let st = self.core.state(); match st.entries.get(&ino.0) { Some(e) if e.node.is_file() => { @@ -300,14 +300,24 @@ impl Filesystem for ProtonFs { }; let is_video = PhotoKind::classify(Some(&e.node.name), media_type) == PhotoKind::Video; - let fsize = self - .core - .pending - .lock() - .get(&e.uid) - .map(|p| p.meta.len) - .unwrap_or_else(|| node_size(&e.node)); - (e.uid.clone(), e.node.modification_time, fsize, is_video) + let queued = self.core.pending.lock().get(&e.uid).map(|p| p.meta.len); + // Still the cheap enumeration's ciphertext size (B12). + let provisional = queued.is_none() + && matches!( + &e.node.kind, + NodeKind::File { + claimed_size: None, + .. + } + ); + let fsize = queued.unwrap_or_else(|| node_size(&e.node)); + ( + e.uid.clone(), + e.node.modification_time, + fsize, + provisional, + is_video, + ) } Some(_) => { reply.error(Errno::EISDIR); @@ -330,6 +340,13 @@ impl Filesystem for ProtonFs { // races with nothing. FUSE does not require replies in request order. let core = self.core.clone(); self.core.workers.run(Lane::Transfer, "read", move || { + // The revision disagrees with a provisional size, and B84 fails + // such a read outright, so settle the size first. On the network, + // hence here and not above (bugs.md B100). + let fsize = match provisional { + true => core.resolve_provisional_size(ino.0, &uid).unwrap_or(fsize), + false => fsize, + }; match core.read_range(&uid, mtime, fsize, offset, size as u64, cache_blocks) { Ok(bytes) => reply.data(&bytes), Err(e) => reply.error(e), @@ -1193,7 +1210,7 @@ impl ProtonFs { /// 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 (uid, base_mtime, base_size, provisional, base_revision_id) = { let mut st = self.core.state(); match st.entries.get_mut(&ino.0) { Some(e) if e.node.is_file() => { @@ -1206,6 +1223,14 @@ impl ProtonFs { e.uid.clone(), e.node.modification_time, node_size(&e.node), + // Still the cheap enumeration's ciphertext size (B12). + matches!( + &e.node.kind, + NodeKind::File { + claimed_size: None, + .. + } + ), node_revision_id(&e.node), ) } @@ -1251,6 +1276,21 @@ impl ProtonFs { // `reply` is consumed by whichever arm answers, so the loop yields an // outcome and the answer is sent once, after it. let outcome = 'install: { + // The handle starts at the base's size. A provisional one is the + // ciphertext size, so the file would grow by the encryption overhead + // and reading the base would fail (bugs.md B100). A queued revision + // stamps a real size, so this only goes to the network for a file + // the listing has not sized yet. + let base_size = match provisional && self.core.pending_blob(&uid).is_none() { + false => base_size, + true => match self.core.resolve_provisional_size(ino.0, &uid) { + Some(size) => size, + None => { + error!(%uid, "could not settle the size of the base for a write open"); + break 'install Err(Errno::EIO); + } + }, + }; for attempt in 0..OPEN_BASE_ATTEMPTS { let pending_base = self.core.pending.lock().get(&uid).cloned(); if pending_base.as_ref().is_some_and(|p| !p.meta.complete) { diff --git a/crates/pdfs-fuse/src/lib.rs b/crates/pdfs-fuse/src/lib.rs index 1279141..69d939a 100644 --- a/crates/pdfs-fuse/src/lib.rs +++ b/crates/pdfs-fuse/src/lib.rs @@ -323,7 +323,8 @@ const SIZE_UPGRADE_CHUNK: usize = 150; /// deduplicates *within* a folder — so without this a recursive listing scaled /// both with the number of folders it walked. Eight keeps an interactive `ls -l` /// (one folder, sometimes a couple) entirely unaffected while putting a ceiling -/// on the recursive case. +/// on the recursive case. A folder past the cap is queued, not dropped: one of +/// the running threads takes it over when its own batch ends (bugs.md B100). const MAX_SIZE_UPGRADES: usize = 8; #[derive(Clone, Debug, Eq, PartialEq)] @@ -2511,10 +2512,11 @@ impl Core { /// /// **Sizes are provisional until this lands.** `node_size` falls back to /// `total_size_on_storage`, the *ciphertext* size, which is slightly larger - /// than the real one. Reads are unaffected — the revision reader carries its - /// own authoritative size — so this is a cosmetic discrepancy in `stat` that - /// closes within a round trip, not a repeat of B11 (which reported **0** and - /// so suppressed reads entirely). + /// than the real one. A read must not plan on it: since B84 a size the + /// revision disagrees with fails the read, so `read` settles the size first + /// ([`Core::resolve_provisional_size`], bugs.md B100). In `stat` it is a + /// discrepancy that closes within a round trip, not a repeat of B11 (which + /// reported **0** and so suppressed reads entirely). /// /// Single-flight per folder: a `stat` of every entry in a fresh listing is /// the normal case, and each one must not start its own upgrade. @@ -2535,32 +2537,16 @@ impl Core { if uids.is_empty() { return None; } - let slot = { - let mut in_flight = self.size_upgrades.lock(); - match in_flight.get(&key) { - // Someone else is already fetching this folder; their batch - // covers us, so just wait on it. - Some(existing) => existing.clone(), - // Nothing bounded how many of these could be in flight: the - // single-flight is per folder, so a recursive listing across - // hundreds of cold folders started a thread and a batch for - // each (audit F4). Past the cap, answer with the provisional - // size — the same fallback the deadline already takes, and the - // upgrade is retried by the next `stat` once a slot frees. - None if in_flight.len() >= MAX_SIZE_UPGRADES => return None, - None => { - let slot = Arc::new(SizeUpgrade::default()); - in_flight.insert(key, slot.clone()); - let core = self.clone(); - let worker = slot.clone(); - std::thread::spawn(move || { - core.run_size_upgrade(key, uids, &worker); - }); - slot - } + let claim = claim_size_upgrade(&mut self.size_upgrades.lock(), key, uids); + match claim { + SizeUpgradeClaim::Joined(slot) | SizeUpgradeClaim::Queued(slot) => Some(slot), + SizeUpgradeClaim::Start(slot, uids) => { + let core = self.clone(); + let worker = slot.clone(); + std::thread::spawn(move || core.run_size_upgrade(key, uids, worker)); + Some(slot) } - }; - Some(slot) + } } /// Hold `respond` back until `ino` has a real size, `slot`'s batch ends, or @@ -2603,16 +2589,52 @@ impl Core { /// single 793-node batch took ~80 s, which outran [`SizeUpgrade::WAIT`] and /// put provisional sizes back in front of callers — the bug this was /// supposed to fix (bugs.md B14). - fn run_size_upgrade(&self, key: u64, uids: Vec, slot: &SizeUpgrade) { - for chunk in uids.chunks(SIZE_UPGRADE_CHUNK) { - let result = self.rt.block_on(self.client.enumerate_nodes(chunk)); - self.apply_size_upgrade(key, result); - slot.chunk_done(); + /// + /// When the batch ends, the thread takes over the oldest batch the cap held + /// back, so at most [`MAX_SIZE_UPGRADES`] threads run however many folders + /// a recursive walk touches (bugs.md B100). + fn run_size_upgrade(&self, key: u64, uids: Vec, slot: Arc) { + let (mut key, mut uids, mut slot) = (key, uids, slot); + loop { + for chunk in uids.chunks(SIZE_UPGRADE_CHUNK) { + let result = self.rt.block_on(self.client.enumerate_nodes(chunk)); + self.apply_size_upgrade(key, result); + slot.chunk_done(); + } + // Both of these must happen however the loop ended: a folder whose + // upgrade failed has to be retryable, and its waiters released. + let next = next_size_upgrade(&mut self.size_upgrades.lock(), key); + slot.finish(); + let Some(next) = next else { + return; + }; + (key, uids, slot) = next; + } + } + + /// Settle the size of a file whose listing only carries the provisional + /// one, before a read plans its blocks on it or a write handle starts from + /// it. Returns the real size, or `None` when the fetch did not produce one. + /// + /// A provisional size is the ciphertext size. The read path checks the + /// size it is given against the revision's own block table and fails the + /// read on a mismatch (B84), so reading on the provisional size is an `EIO` + /// every time. `getattr` usually settles the size first, but not when its + /// wait times out, so the read cannot rely on it (bugs.md B100). + fn resolve_provisional_size(&self, ino: u64, uid: &NodeUid) -> Option { + let parent = self.state().entries.get(&ino)?.parent; + let result = self + .rt + .block_on(self.client.enumerate_nodes(std::slice::from_ref(uid))); + self.apply_size_upgrade(parent, result); + let st = self.state(); + match &st.entries.get(&ino)?.node.kind { + NodeKind::File { + claimed_size: Some(size), + .. + } => Some((*size).max(0) as u64), + _ => None, } - // Both of these must happen however the loop ended: a folder whose - // upgrade failed has to be retryable, and its waiters released. - self.size_upgrades.lock().remove(&key); - slot.finish(); } /// Whether `ino` has a real size — the condition a waiter is waiting on. @@ -5295,6 +5317,64 @@ struct Progress { struct SizeUpgrade { inner: Mutex, ready: Condvar, + /// The batch, while it waits for a thread because [`MAX_SIZE_UPGRADES`] + /// were already running, with when it was queued so the oldest goes first. + /// `None` once a thread has it. + queued: Mutex)>>, +} + +/// What [`Core::upgrade_sizes`] does with one folder's batch. +enum SizeUpgradeClaim { + /// A batch for this folder is already running or queued; wait on it. + Joined(Arc), + /// The cap is reached. The batch waits for a running thread to take it. + Queued(Arc), + /// Start a thread for the batch. + Start(Arc, Vec), +} + +/// Register the batch for `key` in `in_flight` and decide who runs it. +/// +/// Past the cap this used to return nothing, and the caller answered with the +/// provisional size at once. Nothing retried the folder until the next `stat`, +/// so a recursive walk over more than [`MAX_SIZE_UPGRADES`] fresh folders showed +/// ciphertext sizes for most of them (bugs.md B100). Queued, the batch is still +/// fetched, and a waiter gets the real size unless [`SizeUpgrade::WAIT`] runs out. +fn claim_size_upgrade( + in_flight: &mut HashMap>, + key: u64, + uids: Vec, +) -> SizeUpgradeClaim { + if let Some(existing) = in_flight.get(&key) { + return SizeUpgradeClaim::Joined(existing.clone()); + } + let running = in_flight + .values() + .filter(|slot| slot.queued.lock().is_none()) + .count(); + let slot = Arc::new(SizeUpgrade::default()); + in_flight.insert(key, slot.clone()); + if running >= MAX_SIZE_UPGRADES { + *slot.queued.lock() = Some((Instant::now(), uids)); + return SizeUpgradeClaim::Queued(slot); + } + SizeUpgradeClaim::Start(slot, uids) +} + +/// Retire the finished batch for `done` and hand its thread the oldest queued +/// batch, if there is one. +fn next_size_upgrade( + in_flight: &mut HashMap>, + done: u64, +) -> Option<(u64, Vec, Arc)> { + in_flight.remove(&done); + let (key, slot) = in_flight + .iter() + .filter_map(|(key, slot)| slot.queued.lock().as_ref().map(|(at, _)| (*at, *key, slot))) + .min_by_key(|(at, _, _)| *at) + .map(|(_, key, slot)| (key, slot.clone()))?; + let (_, uids) = slot.queued.lock().take()?; + Some((key, uids, slot)) } impl SizeUpgrade { @@ -5882,7 +5962,11 @@ mod merge_over_pending_tests { #[cfg(test)] mod size_upgrade_tests { - use super::{SizeUpgrade, SizeWaitQueue, SizeWaiter}; + use super::{ + LinkId, MAX_SIZE_UPGRADES, NodeUid, SizeUpgrade, SizeUpgradeClaim, SizeWaitQueue, + SizeWaiter, VolumeId, claim_size_upgrade, next_size_upgrade, + }; + use std::collections::HashMap; use std::sync::Arc; use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::mpsc; @@ -6074,6 +6158,60 @@ mod size_upgrade_tests { queue.answer_all(); assert_eq!(rx.into_iter().count(), 4); } + + fn uids(links: &[&str]) -> Vec { + links + .iter() + .map(|link| NodeUid::new(VolumeId::from("vol"), LinkId::from(*link))) + .collect() + } + + /// Past the cap a folder used to get no upgrade at all, so a recursive + /// walk reported ciphertext sizes for most of what it listed (bugs.md B100). + #[test] + fn a_folder_past_the_cap_is_queued_not_dropped() { + let mut in_flight = HashMap::new(); + for key in 0..MAX_SIZE_UPGRADES as u64 { + let claim = claim_size_upgrade(&mut in_flight, key, uids(&["a"])); + assert!(matches!(claim, SizeUpgradeClaim::Start(..))); + } + let key = MAX_SIZE_UPGRADES as u64; + let claim = claim_size_upgrade(&mut in_flight, key, uids(&["late"])); + let SizeUpgradeClaim::Queued(slot) = claim else { + panic!("expected the batch to be queued"); + }; + assert!(Arc::ptr_eq(&slot, &in_flight[&key])); + // A second caller for the queued folder waits on the same batch. + let again = claim_size_upgrade(&mut in_flight, key, uids(&["late"])); + assert!(matches!(again, SizeUpgradeClaim::Joined(s) if Arc::ptr_eq(&s, &slot))); + } + + #[test] + fn a_finished_batch_hands_its_thread_the_oldest_queued_one() { + let mut in_flight = HashMap::new(); + for key in 0..MAX_SIZE_UPGRADES as u64 { + claim_size_upgrade(&mut in_flight, key, uids(&["a"])); + } + let first = MAX_SIZE_UPGRADES as u64; + claim_size_upgrade(&mut in_flight, first, uids(&["first"])); + std::thread::sleep(Duration::from_millis(2)); + claim_size_upgrade(&mut in_flight, first + 1, uids(&["second"])); + + let (key, taken, slot) = next_size_upgrade(&mut in_flight, 0).expect("a queued batch"); + assert_eq!((key, taken), (first, uids(&["first"]))); + assert!(slot.queued.lock().is_none()); + assert!(!in_flight.contains_key(&0)); + // Still at the cap: the handed-over batch counts as running. + let running = in_flight + .values() + .filter(|s| s.queued.lock().is_none()) + .count(); + assert_eq!(running, MAX_SIZE_UPGRADES); + + let (key, _, _) = next_size_upgrade(&mut in_flight, 1).expect("the other one"); + assert_eq!(key, first + 1); + assert!(next_size_upgrade(&mut in_flight, 2).is_none()); + } } #[cfg(test)] diff --git a/docs/BUGS.md b/docs/BUGS.md index 7731f55..ff130e1 100644 --- a/docs/BUGS.md +++ b/docs/BUGS.md @@ -12,6 +12,46 @@ Conventions: --- +## B100 — A freshly copied file reads as `EIO` or 53 bytes too big + +**Status:** Fixed (unverified). No daemon carrying the change has read a file with a provisional +size yet. +**Found:** 2026-09-29, checking a 222-file copy into `Documents/pi-backup` on 2.7.0. `find -printf +%s` showed 10 small files 52 to 59 bytes bigger than their originals. `cat` of one of them failed +with `Input/output error`, and the journal had `the revision's block table disagrees with the +node size ... fsize=221 reader_size=169` followed by `repair download did not match the node +size`. The node's DB row and `claimed_size` were correct (169). A few minutes later the same files +reported the right size and matched the originals byte for byte. + +**Where:** `crates/pdfs-fuse/src/filesystem.rs` (`read`, `serve_open`), +`crates/pdfs-fuse/src/lib.rs` (`Core::upgrade_sizes`, `Core::run_size_upgrade`). + +**Cause.** Three things together. + +- A listing built from the cheap enumeration (B12) has no `claimed_size`, so a file reports its + ciphertext size until the size upgrade lands. The upgrade's own documentation said reads were + unaffected because the revision reader carries its own size. +- B84 made that untrue. `read_block` now treats a revision that disagrees with the size it was + given as damage: it repairs from the whole-file download, and when that does not match either, + it fails the read. With a provisional size the two never match, so every read failed. +- `getattr` normally settles the size before answering (B14), which kept this rare. But only + `MAX_SIZE_UPGRADES` (8) folders may upgrade at once, and past that cap `upgrade_sizes` returned + nothing. The `stat` was answered with the ciphertext size at once, and nothing retried the + folder until the next `stat`. A recursive walk over many fresh folders goes past the cap. + +A write open had the same fault: its handle started at the provisional size, so the file would +gain the encryption overhead and reading the base would fail. + +**Fix.** `read` and a write `open` settle a provisional size before using it +(`Core::resolve_provisional_size`, one node fetch, on the worker that already goes to the +network). A folder past the cap is now queued instead of dropped. When a running upgrade ends, +its thread takes over the oldest queued folder, so at most 8 threads still run. +Regression tests: `a_folder_past_the_cap_is_queued_not_dropped`, +`a_finished_batch_hands_its_thread_the_oldest_queued_one`. The read and open paths need a live +mount and have no unit test. + +--- + ## B99 — Conflict copies in synced folders are logged again on every start and shown as "Resolved" **Status:** Fixed (unverified). No daemon carrying the change has been restarted against the diff --git a/docs/CHANGELOG.md b/docs/CHANGELOG.md index 8e040b3..3d30b17 100644 --- a/docs/CHANGELOG.md +++ b/docs/CHANGELOG.md @@ -11,6 +11,13 @@ scratch (user data in `staging/` and `recovery/` is never touched by this). ## [Unreleased] +### Fixed +- **Freshly listed files read correctly.** A file whose real size had not been fetched yet + reported its encrypted size, about 53 bytes too big, and reading it failed with + `Input/output error`. This happened most often right after copying a large folder. Reads and + write opens now fetch the real size first, and a big folder walk no longer skips size lookups + (B100). + ## [2.7.0] — 2026-09-29 Schema: **35** (unchanged).