diff --git a/Cargo.lock b/Cargo.lock index 0498dfd..5486f06 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3538,7 +3538,7 @@ checksum = "35fb2e5f958ec131621fdd531e9fc186ed768cbe395337403ae56c17a74c68ec" [[package]] name = "pdfs-cli" -version = "1.8.1" +version = "1.8.2" dependencies = [ "anyhow", "clap 4.6.4", @@ -3553,7 +3553,7 @@ dependencies = [ [[package]] name = "pdfs-core" -version = "1.8.1" +version = "1.8.2" dependencies = [ "async-trait", "directories", @@ -3574,7 +3574,7 @@ dependencies = [ [[package]] name = "pdfs-fuse" -version = "1.8.1" +version = "1.8.2" dependencies = [ "fuser", "futures", @@ -3591,7 +3591,7 @@ dependencies = [ [[package]] name = "pdfs-gui" -version = "1.8.1" +version = "1.8.2" dependencies = [ "anyhow", "async-channel", diff --git a/Cargo.toml b/Cargo.toml index 44550ea..cb0d1e8 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -3,7 +3,7 @@ resolver = "2" members = ["crates/pdfs-core", "crates/pdfs-fuse", "crates/pdfs-cli", "crates/pdfs-gui"] [workspace.package] -version = "1.8.1" +version = "1.8.2" edition = "2024" license = "MIT" authors = ["Nils Pukropp "] diff --git a/crates/pdfs-core/src/cache.rs b/crates/pdfs-core/src/cache.rs index fa9f06f..77d2e24 100644 --- a/crates/pdfs-core/src/cache.rs +++ b/crates/pdfs-core/src/cache.rs @@ -72,9 +72,12 @@ fn now_millis() -> i64 { .unwrap_or(0) } -/// Block size for the on-demand block cache. Matches the SDK content block size -/// (`DEFAULT_BLOCK_SIZE`, 4 MiB) so each cached block maps to exactly one -/// `download_range` fetch with no straddling. +/// Block size the on-demand block cache falls back to when a revision's real +/// geometry is not known yet. Matches the SDK's default content block size +/// (`DEFAULT_BLOCK_SIZE`, 4 MiB), which is what this client's own uploads +/// produce — so for a file this client wrote, the fallback *is* the geometry. +/// +/// It is only a fallback, though. See [`BlockGeometry`]. pub const BLOCK_SIZE: u64 = 1 << 22; /// Buffered LRU touches that trigger a flush. Sized so a sequential read of a @@ -93,6 +96,135 @@ pub fn block_len(size: u64, idx: u64) -> u64 { size.saturating_sub(start).min(BLOCK_SIZE) } +/// One block of a revision, as the read path and the block cache address it: +/// its index in the revision's block table, and the plaintext range it covers. +/// +/// `start` travels with `idx` because the index alone stopped being enough once +/// the geometry came from the revision rather than from a constant — see +/// [`BlockGeometry`]. +#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)] +pub struct BlockSpan { + /// Position in the revision's block table. + pub idx: u64, + /// Plaintext offset the block begins at. + pub start: u64, + /// Plaintext length of the block. + pub len: u64, +} + +impl BlockSpan { + /// One past the block's last plaintext byte. + pub fn end(&self) -> u64 { + self.start + self.len + } +} + +/// Where a revision's content blocks actually begin and end in its plaintext. +/// +/// The read path used to assume the answer: [`BLOCK_SIZE`] blocks with a +/// remainder at the end, encoded as `idx * BLOCK_SIZE` and [`block_len`]. That +/// holds for revisions this client uploaded, and it does not hold in general — +/// `RevisionReader::block_sizes()` reports per-block plaintext sizes that some +/// accounts have revisions for which are neither 4 MiB nor uniform +/// (`docs/BUGS.md` B85). +/// +/// Getting it wrong is not a truncation: `read_at` plans over the real sizes +/// and clamps to the range asked for, so the bytes are right either way. What +/// it costs is work. A client block that straddles two server blocks makes the +/// reader fetch and decrypt both to answer for one, and the block written to +/// the on-disk cache then lines up with nothing the next read will ask for. +/// Planning from the revision restores the one-cached-block-is-one-fetched-block +/// property the cache was designed around. +/// +/// [`BlockGeometry::uniform`] is the old assumption, kept as the answer for a +/// file whose real geometry has not been learned yet — the first read of a file +/// has no reader open, and resolving one just to plan would reintroduce the +/// per-read key derivation the block cache exists to avoid. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct BlockGeometry { + /// Plaintext start offset of each block; same length as `sizes`. + starts: Vec, + sizes: Vec, + size: u64, +} + +impl BlockGeometry { + /// The geometry a revision reports, as a prefix sum over its block sizes. + /// + /// Trailing zero-length blocks are dropped: they name no bytes, so they can + /// only produce empty spans that every caller would have to special-case. + pub fn from_sizes(sizes: &[u64]) -> Self { + let mut kept: Vec = sizes.to_vec(); + while kept.last() == Some(&0) { + kept.pop(); + } + let mut starts = Vec::with_capacity(kept.len()); + let mut at = 0_u64; + for &len in &kept { + starts.push(at); + at = at.saturating_add(len); + } + Self { + starts, + sizes: kept, + size: at, + } + } + + /// The [`BLOCK_SIZE`] assumption, for a file of `size` bytes. + pub fn uniform(size: u64) -> Self { + let blocks = size.div_ceil(BLOCK_SIZE); + Self::from_sizes(&(0..blocks).map(|i| block_len(size, i)).collect::>()) + } + + /// Total plaintext size this geometry accounts for. + pub fn size(&self) -> u64 { + self.size + } + + /// Number of blocks. + pub fn blocks(&self) -> u64 { + self.sizes.len() as u64 + } + + /// Block `idx`, or `None` past the end of the table. + pub fn span(&self, idx: u64) -> Option { + let i = usize::try_from(idx).ok()?; + Some(BlockSpan { + idx, + start: *self.starts.get(i)?, + len: self.sizes[i], + }) + } + + /// The index of the block containing plaintext offset `at`, or `None` at or + /// past the end. + pub fn index_at(&self, at: u64) -> Option { + if at >= self.size { + return None; + } + // `starts` is sorted, so the containing block is the last one that + // begins at or before `at`. + let i = self.starts.partition_point(|&s| s <= at).checked_sub(1)?; + Some(i as u64) + } + + /// Every block overlapping the plaintext range `[offset, end)`, in order. + /// Empty when the range is empty or lies wholly past the end. + pub fn spans(&self, offset: u64, end: u64) -> Vec { + let Some(first) = self.index_at(offset) else { + return Vec::new(); + }; + if end <= offset { + return Vec::new(); + } + let last = self + .index_at(end - 1) + .unwrap_or_else(|| self.blocks().saturating_sub(1)); + (first..=last).filter_map(|i| self.span(i)).collect() + } +} + /// 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 @@ -109,6 +241,40 @@ struct Meta { size: u64, } +/// Validity tag stored alongside a cached *block*: [`Meta`] plus where in the +/// plaintext the block begins. +/// +/// The offset is what makes the block re-plannable. See +/// [`ContentCache::block_tag`] for why `(mtime, size)` alone stopped being +/// enough, and why `start` is optional. +#[derive(Serialize, Deserialize, Clone, Copy, PartialEq, Eq)] +struct BlockMeta { + mtime: i64, + size: u64, + /// Absent in sidecars written before the geometry became per-revision; + /// those are read as "wherever the 4 MiB assumption would have put it". + #[serde(default)] + start: Option, +} + +/// A revision's block geometry as cached on disk, so a read can plan without +/// opening a [`RevisionReader`](proton_sdk) first. +/// +/// Learning the geometry costs a revision listing and the file's key +/// derivation, which is exactly the per-read work the block cache exists to +/// avoid. So it is learned as a side effect of the first read — which has to +/// open a reader anyway to fetch anything — written here, and used to plan +/// every read after it. The first read of a file therefore still plans on +/// [`BlockGeometry::uniform`]; if that turns out to be wrong, the blocks it +/// cached are simply not the blocks the next read asks for, and they age out. +#[derive(Serialize, Deserialize, Clone)] +struct GeometryMeta { + mtime: i64, + size: u64, + /// Plaintext size of each content block, in block order. + sizes: Vec, +} + /// One pin, as carried over the control socket and listed in `status`. #[derive(Serialize, Deserialize, Clone, Debug)] pub struct Pin { @@ -425,7 +591,10 @@ impl ContentCache { for entry in rd.flatten() { let name = entry.file_name(); let Some(name) = name.to_str() else { continue }; - if name.ends_with(".meta") { + // `.geom` is a block-geometry sidecar (see `GeometryMeta`), tens + // of bytes and not a cached block; like `.meta` it is neither + // budgeted nor evictable on its own. + if name.ends_with(".meta") || name.ends_with(".geom") { continue; } // A `.tmp` is a staging file from a store that never completed @@ -615,60 +784,133 @@ impl ContentCache { .join(format!("{}.b{idx}.meta", Self::key(uid))) } - /// Whether block `idx` of `uid` is cached and fresh for `(mtime, size)`, - /// without reading it. For a caller deciding whether a block is worth - /// *fetching* — reading 4 MiB to answer that would defeat the point. - pub fn has_block(&self, uid: &NodeUid, mtime: i64, size: u64, idx: u64) -> bool { - self.block_tag(uid, idx) == Some(Meta { mtime, size }) - && std::fs::metadata(self.block_blob(uid, idx)) - .is_ok_and(|m| m.len() == block_len(size, idx)) + fn geometry_meta(&self, uid: &NodeUid) -> PathBuf { + self.block_dir.join(format!("{}.geom", Self::key(uid))) } - /// The `(mtime, size)` a cached block claims to belong to, from its sidecar. - fn block_tag(&self, uid: &NodeUid, idx: u64) -> Option { - serde_json::from_slice(&std::fs::read(self.block_meta(uid, idx)).ok()?).ok() - } - - /// Serve cached block `idx` (a [`BLOCK_SIZE`]-aligned chunk) of `uid`, or - /// `None` on miss/stale. Validated against `(mtime, size)` like a whole-file - /// blob, so a new revision (which bumps the mtime) is detected. Bumps the - /// block's mtime for LRU, best effort. - pub fn cached_block(&self, uid: &NodeUid, mtime: i64, size: u64, idx: u64) -> Option> { - if self.block_tag(uid, idx)? != (Meta { mtime, size }) { + /// The block geometry recorded for this revision of `uid`, or `None` if it + /// has not been learned (or belongs to an older revision). + /// + /// Callers fall back to [`BlockGeometry::uniform`], which is what every + /// read did unconditionally before `docs/BUGS.md` B85. + pub fn block_geometry(&self, uid: &NodeUid, mtime: i64, size: u64) -> Option { + let meta: GeometryMeta = + serde_json::from_slice(&std::fs::read(self.geometry_meta(uid)).ok()?).ok()?; + if meta.mtime != mtime || meta.size != size { return None; } - let mut f = std::fs::File::open(self.block_blob(uid, idx)).ok()?; + let geometry = BlockGeometry::from_sizes(&meta.sizes); + // A table that does not add up to the node's size describes a different + // file than the one being read; planning on it would place every block + // after the discrepancy wrongly. Refusing it costs a fallback to the + // uniform assumption, which is where B84's repair path takes over. + (geometry.size() == size).then_some(geometry) + } + + /// Record the block geometry `sizes` for this revision of `uid`. + /// + /// Best effort and disposable, like every other file under the block cache: + /// losing it costs the next read a plan on the uniform assumption, never + /// correctness. Not counted against the block budget — it is tens of bytes + /// per file and is what keeps the blocks that *are* counted addressable. + pub fn store_block_geometry(&self, uid: &NodeUid, mtime: i64, size: u64, sizes: &[u64]) { + let meta = GeometryMeta { + mtime, + size, + sizes: sizes.to_vec(), + }; + if let Ok(bytes) = serde_json::to_vec(&meta) { + let _ = std::fs::write(self.geometry_meta(uid), bytes); + } + } + + /// Whether `span` of `uid` is cached and fresh for `(mtime, size)`, without + /// reading it. For a caller deciding whether a block is worth *fetching* — + /// reading 4 MiB to answer that would defeat the point. + pub fn has_block(&self, uid: &NodeUid, mtime: i64, size: u64, span: BlockSpan) -> bool { + self.block_tag(uid, mtime, size, span).is_some() + && std::fs::metadata(self.block_blob(uid, span.idx)).is_ok_and(|m| m.len() == span.len) + } + + /// The sidecar of block `span.idx`, if it describes the block the caller + /// means. + /// + /// `(mtime, size)` catches a new revision, and used to be the whole check: + /// with the geometry fixed at [`BLOCK_SIZE`], an index named one byte range + /// and only one. It no longer does — the same index names a different range + /// under the revision's own geometry than under the fallback — so the + /// sidecar records the block's plaintext `start` and it has to agree too + /// (`docs/BUGS.md` B85). + /// + /// A sidecar written before that field existed deserialises with `start` + /// absent. Those are honoured only where the old assumption and the caller + /// agree on the offset, which is every file this client uploaded: the + /// existing cache survives intact for them, and is quietly refetched for the + /// files it was placing wrongly. + fn block_tag( + &self, + uid: &NodeUid, + mtime: i64, + size: u64, + span: BlockSpan, + ) -> Option { + let meta: BlockMeta = + serde_json::from_slice(&std::fs::read(self.block_meta(uid, span.idx)).ok()?).ok()?; + if meta.mtime != mtime || meta.size != size { + return None; + } + let recorded = meta + .start + .unwrap_or_else(|| span.idx.saturating_mul(BLOCK_SIZE)); + (recorded == span.start).then_some(meta) + } + + /// Serve the cached block at `span` of `uid`, or `None` on miss/stale. + /// Validated against `(mtime, size)` and the block's plaintext start, so + /// both a new revision and a re-planned geometry are detected. Bumps the + /// block's mtime for LRU, best effort. + pub fn cached_block( + &self, + uid: &NodeUid, + mtime: i64, + size: u64, + span: BlockSpan, + ) -> Option> { + self.block_tag(uid, mtime, size, span)?; + let mut f = std::fs::File::open(self.block_blob(uid, span.idx)).ok()?; // 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) { + if f.metadata().ok()?.len() != span.len { return None; } // LRU touch, buffered rather than written (see `touch`). - self.touch(self.block_key(uid, idx)); + self.touch(self.block_key(uid, span.idx)); let mut buf = Vec::new(); f.read_to_end(&mut buf).ok()?; Some(buf) } - /// Store `bytes` as cached block `idx` of `uid`, tagged `(mtime, size)`. - /// Temp-file-then-rename like [`store`](Self::store); meta written last so a - /// crash mid-store fails validation. Enforces the block-cache LRU budget. + /// Store `bytes` as the cached block at `span` of `uid`, tagged + /// `(mtime, size)` and the block's plaintext start. Temp-file-then-rename + /// like [`store`](Self::store); meta written last so a crash mid-store fails + /// validation. Enforces the block-cache LRU budget. /// - /// The caller owes it a block of exactly [`block_len`] bytes; anything else - /// reads back as a miss ([`cached_block`](Self::cached_block) checks), so a - /// bad block cannot escape this cache even though this does not check. - /// Validating a block is the *reader's* job, at the point it is fetched — - /// see `Core::read_block` (bugs.md B84). + /// The caller owes it exactly `span.len` bytes; anything else reads back as + /// a miss ([`cached_block`](Self::cached_block) checks), so a bad block + /// cannot escape this cache even though this does not check. Validating a + /// block is the *reader's* job, at the point it is fetched — see + /// `Core::read_block` (bugs.md B84). pub fn store_block( &self, uid: &NodeUid, mtime: i64, size: u64, - idx: u64, + span: BlockSpan, bytes: &[u8], ) -> Result<()> { + let idx = span.idx; let blob = self.block_blob(uid, idx); let tmp = self .block_dir @@ -685,7 +927,11 @@ impl ContentCache { std::fs::rename(&tmp, &blob)?; std::fs::write( self.block_meta(uid, idx), - serde_json::to_vec(&Meta { mtime, size })?, + serde_json::to_vec(&BlockMeta { + mtime, + size, + start: Some(span.start), + })?, )?; self.db.cache_touch( &self.block_key(uid, idx), @@ -1165,9 +1411,21 @@ impl ContentCache { /// failure here means we could not save the bytes at all, so it is reported /// rather than swallowed. pub fn stage_write(&self, meta: &StagedWrite, scratch: &Path) -> Result { + let path = self.staging_path(meta); + self.stage_write_at(meta, scratch, &path) + } + + /// [`stage_write`](Self::stage_write) with the destination named by the + /// caller. + /// + /// The public entry point picks a timestamped name, which is right for + /// production and untestable: a test cannot arrange for publication to fail + /// at a path it cannot predict. This is the seam `preserve_write_at` is for + /// the rescue path. + fn stage_write_at(&self, meta: &StagedWrite, scratch: &Path, path: &Path) -> Result { use std::io::Write as _; - let path = self.staging_path(meta); + let path = path.to_path_buf(); let copied = if let Err(e) = std::fs::rename(scratch, &path) { if e.kind() != std::io::ErrorKind::CrossesDevices { return Err(e.into()); @@ -1367,10 +1625,14 @@ impl ContentCache { let _ = std::fs::remove_file(self.thumb_blob(uid, ttype)); let _ = std::fs::remove_file(self.thumb_meta(uid, ttype)); } - // Drop every cached block (and its meta/tmp) for this uid. The block - // dir is still scanned here because eviction targets one uid's files by - // name prefix, not by LRU order; the index rows are removed in one shot - // by `cache_remove_all` below. + // Drop every cached block (and its meta/tmp) for this uid, plus the + // recorded block geometry — it describes the revision these blocks came + // from, and keeping it past them would have the next read plan against a + // table for a file it no longer holds. The block dir is still scanned + // here because eviction targets one uid's files by name prefix, not by + // LRU order; the index rows are removed in one shot by + // `cache_remove_all` below. + let _ = std::fs::remove_file(self.geometry_meta(uid)); let prefix = format!("{}.b", Self::key(uid)); if let Ok(rd) = std::fs::read_dir(&self.block_dir) { for entry in rd.flatten() { @@ -1934,17 +2196,20 @@ mod tests { // Cost of a store against a near-empty index... let t = Instant::now(); for i in 0..500u64 { - c.store_block(&uid("a"), 1, 1 << 30, i, &payload).unwrap(); + c.store_block(&uid("a"), 1, 1 << 30, blk(1 << 30, i), &payload) + .unwrap(); } let empty = t.elapsed() / 500; // ...and against one with a few thousand entries in it. for i in 500..2500u64 { - c.store_block(&uid("a"), 1, 1 << 30, i, &payload).unwrap(); + c.store_block(&uid("a"), 1, 1 << 30, blk(1 << 30, i), &payload) + .unwrap(); } let t = Instant::now(); for i in 2500..3000u64 { - c.store_block(&uid("a"), 1, 1 << 30, i, &payload).unwrap(); + c.store_block(&uid("a"), 1, 1 << 30, blk(1 << 30, i), &payload) + .unwrap(); } let full = t.elapsed() / 500; @@ -1967,7 +2232,8 @@ mod tests { fn running_totals_stay_aligned_with_the_index() { let (c, _d) = cache_with_pool_cap(KIND_BLOCK, 8); for i in 0..12u64 { - c.store_block(&uid("a"), 1, 100, i, b"aaaa").unwrap(); + c.store_block(&uid("a"), 1, 100, blk(100, i), b"aaaa") + .unwrap(); std::thread::sleep(std::time::Duration::from_millis(2)); } let indexed: u64 = @@ -2031,13 +2297,16 @@ mod tests { c.add_pin(&a, Path::new("a"), false).unwrap(); c.store(&b, 1, 4, b"bbbb").unwrap(); // A stray on-demand block of the unpinned file is purged too. - c.store_block(&b, 1, 8, 0, b"bbbb").unwrap(); + c.store_block(&b, 1, 8, blk(8, 0), b"bbbb").unwrap(); let freed = c.clear_unpinned(); assert_eq!(freed, 8, "one 4-byte blob + one 4-byte block freed"); assert!(c.is_cached(&a, 1, 4), "pinned blob survives purge"); assert!(!c.is_cached(&b, 1, 4), "unpinned blob purged"); - assert!(c.cached_block(&b, 1, 8, 0).is_none(), "block purged"); + assert!( + c.cached_block(&b, 1, 8, blk(8, 0)).is_none(), + "block purged" + ); } #[test] @@ -2171,7 +2440,7 @@ mod tests { for i in 0..100u64 { c.store(&uid(&format!("file{i}")), 1, payload.len() as u64, &payload) .unwrap(); - c.store_block(&uid("streamed"), 1, 1 << 30, i, &payload) + c.store_block(&uid("streamed"), 1, 1 << 30, blk(1 << 30, i), &payload) .unwrap(); } @@ -2249,13 +2518,28 @@ mod tests { // 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"); + c.store_block(&u, 100, 10, blk(10, 0), b"block-zero") + .unwrap(); + assert_eq!( + c.cached_block(&u, 100, 10, blk(10, 0)).unwrap(), + b"block-zero" + ); // A different index is a separate cache entry, absent here. - assert!(c.cached_block(&u, 100, 10, 1).is_none()); + assert!(c.cached_block(&u, 100, 10, blk(10, 1)).is_none()); // A new revision (mtime/size bump) invalidates the block. - assert!(c.cached_block(&u, 101, 10, 0).is_none()); - assert!(c.cached_block(&u, 100, 5000, 0).is_none()); + assert!(c.cached_block(&u, 101, 10, blk(10, 0)).is_none()); + assert!(c.cached_block(&u, 100, 5000, blk(5000, 0)).is_none()); + } + + /// The [`BlockSpan`] the uniform geometry gives block `idx` of a file of + /// `size` bytes — the shape these tests were written against, and the one + /// this client's own uploads still have. + fn blk(size: u64, idx: u64) -> BlockSpan { + BlockSpan { + idx, + start: idx.saturating_mul(BLOCK_SIZE), + len: block_len(size, idx), + } } /// The geometry the whole read path validates against, here and over the @@ -2273,6 +2557,137 @@ mod tests { assert_eq!(block_len(BLOCK_SIZE, 1), 0); } + /// The uniform geometry has to reproduce the constant-block arithmetic it + /// replaces exactly, or every file this client uploaded changes shape. + #[test] + fn the_uniform_geometry_is_the_old_block_arithmetic() { + let size = BLOCK_SIZE * 2 + 17; + let g = BlockGeometry::uniform(size); + assert_eq!(g.size(), size); + assert_eq!(g.blocks(), 3); + for idx in 0..3 { + let span = g.span(idx).unwrap(); + assert_eq!(span.start, idx * BLOCK_SIZE); + assert_eq!(span.len, block_len(size, idx)); + } + assert!(g.span(3).is_none(), "past the end there is no block"); + + // An empty file has no blocks at all, and no range covers anything. + let empty = BlockGeometry::uniform(0); + assert_eq!(empty.blocks(), 0); + assert!(empty.spans(0, 10).is_empty()); + } + + /// The B85 case: block sizes that are neither 4 MiB nor uniform. Offsets + /// have to come from the prefix sum, not from a constant. + #[test] + fn a_non_uniform_geometry_places_blocks_by_prefix_sum() { + let sizes = [5_239_804_u64, 5_670_228, 5_233_034]; + let g = BlockGeometry::from_sizes(&sizes); + assert_eq!(g.size(), sizes.iter().sum::()); + assert_eq!(g.blocks(), 3); + assert_eq!(g.span(0).unwrap().start, 0); + assert_eq!(g.span(1).unwrap().start, 5_239_804); + assert_eq!(g.span(2).unwrap().start, 5_239_804 + 5_670_228); + assert_eq!(g.span(2).unwrap().end(), g.size()); + + // A one-byte read in the middle of block 1 asks for block 1 alone — + // under the 4 MiB assumption the same offset falls in block 2. + let at = 6_000_000; + assert_eq!(g.index_at(at), Some(1)); + assert_eq!(BlockGeometry::uniform(g.size()).index_at(at), Some(1)); + let spans = g.spans(at, at + 1); + assert_eq!(spans.len(), 1); + assert_eq!(spans[0].idx, 1); + + // A range crossing a boundary asks for both, and only both. + let crossing = g.spans(5_239_803, 5_239_805); + assert_eq!( + crossing.iter().map(|s| s.idx).collect::>(), + vec![0, 1] + ); + + // At and past the end there is nothing to read. + assert_eq!(g.index_at(g.size()), None); + assert!(g.spans(g.size(), g.size() + 100).is_empty()); + assert!(g.spans(10, 10).is_empty(), "an empty range wants no blocks"); + + // The single-block case from the same log: one block holding the whole + // file, so there is no block 1 to ask for. + let whole = BlockGeometry::from_sizes(&[4_939_506]); + assert_eq!(whole.blocks(), 1); + assert!(whole.span(1).is_none()); + assert_eq!(whole.spans(0, 4_939_506).len(), 1); + } + + /// A cached block is addressed by the range it covers, not by its index + /// alone. Re-planning a file onto its real geometry must therefore miss the + /// blocks the uniform fallback cached, rather than serve their bytes at the + /// wrong offset (bugs.md B85). + #[test] + fn a_cached_block_is_not_served_to_a_different_geometry() { + let (c, _d) = cache(); + let u = uid("geom"); + let size = BLOCK_SIZE * 2; + + let uniform = BlockGeometry::uniform(size); + let first = uniform.span(0).unwrap(); + let payload = vec![7u8; first.len as usize]; + c.store_block(&u, 1, size, first, &payload).unwrap(); + assert!(c.has_block(&u, 1, size, first), "cached where it was put"); + + // The same file, re-planned: block 0 is now a different length, and + // block 1 starts somewhere the old plan never had a block. + let real = BlockGeometry::from_sizes(&[BLOCK_SIZE + 1, BLOCK_SIZE - 1]); + assert_eq!(real.size(), size); + assert!( + !c.has_block(&u, 1, size, real.span(0).unwrap()), + "same index, different length: not the block that was cached" + ); + assert!( + c.cached_block(&u, 1, size, real.span(1).unwrap()).is_none(), + "same index, different start: not the block that was cached" + ); + } + + /// The geometry sidecar is what lets a later read plan without opening a + /// reader. It is tied to the revision, and refused if it does not describe + /// a file of the size being read. + #[test] + fn a_recorded_geometry_is_returned_only_for_the_revision_it_describes() { + let (c, _d) = cache(); + let u = uid("geom"); + let sizes = [5_239_804_u64, 5_670_228]; + let size: u64 = sizes.iter().sum(); + + assert!( + c.block_geometry(&u, 1, size).is_none(), + "nothing is known before the first read" + ); + + c.store_block_geometry(&u, 1, size, &sizes); + assert_eq!( + c.block_geometry(&u, 1, size), + Some(BlockGeometry::from_sizes(&sizes)) + ); + assert!( + c.block_geometry(&u, 2, size).is_none(), + "a new revision bumps the mtime, and this table described the old one" + ); + assert!(c.block_geometry(&u, 1, size + 1).is_none()); + + // A table that does not add up to the file is refused outright rather + // than used to place blocks that would land wrongly. + c.store_block_geometry(&u, 1, size, &[1, 2, 3]); + assert!(c.block_geometry(&u, 1, size).is_none()); + + // Evicting the file's content takes its geometry with it, so the next + // read does not plan against a table for bytes the cache no longer has. + c.store_block_geometry(&u, 1, size, &sizes); + c.evict(&u); + assert!(c.block_geometry(&u, 1, size).is_none()); + } + /// A block of the wrong length cannot escape the cache even though the /// store does not check: the read validates it and reports a miss, so the /// next read refetches rather than serving a hole (bugs.md B84). @@ -2281,9 +2696,9 @@ mod tests { let (c, _d) = cache(); let u = uid("a"); // A 10-byte file's only block is 10 bytes; 5 is not a short tail. - c.store_block(&u, 100, 10, 0, b"block").unwrap(); - assert!(c.cached_block(&u, 100, 10, 0).is_none()); - assert!(!c.has_block(&u, 100, 10, 0)); + c.store_block(&u, 100, 10, blk(10, 0), b"block").unwrap(); + assert!(c.cached_block(&u, 100, 10, blk(10, 0)).is_none()); + assert!(!c.has_block(&u, 100, 10, blk(10, 0))); } /// The block cache is written without an fsync, so a crash can leave a blob @@ -2293,9 +2708,10 @@ mod tests { 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(); + c.store_block(&u, 100, 10, blk(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()); + assert!(c.cached_block(&u, 100, 10, blk(10, 0)).is_none()); } #[test] @@ -2305,10 +2721,10 @@ mod tests { // 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.store_block(&u, 1, 4, blk(4, 0), b"aaaa").unwrap(); + assert!(c.cached_block(&u, 1, 4, blk(4, 0)).is_some()); c.evict(&u); - assert!(c.cached_block(&u, 1, 4, 0).is_none()); + assert!(c.cached_block(&u, 1, 4, blk(4, 0)).is_none()); } #[test] @@ -2317,20 +2733,26 @@ mod tests { // so each fixture is a geometrically real single-block file. let (c, _d) = cache_with_pool_cap(KIND_BLOCK, 8); let (a, b, d) = (uid("a"), uid("b"), uid("d")); - c.store_block(&a, 1, 4, 0, b"aaaa").unwrap(); + c.store_block(&a, 1, 4, blk(4, 0), b"aaaa").unwrap(); std::thread::sleep(std::time::Duration::from_millis(10)); - c.store_block(&b, 1, 4, 0, b"bbbb").unwrap(); + c.store_block(&b, 1, 4, blk(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(&a, 1, 4, 0).is_some()); + assert!(c.cached_block(&a, 1, 4, blk(4, 0)).is_some()); std::thread::sleep(std::time::Duration::from_millis(10)); - c.store_block(&d, 1, 4, 0, b"cccc").unwrap(); + c.store_block(&d, 1, 4, blk(4, 0), b"cccc").unwrap(); assert!( - c.cached_block(&a, 1, 4, 0).is_some(), + c.cached_block(&a, 1, 4, blk(4, 0)).is_some(), "recently-read 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"); + assert!( + c.cached_block(&b, 1, 4, blk(4, 0)).is_none(), + "LRU block evicted" + ); + assert!( + c.cached_block(&d, 1, 4, blk(4, 0)).is_some(), + "newest survives" + ); } /// A failed upload must never cost the user their bytes (offline.md Phase 2): @@ -2634,6 +3056,379 @@ mod tests { assert!(c2.recovered_writes().is_empty()); } + /// A directory on a *different* filesystem from the cache, so the + /// cross-device fallbacks can be driven for real rather than reasoned + /// about. `None` when this machine offers no second filesystem to use, in + /// which case the callers skip rather than fail — the fallback is still + /// covered by the same-device paths, and a green suite that quietly means + /// "not checked" is worse than a skip that says so. + fn other_device_dir(label: &str) -> Option { + use std::os::unix::fs::MetadataExt as _; + let here = std::fs::metadata(std::env::temp_dir()).ok()?.dev(); + let candidate = Path::new("/dev/shm"); + if std::fs::metadata(candidate).ok()?.dev() == here { + return None; + } + let dir = candidate.join(format!("pdfs-xdev-{label}-{}", std::process::id())); + let _ = std::fs::remove_dir_all(&dir); + std::fs::create_dir_all(&dir).ok()?; + Some(dir) + } + + // ---- B50: an orphaned write's only copy is never the one deleted ------- + + /// A staging directory that cannot be written to at all. Rename fails, the + /// copy fallback fails, and there is nowhere else to put the bytes — so the + /// scratch file, which is the only copy, has to still be there and still be + /// the bytes the user wrote. + #[test] + fn orphan_preservation_keeps_the_scratch_copy_when_staging_is_unwritable() { + use std::io::Write as _; + use std::os::unix::fs::PermissionsExt as _; + + let (c, _dir) = cache(); + let (mut file, scratch) = c.create_scratch().unwrap(); + file.write_all(b"the only copy there is").unwrap(); + file.sync_all().unwrap(); + drop(file); + + std::fs::set_permissions(&c.staging_dir, std::fs::Permissions::from_mode(0o500)).unwrap(); + let error = c + .preserve_write(&preservation_meta(), &scratch) + .unwrap_err(); + std::fs::set_permissions(&c.staging_dir, std::fs::Permissions::from_mode(0o700)).unwrap(); + + assert_eq!(error.surviving_path, scratch); + assert_eq!( + std::fs::read(&scratch).unwrap(), + b"the only copy there is", + "byte-identical, and still where the error says it is" + ); + assert!( + error.marker_path.is_some(), + "and carrying the marker that makes it recoverable" + ); + } + + /// The same failure, seen from the next startup: the retained scratch file + /// is not just present, it is *discoverable* — recovery finds it with its + /// metadata, which is what makes it a preserved write rather than debris. + #[test] + fn a_retained_orphan_write_is_discoverable_after_a_restart() { + use std::io::Write as _; + use std::os::unix::fs::PermissionsExt as _; + + let dir = TempDir::new(); + let db = Arc::new(Db::open_in_memory().unwrap()); + let content = dir.path().join("content"); + let pins = dir.path().join("pins.json"); + let c = ContentCache::open(content.clone(), pins.clone(), 0, db.clone()).unwrap(); + + let (mut file, scratch) = c.create_scratch().unwrap(); + file.write_all(b"accepted bytes").unwrap(); + file.sync_all().unwrap(); + drop(file); + + std::fs::set_permissions(&c.staging_dir, std::fs::Permissions::from_mode(0o500)).unwrap(); + assert!(c.preserve_write(&preservation_meta(), &scratch).is_err()); + std::fs::set_permissions(&c.staging_dir, std::fs::Permissions::from_mode(0o700)).unwrap(); + drop(c); + + let reopened = ContentCache::open(content, pins, 0, db).unwrap(); + let recovered = reopened.recovered_writes(); + assert_eq!(recovered.len(), 1); + assert_eq!(std::fs::read(&recovered[0].0).unwrap(), b"accepted bytes"); + assert_eq!(recovered[0].1.uid, preservation_meta().uid); + assert_eq!(recovered[0].1.authored, preservation_meta().authored); + } + + /// The copy fallback, driven across a real device boundary. Its whole point + /// is ordering: the destination and its sidecar are published and synced + /// *first*, and only then is the source removed. + #[test] + fn orphan_preservation_across_a_device_boundary_publishes_before_removing() { + use std::io::Write as _; + + let Some(elsewhere) = other_device_dir("preserve") else { + eprintln!("skipped: no second filesystem available to cross"); + return; + }; + let (c, _dir) = cache(); + let scratch = elsewhere.join("scratch-blob"); + { + let mut file = std::fs::File::create(&scratch).unwrap(); + file.write_all(b"accepted bytes").unwrap(); + file.sync_all().unwrap(); + } + + let preserved = c.preserve_write(&preservation_meta(), &scratch).unwrap(); + + assert_eq!(std::fs::read(&preserved).unwrap(), b"accepted bytes"); + assert!( + preserved.with_extension("json").is_file(), + "the sidecar is published before the source is dropped" + ); + assert!(!scratch.exists(), "and only then is the source removed"); + assert!( + !ContentCache::scratch_sidecar(&scratch).exists(), + "the source's durability marker goes with it" + ); + let _ = std::fs::remove_dir_all(&elsewhere); + } + + /// And when publication across that boundary fails partway, the marked + /// source stays put. This is the case B50 was: the old code removed it. + #[test] + fn a_failed_cross_device_preservation_keeps_the_marked_source() { + use std::io::Write as _; + + let Some(elsewhere) = other_device_dir("preserve-fail") else { + eprintln!("skipped: no second filesystem available to cross"); + return; + }; + let (c, _dir) = cache(); + let scratch = elsewhere.join("scratch-blob"); + { + let mut file = std::fs::File::create(&scratch).unwrap(); + file.write_all(b"only surviving copy").unwrap(); + file.sync_all().unwrap(); + } + + let destination = c.staging_dir.join("cross-device-failure"); + std::fs::create_dir(destination.with_extension("json")).unwrap(); + + let error = c + .preserve_write_at(&preservation_meta(), &scratch, &destination) + .unwrap_err(); + + assert_eq!(error.surviving_path, scratch); + assert_eq!( + std::fs::read(&scratch).unwrap(), + b"only surviving copy", + "the source outlives a publication that did not finish" + ); + assert_eq!( + error.marker_path, + Some(ContentCache::scratch_sidecar(&scratch)), + "and is marked, so the next startup rescues it" + ); + let _ = std::fs::remove_dir_all(&elsewhere); + } + + // ---- B51: the scratch sidecar is a durable, self-validating record ----- + + /// A sidecar is published by rename, so a crash leaves either the whole + /// record or none of it — never a half-written one under the real name. + /// The temporary it is renamed from must also not be mistaken for a record. + #[test] + fn a_half_written_scratch_sidecar_is_never_read_as_a_record() { + use std::io::Write as _; + + let dir = TempDir::new(); + let db = Arc::new(Db::open_in_memory().unwrap()); + let content = dir.path().join("content"); + let pins = dir.path().join("pins.json"); + let c = ContentCache::open(content.clone(), pins.clone(), 0, db.clone()).unwrap(); + + let (mut file, scratch) = c.create_scratch().unwrap(); + file.write_all(b"authored bytes").unwrap(); + file.sync_all().unwrap(); + drop(file); + + // A crash between writing the temporary and renaming it: exactly what + // is on disk if the process dies inside `mark_scratch_durable`. + let side = ContentCache::scratch_sidecar(&scratch); + std::fs::write(side.with_extension("json.tmp"), b"{\"uid\":\"prot").unwrap(); + assert!(!side.exists(), "no record was ever published"); + + drop(c); + let reopened = ContentCache::open(content, pins, 0, db).unwrap(); + assert!( + reopened.recovered_writes().is_empty(), + "an unpublished temporary describes no write" + ); + } + + /// A sidecar that survived but does not parse — a torn tail, a truncated + /// file — must not take its blob down with it. The bytes are the user's + /// work; the metadata only says what to do with them. + #[test] + fn a_corrupt_scratch_sidecar_still_rescues_its_bytes() { + use std::io::Write as _; + + let dir = TempDir::new(); + let db = Arc::new(Db::open_in_memory().unwrap()); + let content = dir.path().join("content"); + let pins = dir.path().join("pins.json"); + let c = ContentCache::open(content.clone(), pins.clone(), 0, db.clone()).unwrap(); + + let (mut file, scratch) = c.create_scratch().unwrap(); + file.write_all(b"user work").unwrap(); + file.sync_all().unwrap(); + drop(file); + let meta = StagedWrite { + uid: uid("corrupt").to_string(), + len: 9, + base_size: 0, + base_mtime: 0, + authored: vec![(0, 9)], + complete: true, + based_on: None, + }; + c.mark_scratch_durable(&scratch, &meta).unwrap(); + + // Truncate the published sidecar to a valid JSON prefix. + let side = ContentCache::scratch_sidecar(&scratch); + let full = std::fs::read(&side).unwrap(); + std::fs::write(&side, &full[..full.len() / 2]).unwrap(); + + drop(c); + let reopened = ContentCache::open(content.clone(), pins, 0, db).unwrap(); + + assert!( + reopened.recovered_writes().is_empty(), + "an unparseable record cannot be replayed" + ); + let rescued: Vec = std::fs::read_dir(content.join("recovery")) + .unwrap() + .flatten() + .map(|e| e.path()) + .filter(|p| p.extension().is_none_or(|e| e != "json")) + .collect(); + assert_eq!(rescued.len(), 1, "the blob is still moved to recovery"); + assert_eq!(std::fs::read(&rescued[0]).unwrap(), b"user work"); + } + + /// `mark_scratch_durable` is called repeatedly as a file is fsynced. Each + /// call has to leave one complete, parseable record and no debris. + #[test] + fn remarking_a_scratch_file_replaces_its_record_atomically() { + use std::io::Write as _; + + let (c, _dir) = cache(); + let (mut file, scratch) = c.create_scratch().unwrap(); + file.write_all(b"first").unwrap(); + file.sync_all().unwrap(); + drop(file); + + let mut meta = StagedWrite { + uid: uid("remark").to_string(), + len: 5, + base_size: 0, + base_mtime: 0, + authored: vec![(0, 5)], + complete: true, + based_on: None, + }; + c.mark_scratch_durable(&scratch, &meta).unwrap(); + meta.len = 11; + meta.authored = vec![(0, 11)]; + c.mark_scratch_durable(&scratch, &meta).unwrap(); + + let side = ContentCache::scratch_sidecar(&scratch); + let got: StagedWrite = serde_json::from_slice(&std::fs::read(&side).unwrap()).unwrap(); + assert_eq!(got.len, 11); + assert_eq!(got.authored, vec![(0, 11)]); + assert!( + !side.with_extension("json.tmp").exists(), + "the temporary is renamed, not left behind" + ); + } + + // ---- B52: staged-write publication is crash-safe ----------------------- + + /// The cross-filesystem half of `stage_write`. The source is copied, not + /// renamed, and must not be removed until the destination and its sidecar + /// are both on disk. + #[test] + fn staging_across_a_device_boundary_removes_the_source_last() { + use std::io::Write as _; + + let Some(elsewhere) = other_device_dir("stage") else { + eprintln!("skipped: no second filesystem available to cross"); + return; + }; + let (c, _dir) = cache(); + let scratch = elsewhere.join("scratch-blob"); + { + let mut file = std::fs::File::create(&scratch).unwrap(); + file.write_all(b"staged across devices").unwrap(); + file.sync_all().unwrap(); + } + + let meta = preservation_meta(); + let staged = c.stage_write(&meta, &scratch).unwrap(); + + assert_eq!(std::fs::read(&staged).unwrap(), b"staged across devices"); + let sidecar: StagedWrite = + serde_json::from_slice(&std::fs::read(staged.with_extension("json")).unwrap()).unwrap(); + assert_eq!(sidecar.uid, meta.uid); + assert!( + !scratch.exists(), + "the source goes last, and only on success" + ); + let _ = std::fs::remove_dir_all(&elsewhere); + } + + /// And when the sidecar cannot be published, the copied source stays: a + /// staged blob nothing describes is not a substitute for the bytes. + #[test] + fn a_failed_staging_sidecar_keeps_the_cross_device_source() { + use std::io::Write as _; + + let Some(elsewhere) = other_device_dir("stage-fail") else { + eprintln!("skipped: no second filesystem available to cross"); + return; + }; + let (c, _dir) = cache(); + let scratch = elsewhere.join("scratch-blob"); + { + let mut file = std::fs::File::create(&scratch).unwrap(); + file.write_all(b"still the source").unwrap(); + file.sync_all().unwrap(); + } + + // Block the sidecar's temporary name with a directory, so the copy + // lands and only publication fails. + let meta = preservation_meta(); + let destination = c.staging_dir.join("blocked-sidecar"); + std::fs::create_dir(destination.with_extension("json.tmp")).unwrap(); + + assert!(c.stage_write_at(&meta, &scratch, &destination).is_err()); + assert_eq!( + std::fs::read(&scratch).unwrap(), + b"still the source", + "an unpublished staging blob does not authorise deleting the source" + ); + let _ = std::fs::remove_dir_all(&elsewhere); + } + + /// A staged blob whose sidecar never landed is still user data. Startup + /// must leave both it and the leftover temporary alone rather than tidying + /// the bytes away. + #[test] + fn an_unaccompanied_staged_blob_survives_a_restart() { + let dir = TempDir::new(); + let db = Arc::new(Db::open_in_memory().unwrap()); + let content = dir.path().join("content"); + let pins = dir.path().join("pins.json"); + let c = ContentCache::open(content.clone(), pins.clone(), 0, db.clone()).unwrap(); + + let orphan = c.staging_dir.join("unaccompanied-bytes"); + std::fs::write(&orphan, b"nobody describes me").unwrap(); + let leftover = c.staging_dir.join("interrupted.json.tmp"); + std::fs::write(&leftover, b"{\"uid\":\"prot").unwrap(); + + drop(c); + let _reopened = ContentCache::open(content, pins, 0, db).unwrap(); + assert_eq!( + std::fs::read(&orphan).unwrap(), + b"nobody describes me", + "staging is never cleared on startup" + ); + assert!(leftover.exists()); + } + #[test] fn scratch_file_is_writable_and_isolated() { use std::io::{Read as _, Seek as _, SeekFrom, Write as _}; diff --git a/crates/pdfs-core/src/db/tests.rs b/crates/pdfs-core/src/db/tests.rs index d34f87b..c44d4ff 100644 --- a/crates/pdfs-core/src/db/tests.rs +++ b/crates/pdfs-core/src/db/tests.rs @@ -1895,6 +1895,123 @@ fn failed_superseding_insert_keeps_the_old_pending_op() { assert_eq!(ops[0].blob_path.as_deref(), Some("/staging/original")); } +/// B53's remaining half. The insert-failure regression above proves the *row* +/// rolls back; this proves the two things that make the rollback usable — the +/// old blob is not reported as superseded (so the caller does not delete the +/// bytes the surviving row still points at), and the rollback survives a +/// restart rather than living only in the connection that failed. +#[test] +fn a_rolled_back_supersede_keeps_its_blob_and_survives_reopen() { + let dir = std::env::temp_dir().join(format!("pdfs-db-b53-{}", std::process::id())); + let _ = std::fs::remove_dir_all(&dir); + std::fs::create_dir_all(&dir).unwrap(); + let path = dir.join("queue.sqlite"); + let op = |blob: &str| PendingOp { + id: 0, + kind: OP_REVISION.to_string(), + uid: uid("durable").to_string(), + parent_uid: None, + name: None, + blob_path: Some(blob.to_string()), + meta_json: Some("{}".to_string()), + created_at: 1, + attempts: 0, + last_error: None, + next_attempt_at: 0, + }; + + { + let db = Db::open(&path).unwrap(); + db.enqueue_op(&op("/staging/original")).unwrap(); + db.with_conn(|conn| { + conn.execute_batch( + "CREATE TRIGGER reject_replacement BEFORE INSERT ON pending_op + WHEN NEW.blob_path = '/staging/replacement' + BEGIN SELECT RAISE(ABORT, 'injected insert failure'); END;", + )?; + Ok(()) + }) + .unwrap(); + + // The failure must not hand back the old blob path. That value is the + // caller's instruction to delete those bytes, and the row that owns + // them is still queued. + assert!(db.enqueue_op(&op("/staging/replacement")).is_err()); + db.with_conn(|conn| { + conn.execute_batch("DROP TRIGGER reject_replacement")?; + Ok(()) + }) + .unwrap(); + } + + let db = Db::open(&path).unwrap(); + let ops = db.pending_ops().unwrap(); + assert_eq!(ops.len(), 1, "old or new — never neither"); + assert_eq!(ops[0].blob_path.as_deref(), Some("/staging/original")); + + // And once the injected failure is gone, superseding works and *does* + // report the blob it retired. + let (_, superseded) = db.enqueue_op(&op("/staging/replacement")).unwrap(); + assert_eq!(superseded.as_deref(), Some("/staging/original")); + assert_eq!(db.pending_ops().unwrap().len(), 1); + + drop(db); + std::fs::remove_dir_all(&dir).unwrap(); +} + +/// A full disk, as SQLite reports it. `max_page_count` makes the database +/// refuse to grow, which is the same `SQLITE_FULL` an out-of-space filesystem +/// produces — and the queue must come out of it with the acknowledged work +/// still in it (B53). +#[test] +fn a_full_database_rolls_the_supersede_back_rather_than_erasing_it() { + let db = Db::open_in_memory().unwrap(); + let op = |blob: &str| PendingOp { + id: 0, + kind: OP_REVISION.to_string(), + uid: uid("full").to_string(), + parent_uid: None, + name: None, + blob_path: Some(blob.to_string()), + meta_json: Some("{}".to_string()), + created_at: 1, + attempts: 0, + last_error: None, + next_attempt_at: 0, + }; + db.enqueue_op(&op("/staging/original")).unwrap(); + + // Cap the file at its current size, so any page the insert needs fails. + let pages: i64 = db + .with_conn(|conn| Ok(conn.query_row("PRAGMA page_count", [], |r| r.get(0))?)) + .unwrap(); + db.with_conn(|conn| { + conn.execute_batch(&format!("PRAGMA max_page_count = {pages}"))?; + Ok(()) + }) + .unwrap(); + + // A payload far larger than one page, so growth is unavoidable. + let mut fat = op("/staging/replacement"); + fat.meta_json = Some("x".repeat(512 * 1024)); + let result = db.enqueue_op(&fat); + + db.with_conn(|conn| { + conn.execute_batch("PRAGMA max_page_count = 1073741823")?; + Ok(()) + }) + .unwrap(); + + assert!(result.is_err(), "a database that cannot grow must refuse"); + let ops = db.pending_ops().unwrap(); + assert_eq!(ops.len(), 1); + assert_eq!( + ops[0].blob_path.as_deref(), + Some("/staging/original"), + "the acknowledged upload is still queued and still owns its bytes" + ); +} + #[test] fn atomic_trash_replacement_returns_owned_blobs_after_commit() { let db = Db::open_in_memory().unwrap(); diff --git a/crates/pdfs-fuse/src/control.rs b/crates/pdfs-fuse/src/control.rs index 64cfc83..93542c0 100644 --- a/crates/pdfs-fuse/src/control.rs +++ b/crates/pdfs-fuse/src/control.rs @@ -104,6 +104,31 @@ fn read_request_line(reader: &mut impl BufRead) -> std::io::Result CoreResult<(Core, PathBuf)> { + let p = Path::new(path); + if !p.is_absolute() { + return Ok((core.clone(), p.to_path_buf())); + } + if let Some(routed) = core.rooted_at(p) { + return Ok(routed); + } + // Fall back to the primary reading so the error text stays the familiar one + // when the path is simply outside everything this daemon has mounted. + rel_to_mount(mountpoint, path).map(|rel| (core.clone(), rel)) +} + /// Turn a CLI-supplied path into a mountpoint-relative path. An absolute path /// must live under `mountpoint`; a relative path is taken as already relative to /// the mount root. @@ -193,8 +218,8 @@ fn handle_control_conn(core: &Core, username: &str, mountpoint: &Path, stream: U Ok(CtlRequest::ListPins) => CtlResponse::Pins { pins: core.cache.list_pins(), }, - Ok(CtlRequest::ListDir { path }) => match rel_to_mount(mountpoint, &path) { - Ok(rel) => match core.list_dir(&rel) { + Ok(CtlRequest::ListDir { path }) => match route_to_mount(core, mountpoint, &path) { + Ok((core, rel)) => match core.list_dir(&rel) { Ok(entries) => CtlResponse::Entries { entries }, Err(e) => CtlResponse::error(e), }, @@ -202,9 +227,8 @@ fn handle_control_conn(core: &Core, username: &str, mountpoint: &Path, stream: U }, Ok(CtlRequest::Refresh { scope }) => { let result = match &scope { - RefreshScope::Dir { path } => { - rel_to_mount(mountpoint, path).and_then(|rel| core.refresh_dir(&rel)) - } + RefreshScope::Dir { path } => route_to_mount(core, mountpoint, path) + .and_then(|(core, rel)| core.refresh_dir(&rel)), RefreshScope::Trash => { core.invalidate_trash(); Ok(()) diff --git a/crates/pdfs-fuse/src/lib.rs b/crates/pdfs-fuse/src/lib.rs index ab460a1..4799fa9 100644 --- a/crates/pdfs-fuse/src/lib.rs +++ b/crates/pdfs-fuse/src/lib.rs @@ -55,7 +55,7 @@ use fuser::{ }; use futures::StreamExt as _; use pdfs_core::batch; -use pdfs_core::cache::{BLOCK_SIZE, Baseline, ContentCache, StagedWrite, block_len}; +use pdfs_core::cache::{BLOCK_SIZE, Baseline, BlockGeometry, BlockSpan, ContentCache, StagedWrite}; use pdfs_core::config::{AppDirs, SweepMode}; use pdfs_core::control::{ ActivityEntry, ActivityKind, DirEntry, ErrorKind, LocalHit, PhotoKind, PublicLinkInfo, @@ -715,6 +715,26 @@ struct MountedState { /// [`Core::notifier`], still empty until its session is spawned. notifier: Arc>, session_live: Arc, + /// This mount's in-flight size upgrades, keyed by *its* folder inodes — so + /// a `Core` rebuilt onto this inode space by [`Core::rooted_at`] shares the + /// batch the mount's own session is already running rather than starting a + /// second one against inode numbers that mean nothing to it. + size_upgrades: Arc>>>, +} + +/// The per-mount half of a [`Core`] — every field [`Core::fork_state`] gives a +/// fork a fresh copy of — recovered from the registry. +/// +/// This is what lets a control request naming a path under a secondary mount be +/// answered *in that mount's inode space* instead of being rejected for not +/// being under the primary mountpoint (`docs/BUGS.md` B86). See +/// [`Core::rooted_at`]. +struct MountParts { + mountpoint: PathBuf, + state: Arc>, + notifier: Arc>, + session_live: Arc, + size_upgrades: Arc>>>, } /// Every mounted inode space in the daemon, shared by the primary `Core` and @@ -737,6 +757,7 @@ impl StateRegistry { state: &Arc>, notifier: Arc>, session_live: Arc, + size_upgrades: Arc>>>, ) { let mut states = self.0.lock(); states.retain(|m| m.state.strong_count() > 0); @@ -751,6 +772,7 @@ impl StateRegistry { state: Arc::downgrade(state), notifier, session_live, + size_upgrades, }); } @@ -777,9 +799,9 @@ impl StateRegistry { /// /// On-demand roots can be nested below another location. Selecting by the /// longest component prefix ensures the nested session wins rather than the - /// broader primary mount. Callers currently need the selected mount only; - /// a future path router can derive a relative suffix with - /// `path.strip_prefix(mountpoint)` without storing a second path value here. + /// broader primary mount. Callers that need the relative suffix derive it + /// with `path.strip_prefix(mountpoint)`; [`StateRegistry::covering_parts`] + /// is the variant that also hands back enough to *serve* a request there. fn covering(&self, path: &Path) -> Option { self.live() .into_iter() @@ -789,6 +811,48 @@ impl StateRegistry { .max_by_key(|(mountpoint, _, _, _)| mountpoint.components().count()) } + /// [`StateRegistry::register`] with an empty size-upgrade map, for the + /// tests that exercise the reap-and-route rules and nothing else. + #[cfg(test)] + fn register_bare( + &self, + mountpoint: &Path, + state: &Arc>, + notifier: Arc>, + session_live: Arc, + ) { + self.register( + mountpoint, + state, + notifier, + session_live, + Arc::new(Mutex::new(HashMap::new())), + ); + } + + /// [`StateRegistry::covering`], but returning every per-mount field a + /// [`Core`] needs to be re-rooted onto that inode space. + /// + /// Kept separate from `covering` so the common callers — which only ask + /// *whether* a path is mounted — keep the cheaper tuple. + fn covering_parts(&self, path: &Path) -> Option { + let mut states = self.0.lock(); + states.retain(|m| m.state.strong_count() > 0); + states + .iter() + .filter(|m| m.session_live.load(Ordering::Acquire) && path.starts_with(&m.mountpoint)) + .max_by_key(|m| m.mountpoint.components().count()) + .and_then(|m| { + Some(MountParts { + mountpoint: m.mountpoint.clone(), + state: m.state.upgrade()?, + notifier: m.notifier.clone(), + session_live: m.session_live.clone(), + size_upgrades: m.size_upgrades.clone(), + }) + }) + } + fn is_mounted_at(&self, path: &Path) -> bool { self.covering(path).is_some_and(|(mountpoint, _, _, live)| { mountpoint == path && live.load(Ordering::Acquire) @@ -908,9 +972,41 @@ impl Core { &self.state, self.notifier.clone(), self.session_live.clone(), + self.size_upgrades.clone(), ); } + /// This `Core`, re-rooted onto whichever live mount most specifically covers + /// the absolute path `abs`, together with `abs` relative to that mount's + /// root. + /// + /// Everything a `Core` holds is per-daemon except the five fields + /// [`Core::fork_state`] replaces, so swapping exactly those turns the + /// primary `Core` into the fork that owns the path — sharing the fork's + /// inode space, notification channel and size-upgrade batches rather than + /// shadowing them. Without this, every path-addressed control request is + /// answered against the primary inode space, so anything under a secondary + /// on-demand mount is rejected as "not under the mountpoint" and the user + /// has no way to re-enumerate a folder they can see (`docs/BUGS.md` B86). + /// + /// `None` when no live mount covers `abs`. + fn rooted_at(&self, abs: &Path) -> Option<(Core, PathBuf)> { + let parts = self.states.covering_parts(abs)?; + let rel = abs.strip_prefix(&parts.mountpoint).ok()?.to_path_buf(); + let mut core = self.clone(); + if !Arc::ptr_eq(&parts.state, &self.state) { + // A different inode space, so adopt its per-mount half wholesale. + // `primary` is a property of the mount, and the only mount this + // `Core` can be the primary of is the one it was built for. + core.primary = false; + core.state = parts.state; + core.notifier = parts.notifier; + core.session_live = parts.session_live; + core.size_upgrades = parts.size_upgrades; + } + Some((core, rel)) + } + /// Run `apply` against every live inode space — this daemon's primary mount /// and each on-demand fork — one lock at a time. /// @@ -3934,6 +4030,33 @@ impl Core { } } + /// Drop the cached child listing of the folder `parent`, everywhere it is + /// held — naming the folder by uid rather than by inode. + /// + /// [`Core::invalidate_parent_listing`] is the path-based sibling, and it + /// only reaches *this* mount's inode space. A remote mutation made by the + /// sync engine needs more than that, for two reasons. The folder it changed + /// is usually not resident in any inode space at the time — a mirror folder + /// has no FUSE session at all — so there is no inode to name. And the stale + /// listing that outlives the pass is not a hot-cache map but the `listed` + /// flag on the folder's DB row, which survives a daemon restart and is what + /// a later `ensure_children` trusts. + /// + /// So the flag is cleared unconditionally, and the resident inode spaces are + /// invalidated on top of it. That is what makes a mirror folder switched to + /// on-demand re-enumerate instead of serving a snapshot that predates the + /// engine's own uploads — `docs/BUGS.md` B86. + pub(crate) fn invalidate_children_of(&self, parent: &NodeUid) { + if let Err(e) = self.db.set_listed(parent, false) { + warn!(uid = %parent, error = ?e, "could not clear cached listing"); + } + self.for_each_state(|st| { + if let Some(&ino) = st.by_uid.get(parent) { + st.invalidate_listing(ino); + } + }); + } + /// Rename a file or folder to `new_name`. `rel` is mountpoint-relative. /// Mirrors the FUSE `rename` write path: rename on the remote, forget the /// node so it re-interns under its new name, and drop the parent listing so @@ -5853,7 +5976,7 @@ mod tests { ("/mnt/fork-a", &fork_a), ("/mnt/fork-b", &fork_b), ] { - registry.register( + registry.register_bare( std::path::Path::new(mountpoint), state, std::sync::Arc::new(std::sync::OnceLock::new()), @@ -5904,19 +6027,19 @@ mod tests { let nested_live = session_flag(false); let sibling_live = session_flag(true); - registry.register( + registry.register_bare( primary_path, &primary, std::sync::Arc::new(std::sync::OnceLock::new()), primary_live.clone(), ); - registry.register( + registry.register_bare( nested_path, &nested, std::sync::Arc::new(std::sync::OnceLock::new()), nested_live.clone(), ); - registry.register( + registry.register_bare( sibling_path, &sibling, std::sync::Arc::new(std::sync::OnceLock::new()), @@ -5994,6 +6117,71 @@ mod tests { assert_eq!(registry.live().len(), 2); } + /// The routing half of `docs/BUGS.md` B86: a path under a secondary mount + /// has to resolve to *that* mount's inode space and a suffix relative to + /// *its* root, or `pdfs ls` and `pdfs refresh` can only ever name the + /// primary mount and a stale on-demand folder has no escape hatch. + #[test] + fn covering_parts_route_a_path_to_the_mount_that_owns_it() { + let registry = StateRegistry::default(); + let (primary, _primary_dir) = rooted_state("primary-volume", "primary"); + let (nested, _nested_dir) = rooted_state("nested-volume", "nested"); + let primary = std::sync::Arc::new(parking_lot::Mutex::new(primary)); + let nested = std::sync::Arc::new(parking_lot::Mutex::new(nested)); + let primary_path = std::path::Path::new("/home/me/ProtonDrive"); + let nested_path = std::path::Path::new("/home/me/ProtonDrive/Device"); + let primary_upgrades = std::sync::Arc::new(parking_lot::Mutex::new(HashMap::new())); + let nested_upgrades = std::sync::Arc::new(parking_lot::Mutex::new(HashMap::new())); + + registry.register( + primary_path, + &primary, + std::sync::Arc::new(std::sync::OnceLock::new()), + session_flag(true), + primary_upgrades.clone(), + ); + registry.register( + nested_path, + &nested, + std::sync::Arc::new(std::sync::OnceLock::new()), + session_flag(true), + nested_upgrades.clone(), + ); + + let deep = nested_path.join("sub/file.txt"); + let parts = registry.covering_parts(&deep).expect("nested mount covers"); + assert_eq!(parts.mountpoint, nested_path, "the nested mount wins"); + assert!( + std::sync::Arc::ptr_eq(&parts.state, &nested), + "the request must be answered in the nested inode space" + ); + assert!( + std::sync::Arc::ptr_eq(&parts.size_upgrades, &nested_upgrades), + "and share that mount's in-flight size upgrades, not the primary's" + ); + assert_eq!( + deep.strip_prefix(&parts.mountpoint).unwrap(), + std::path::Path::new("sub/file.txt"), + "the suffix is relative to the mount that owns it, not the primary" + ); + + let shallow = primary_path.join("Documents"); + let parts = registry + .covering_parts(&shallow) + .expect("primary mount covers"); + assert!( + std::sync::Arc::ptr_eq(&parts.state, &primary), + "a path outside the nested root still routes to the primary" + ); + + assert!( + registry + .covering_parts(std::path::Path::new("/tmp/outside")) + .is_none(), + "a path under no mount routes nowhere, and the caller reports that" + ); + } + #[test] fn state_registry_resolves_uids_from_any_resident_mount() { let registry = StateRegistry::default(); @@ -6007,13 +6195,13 @@ mod tests { fork.children.insert(super::ROOT_INO, vec![fork_ino]); let primary = std::sync::Arc::new(parking_lot::Mutex::new(primary)); let fork = std::sync::Arc::new(parking_lot::Mutex::new(fork)); - registry.register( + registry.register_bare( std::path::Path::new("/mnt/primary"), &primary, std::sync::Arc::new(std::sync::OnceLock::new()), session_flag(true), ); - registry.register( + registry.register_bare( std::path::Path::new("/mnt/device"), &fork, std::sync::Arc::new(std::sync::OnceLock::new()), @@ -6041,7 +6229,7 @@ mod tests { let child_ino = state.intern(super::ROOT_INO, child_node); state.children.insert(super::ROOT_INO, vec![child_ino]); let state = std::sync::Arc::new(parking_lot::Mutex::new(state)); - registry.register( + registry.register_bare( std::path::Path::new("/mnt/device"), &state, std::sync::Arc::new(std::sync::OnceLock::new()), @@ -6104,7 +6292,7 @@ mod tests { ); let state = std::sync::Arc::new(parking_lot::Mutex::new(state)); - registry.register( + registry.register_bare( std::path::Path::new("/mnt/device"), &state, std::sync::Arc::new(std::sync::OnceLock::new()), @@ -6372,7 +6560,7 @@ mod tests { state.children.insert(revoked_ino, vec![child_ino]); let state = std::sync::Arc::new(parking_lot::Mutex::new(state)); - registry.register( + registry.register_bare( std::path::Path::new("/mnt/device"), &state, std::sync::Arc::new(std::sync::OnceLock::new()), @@ -6426,7 +6614,7 @@ mod tests { state.children.insert(cycle_b, vec![cycle_a]); let state = std::sync::Arc::new(parking_lot::Mutex::new(state)); - registry.register( + registry.register_bare( std::path::Path::new("/mnt/device"), &state, std::sync::Arc::new(std::sync::OnceLock::new()), diff --git a/crates/pdfs-fuse/src/reads.rs b/crates/pdfs-fuse/src/reads.rs index 6158291..e93382f 100644 --- a/crates/pdfs-fuse/src/reads.rs +++ b/crates/pdfs-fuse/src/reads.rs @@ -102,22 +102,38 @@ static NEXT_OPEN_ID: AtomicU64 = AtomicU64::new(0); /// Validated by the same `(mtime, size)` pair the on-disk caches use: a node /// whose tag no longer matches has its blocks dropped rather than served, so a /// new revision can never be stitched together from the old one. +/// A ring entry's identity: the node, and the plaintext range the block covers. +/// +/// The range, rather than the block index, because the index alone stopped +/// naming one range when the geometry started coming from the revision — a +/// block cached under the [`BlockGeometry::uniform`] fallback and the block at +/// the same index under the real geometry are different bytes +/// (`docs/BUGS.md` B85). A range is the same identity under either plan, so a +/// re-planned read misses rather than being served the wrong bytes. +type RingKey = (NodeUid, BlockSpan); + #[derive(Default)] pub(super) struct BlockRing { - blocks: HashMap<(NodeUid, u64), Arc>>, + blocks: HashMap>>, /// Keys least-recently-used first, for eviction. - order: VecDeque<(NodeUid, u64)>, + order: VecDeque, /// Per-node validity tag, `(mtime, size)`. tags: HashMap, bytes: u64, } impl BlockRing { - fn get(&mut self, uid: &NodeUid, mtime: i64, size: u64, idx: u64) -> Option>> { + fn get( + &mut self, + uid: &NodeUid, + mtime: i64, + size: u64, + span: BlockSpan, + ) -> Option>> { if self.tags.get(uid) != Some(&(mtime, size)) { return None; } - let key = (uid.clone(), idx); + let key = (uid.clone(), span); let hit = self.blocks.get(&key).cloned()?; // Move to the young end: a block being read repeatedly (the whole reason // this exists) must not age out under a sequential read passing through. @@ -128,14 +144,21 @@ impl BlockRing { Some(hit) } - fn insert(&mut self, uid: &NodeUid, mtime: i64, size: u64, idx: u64, bytes: Arc>) { + fn insert( + &mut self, + uid: &NodeUid, + mtime: i64, + size: u64, + span: BlockSpan, + bytes: Arc>, + ) { if self.tags.get(uid) != Some(&(mtime, size)) { // Revision changed under us (or first sight): anything held for this // node describes the old one. self.drop_node(uid); self.tags.insert(uid.clone(), (mtime, size)); } - let key = (uid.clone(), idx); + let key = (uid.clone(), span); if self.blocks.contains_key(&key) { return; } @@ -176,8 +199,10 @@ impl BlockRing { /// The result of fetching one block, shared between everyone who wanted it. type BlockResult = Result>, Errno>; -/// Identifies a block *of a revision*: node, validity tag, index. -type BlockKey = (NodeUid, i64, u64, u64); +/// Identifies a block *of a revision*: node, validity tag, and the plaintext +/// range the block covers — see [`RingKey`] for why it is the range and not the +/// index. +type BlockKey = (NodeUid, i64, u64, BlockSpan); /// A claim on a block key: who owns the fetch, and where its result is published. type Claim = (u64, tokio::sync::watch::Receiver>); @@ -417,11 +442,30 @@ impl Core { self.prefetch.streams.lock().remove(uid); } + /// How this revision's plaintext is divided into content blocks. + /// + /// The revision's own geometry when a previous read has learned and recorded + /// it, and the [`BLOCK_SIZE`] assumption when it has not — which is every + /// first read, because learning it means opening a `RevisionReader`, and + /// opening one per read is precisely the per-file key derivation the block + /// cache exists to avoid (bugs.md B12, B85). + /// + /// Being wrong here is not a correctness problem: `read_at` plans over the + /// real sizes and clamps to what was asked for either way. It costs a + /// straddling fetch and a cached block the next read will not ask for, and + /// both settle once the geometry is known. + fn block_geometry(&self, uid: &NodeUid, mtime: i64, fsize: u64) -> BlockGeometry { + self.cache + .block_geometry(uid, mtime, fsize) + .unwrap_or_else(|| BlockGeometry::uniform(fsize)) + } + /// Serve bytes `[offset, offset + len)` of `uid`'s active revision, hitting /// the on-disk caches before the network: a whole-file blob (pinned files) - /// first, then the block cache — fetching only the [`BLOCK_SIZE`]-aligned - /// blocks that overlap the request and caching each. `mtime`/`fsize` validate - /// both caches. Network I/O runs without any lock held. + /// first, then the block cache — fetching only the blocks that overlap the + /// request, on the revision's own boundaries ([`Core::block_geometry`]), and + /// caching each. `mtime`/`fsize` validate both caches. Network I/O runs + /// without any lock held. pub(super) fn read_range( &self, uid: &NodeUid, @@ -500,7 +544,7 @@ impl Core { Ok(out) } - /// Fetch block `bidx`, joining a fetch already in flight for it rather than + /// Fetch the block at `span`, joining a fetch already in flight rather than /// issuing a second one, and publish the result to everyone who joined. /// /// See [`BlockFlight`]. A follower whose leader disappears (a panic, or a @@ -512,10 +556,10 @@ impl Core { uid: &NodeUid, mtime: i64, fsize: u64, - bidx: u64, + span: BlockSpan, cache_blocks: bool, ) -> BlockResult { - let key: BlockKey = (uid.clone(), mtime, fsize, bidx); + let key: BlockKey = (uid.clone(), mtime, fsize, span); let claim = { let mut inflight = self.block_flight.inflight.lock(); match inflight.get(&key) { @@ -547,12 +591,12 @@ impl Core { } } return self - .read_block(reader, uid, mtime, fsize, bidx, cache_blocks) + .read_block(reader, uid, mtime, fsize, span, cache_blocks) .await; } }; let result = self - .read_block(reader, uid, mtime, fsize, bidx, cache_blocks) + .read_block(reader, uid, mtime, fsize, span, cache_blocks) .await; // Retire the claim before publishing, so a racer that wakes and re-enters // finds no stale entry to join. @@ -578,11 +622,9 @@ impl Core { uid: &NodeUid, mtime: i64, fsize: u64, - bidx: u64, + span: BlockSpan, cache_blocks: bool, ) -> BlockResult { - let bstart = bidx * BLOCK_SIZE; - let blen = block_len(fsize, bidx); // The reader derives every block boundary from the revision's recorded // block sizes. If they do not add up to the size the node reports, this // file's boundaries are wrong, and the failure is not confined to short @@ -591,25 +633,33 @@ impl Core { // kind takes the file off the range path entirely (bugs.md B84). let bytes = if reader.size() != fsize { warn!( - %uid, bidx, fsize, + %uid, bidx = span.idx, fsize, reader_size = reader.size(), reader_blocks = reader.block_sizes().len(), "the revision's block table disagrees with the node size" ); - self.repair_block(uid, mtime, fsize, bidx).await? + self.repair_block(uid, mtime, fsize, span).await? } else { - let bytes = reader.read_at(bstart, blen).await.map_err(|e| { - warn!(%uid, bstart, blen, error = %e, "block read failed"); + // Learned here rather than on a path of its own: this is where a + // reader is open anyway, and recording the real geometry is what + // lets the *next* read plan its blocks on the revision's own + // boundaries instead of the 4 MiB fallback (bugs.md B85). + if cache_blocks { + self.cache + .store_block_geometry(uid, mtime, fsize, reader.block_sizes()); + } + let bytes = reader.read_at(span.start, span.len).await.map_err(|e| { + warn!(%uid, start = span.start, len = span.len, error = %e, "block read failed"); Errno::EIO })?; - if bytes.len() as u64 == blen { + if bytes.len() as u64 == span.len { bytes } else { warn!( - %uid, bidx, blen, got = bytes.len(), fsize, - "block came back shorter than the file's size implies" + %uid, bidx = span.idx, blen = span.len, got = bytes.len(), fsize, + "block came back shorter than the range asked for" ); - self.repair_block(uid, mtime, fsize, bidx).await? + self.repair_block(uid, mtime, fsize, span).await? } }; let bytes = Arc::new(bytes); @@ -618,16 +668,16 @@ impl Core { let uid = uid.clone(); let bytes = bytes.clone(); tokio::task::spawn_blocking(move || { - let _ = cache.store_block(&uid, mtime, fsize, bidx, &bytes); + let _ = cache.store_block(&uid, mtime, fsize, span, &bytes); }); } self.block_ring .lock() - .insert(uid, mtime, fsize, bidx, bytes.clone()); + .insert(uid, mtime, fsize, span, bytes.clone()); Ok(bytes) } - /// Recover block `bidx` of a file whose per-block reads cannot be trusted, + /// Recover the block at `span` of a file whose per-block reads cannot be /// by fetching the whole revision through the manifest-verified download and /// serving the block out of that. /// @@ -648,10 +698,9 @@ impl Core { uid: &NodeUid, mtime: i64, fsize: u64, - bidx: u64, + span: BlockSpan, ) -> Result, Errno> { - let blen = block_len(fsize, bidx); - let bstart = bidx * BLOCK_SIZE; + let (bstart, blen) = (span.start, span.len); if fsize > REPAIR_MAX { warn!( %uid, fsize, @@ -683,11 +732,11 @@ impl Core { } match self.cache.read_range(uid, mtime, fsize, bstart, blen) { Some(bytes) if bytes.len() as u64 == blen => { - info!(%uid, bidx, "served a short block from a repaired whole-file download"); + info!(%uid, bidx = span.idx, "served a short block from a repaired whole-file download"); Ok(bytes) } _ => { - error!(%uid, bidx, fsize, "repairing a short block did not produce the block"); + error!(%uid, bidx = span.idx, fsize, "repairing a short block did not produce the block"); Err(Errno::EIO) } } @@ -776,17 +825,19 @@ impl Core { if depth == 0 { return; } - let last_block = fsize.saturating_sub(1) / BLOCK_SIZE; - let wanted: Vec = ((last + 1)..=(last + depth).min(last_block)) - .filter(|&bidx| { + let geometry = self.block_geometry(uid, mtime, fsize); + let last_block = geometry.blocks().saturating_sub(1); + let wanted: Vec = ((last + 1)..=(last + depth).min(last_block)) + .filter_map(|bidx| geometry.span(bidx)) + .filter(|&span| { // Blocks already in memory need nothing; blocks already on disk // are cheap enough to serve that spending a round-trip to move // them into the ring would be the wrong trade. self.block_ring .lock() - .get(uid, mtime, fsize, bidx) + .get(uid, mtime, fsize, span) .is_none() - && !(cache_blocks && self.cache.has_block(uid, mtime, fsize, bidx)) + && !(cache_blocks && self.cache.has_block(uid, mtime, fsize, span)) }) .collect(); if wanted.is_empty() { @@ -803,7 +854,7 @@ impl Core { // round-trips, and the reader catches up with the window before it // finishes. let mut set = tokio::task::JoinSet::new(); - for bidx in wanted { + for span in wanted { let Ok(permit) = core.prefetch_budget.clone().try_acquire_owned() else { break; // the budget is spent; the demand path can have it }; @@ -811,10 +862,10 @@ impl Core { set.spawn(async move { let _permit = permit; if let Err(error) = core - .fetch_block(&reader, &uid, mtime, fsize, bidx, cache_blocks) + .fetch_block(&reader, &uid, mtime, fsize, span, cache_blocks) .await { - debug!(%uid, bidx, ?error, "prefetch failed"); + debug!(%uid, bidx = span.idx, ?error, "prefetch failed"); } }); } @@ -844,8 +895,13 @@ impl Core { } let end = offset.saturating_add(len).min(fsize); let mut out = Vec::with_capacity((end - offset) as usize); - let first = offset / BLOCK_SIZE; - let last = (end - 1) / BLOCK_SIZE; + // On the revision's own boundaries where they are known, and on the + // 4 MiB fallback where they are not (bugs.md B85). + let wanted = self.block_geometry(uid, mtime, fsize).spans(offset, end); + let (Some(&head), Some(&tail)) = (wanted.first(), wanted.last()) else { + return Ok(out); + }; + let (first, last) = (head.idx, tail.idx); // Collect the blocks overlapping the request, serving any already cached // and fetching the rest concurrently. A multi-block read (e.g. a media @@ -853,31 +909,31 @@ impl Core { // would otherwise stall on each block round-trip in turn; downloading the // misses in parallel saturates the connection and bounds latency at the // slowest single block instead of their sum. - let mut blocks: Vec>>> = Vec::with_capacity((last - first + 1) as usize); - let mut misses: Vec = Vec::new(); - for bidx in first..=last { + let mut blocks: Vec>>> = Vec::with_capacity(wanted.len()); + let mut misses: Vec = Vec::new(); + for &span in &wanted { // The in-memory ring comes first on both paths: a streaming read has // no other cache at all, and a cached one would otherwise pay two // opens, a JSON parse and a 4 MiB read+alloc for every 128 KiB the // kernel asks for (audit T3). - let hit = self.block_ring.lock().get(uid, mtime, fsize, bidx); + let hit = self.block_ring.lock().get(uid, mtime, fsize, span); let hit = hit.or_else(|| { if !cache_blocks { return None; } - let bytes = Arc::new(self.cache.cached_block(uid, mtime, fsize, bidx)?); + let bytes = Arc::new(self.cache.cached_block(uid, mtime, fsize, span)?); // Promote it, so the other ~31 reads of this block are answered // from memory. self.block_ring .lock() - .insert(uid, mtime, fsize, bidx, bytes.clone()); + .insert(uid, mtime, fsize, span, bytes.clone()); Some(bytes) }); match hit { Some(b) => blocks.push(Some(b)), None => { blocks.push(None); - misses.push(bidx); + misses.push(span); } } } @@ -890,12 +946,12 @@ impl Core { let reader = self.revision_reader(uid, mtime, fsize).await?; let mut set = tokio::task::JoinSet::new(); - for &bidx in &misses { + for &span in &misses { let (core, reader, uid) = (self.clone(), reader.clone(), uid.clone()); set.spawn(async move { - core.fetch_block(&reader, &uid, mtime, fsize, bidx, cache_blocks) + core.fetch_block(&reader, &uid, mtime, fsize, span, cache_blocks) .await - .map(|bytes| (bidx, bytes)) + .map(|bytes| (span.idx, bytes)) }); } let mut out = Vec::with_capacity(misses.len()); @@ -913,18 +969,20 @@ impl Core { // ahead of a sequential reader instead of stalling once it catches up. self.prefetch_ahead(uid, mtime, fsize, first, last, cache_blocks); - for (i, block) in blocks.into_iter().enumerate() { - let bidx = first + i as u64; - let bstart = bidx * BLOCK_SIZE; + for (span, block) in wanted.into_iter().zip(blocks) { + let bstart = span.start; // Every slot is populated: cache hits up front, misses by the fetch above. let block = block.expect("block fetched or cached"); - // Both sources guarantee the length the file's size implies — the - // cache validates it on read, the network path in `read_block` — so - // this refuses rather than quietly contributing fewer bytes than the - // range covers, which is what turned a bad block into a truncated - // file (bugs.md B84). - if block.len() as u64 != block_len(fsize, bidx) { - error!(%uid, bidx, len = block.len(), fsize, "assembling a read hit a short block"); + // Both sources guarantee the length the geometry says the block has + // — the cache validates it on read, the network path in `read_block` + // — so this refuses rather than quietly contributing fewer bytes + // than the range covers, which is what turned a bad block into a + // truncated file (bugs.md B84). + if block.len() as u64 != span.len { + error!( + %uid, bidx = span.idx, len = block.len(), expected = span.len, fsize, + "assembling a read hit a short block" + ); return Err(Errno::EIO); } let s = (offset.max(bstart) - bstart) as usize; @@ -1016,13 +1074,13 @@ impl Core { return None; } let mut out = Vec::with_capacity((end - offset) as usize); - for bidx in (offset / BLOCK_SIZE)..=((end - 1) / BLOCK_SIZE) { - let ring = self.block_ring.lock().get(uid, mtime, fsize, bidx); + for span in self.block_geometry(uid, mtime, fsize).spans(offset, end) { + let ring = self.block_ring.lock().get(uid, mtime, fsize, span); let block = match ring { Some(bytes) => bytes, - None => Arc::new(self.cache.cached_block(uid, mtime, fsize, bidx)?), + None => Arc::new(self.cache.cached_block(uid, mtime, fsize, span)?), }; - let bstart = bidx * BLOCK_SIZE; + let bstart = span.start; let s = (offset.max(bstart) - bstart) as usize; let e = (end.min(bstart + block.len() as u64) - bstart) as usize; if s >= e { @@ -1142,31 +1200,42 @@ mod block_ring_tests { RING_BYTES / BLOCK_SIZE } + /// Block `idx` under the uniform geometry — the shape these tests are + /// written against, since what they exercise is tagging and eviction rather + /// than block layout. + fn span(idx: u64) -> BlockSpan { + BlockSpan { + idx, + start: idx * BLOCK_SIZE, + len: BLOCK_SIZE, + } + } + #[test] fn a_stored_block_is_served_back() { let mut ring = BlockRing::default(); let node = uid("a"); - ring.insert(&node, 7, 100, 3, block(0xab)); - assert_eq!(ring.get(&node, 7, 100, 3).map(|b| b[0]), Some(0xab)); + ring.insert(&node, 7, 100, span(3), block(0xab)); + assert_eq!(ring.get(&node, 7, 100, span(3)).map(|b| b[0]), Some(0xab)); } #[test] fn a_block_of_another_revision_is_never_served() { let mut ring = BlockRing::default(); let node = uid("a"); - ring.insert(&node, 7, 100, 0, block(1)); + ring.insert(&node, 7, 100, span(0), block(1)); // Same node, newer revision: the tag moved, so the old bytes go. - assert!(ring.get(&node, 8, 100, 0).is_none()); - assert!(ring.get(&node, 7, 101, 0).is_none()); + assert!(ring.get(&node, 8, 100, span(0)).is_none()); + assert!(ring.get(&node, 7, 101, span(0)).is_none()); } #[test] fn a_new_revision_drops_what_the_old_one_left() { let mut ring = BlockRing::default(); let node = uid("a"); - ring.insert(&node, 7, 100, 0, block(1)); - ring.insert(&node, 8, 100, 0, block(2)); - assert_eq!(ring.get(&node, 8, 100, 0).map(|b| b[0]), Some(2)); + ring.insert(&node, 7, 100, span(0), block(1)); + ring.insert(&node, 8, 100, span(0), block(2)); + assert_eq!(ring.get(&node, 8, 100, span(0)).map(|b| b[0]), Some(2)); assert_eq!(ring.bytes, BLOCK_SIZE); } @@ -1175,13 +1244,13 @@ mod block_ring_tests { let mut ring = BlockRing::default(); let node = uid("a"); for idx in 0..capacity() { - ring.insert(&node, 1, u64::MAX, idx, block(idx as u8)); + ring.insert(&node, 1, u64::MAX, span(idx), block(idx as u8)); } // Re-reading block 0 makes block 1 the oldest instead. - assert!(ring.get(&node, 1, u64::MAX, 0).is_some()); - ring.insert(&node, 1, u64::MAX, capacity(), block(0xff)); - assert!(ring.get(&node, 1, u64::MAX, 0).is_some()); - assert!(ring.get(&node, 1, u64::MAX, 1).is_none()); + assert!(ring.get(&node, 1, u64::MAX, span(0)).is_some()); + ring.insert(&node, 1, u64::MAX, span(capacity()), block(0xff)); + assert!(ring.get(&node, 1, u64::MAX, span(0)).is_some()); + assert!(ring.get(&node, 1, u64::MAX, span(1)).is_none()); assert_eq!(ring.bytes, RING_BYTES); } @@ -1189,9 +1258,9 @@ mod block_ring_tests { fn a_node_whose_last_block_ages_out_keeps_no_tag() { let mut ring = BlockRing::default(); let (old, new) = (uid("old"), uid("new")); - ring.insert(&old, 1, u64::MAX, 0, block(1)); + ring.insert(&old, 1, u64::MAX, span(0), block(1)); for idx in 0..capacity() { - ring.insert(&new, 1, u64::MAX, idx, block(2)); + ring.insert(&new, 1, u64::MAX, span(idx), block(2)); } assert!(!ring.tags.contains_key(&old)); assert!(ring.tags.contains_key(&new)); @@ -1201,11 +1270,11 @@ mod block_ring_tests { fn dropping_a_node_frees_only_its_own_bytes() { let mut ring = BlockRing::default(); let (a, b) = (uid("a"), uid("b")); - ring.insert(&a, 1, 100, 0, block(1)); - ring.insert(&b, 1, 100, 0, block(2)); + ring.insert(&a, 1, 100, span(0), block(1)); + ring.insert(&b, 1, 100, span(0), block(2)); ring.drop_node(&a); - assert!(ring.get(&a, 1, 100, 0).is_none()); - assert_eq!(ring.get(&b, 1, 100, 0).map(|x| x[0]), Some(2)); + assert!(ring.get(&a, 1, 100, span(0)).is_none()); + assert_eq!(ring.get(&b, 1, 100, span(0)).map(|x| x[0]), Some(2)); assert_eq!(ring.bytes, BLOCK_SIZE); } } diff --git a/crates/pdfs-fuse/src/sync.rs b/crates/pdfs-fuse/src/sync.rs index d673f76..fa12058 100644 --- a/crates/pdfs-fuse/src/sync.rs +++ b/crates/pdfs-fuse/src/sync.rs @@ -525,6 +525,9 @@ impl Core { continue; } trashed.push(rel.clone()); + if let Some(parent) = parent_uid_of(rel, &remote_dirs, &baseline) { + self.invalidate_children_of(&parent); + } let _ = self.db.sync_entry_remove(folder_id, rel); self.log_activity( ActivityKind::Trash, @@ -777,6 +780,9 @@ impl Core { outcome.errors += 1; continue; } + if let Some(parent) = parent_uid_of(rel, &remote_dirs, &baseline) { + self.invalidate_children_of(&parent); + } let _ = self.db.sync_entry_remove(folder_id, rel); self.log_activity( ActivityKind::Trash, @@ -832,6 +838,9 @@ impl Core { outcome.errors += 1; continue; } + if let Some(parent) = parent_uid_of(&rel, &remote_dirs, &baseline) { + self.invalidate_children_of(&parent); + } let _ = self.db.sync_entry_remove(folder_id, &rel); self.log_activity( ActivityKind::Trash, @@ -857,70 +866,9 @@ impl Core { writing: &HashSet, out: &mut HashMap, ) -> Result<(), String> { - let entries = std::fs::read_dir(dir).map_err(|e| format!("read {}: {e}", dir.display()))?; - for entry in entries { - let entry = entry.map_err(|e| format!("read entry in {}: {e}", dir.display()))?; - let path = entry.path(); - let meta = std::fs::symlink_metadata(&path) - .map_err(|e| format!("stat {}: {e}", path.display()))?; - let stripped = path.strip_prefix(root).map_err(|e| { - format!( - "local path {} escaped sync root {}: {e}", - path.display(), - root.display() - ) - })?; - let rel = stripped.to_str().ok_or_else(|| { - format!( - "local path {} is not valid UTF-8; refusing a destructive sync pass", - path.display() - ) - })?; - // Ignore our own in-flight download temp files. - if rel.contains(".pdfs-tmp-") { - continue; - } - // Ignored paths are dropped here as well as from the classification - // union — not for correctness (the union filter is what makes this - // safe) but so an ignored `node_modules/` is never descended into or - // counted against scan progress. - if rules.is_ignored(rel, meta.is_dir()) { - continue; - } - if meta.is_dir() { - out.insert( - rel.to_string(), - LocalItem { - is_dir: true, - mtime: 0, - mtime_ns: None, - size: 0, - open_for_write: false, - }, - ); - self.progress_scanned(folder_id, base_name(rel)); - self.walk_local(folder_id, root, &path, rules, writing, out)?; - } else if meta.is_file() { - // Check whether any process holds this file open for writing. - // The canonical path is used because /proc/*/fd links resolve - // to canonical targets. - let ofw = std::fs::canonicalize(&path) - .map(|canon| writing.contains(&canon)) - .unwrap_or(false); - out.insert( - rel.to_string(), - LocalItem { - is_dir: false, - mtime: system_mtime(&meta), - mtime_ns: system_mtime_ns(&meta), - size: meta.len() as i64, - open_for_write: ofw, - }, - ); - self.progress_scanned(folder_id, base_name(rel)); - } - } - Ok(()) + walk_local_tree(root, dir, rules, writing, out, &mut |rel| { + self.progress_scanned(folder_id, base_name(rel)) + }) } /// Map every remote folder under `folder` to its uid by rel path, ignoring @@ -1186,6 +1134,9 @@ impl Core { .create_folder(parent, base_name(rel), Some(now_secs())) .await .map_err(|e| format!("create remote folder {rel}: {e}"))?; + // The parent has gained a child that no cached listing of it + // knows about (B86). + self.invalidate_children_of(parent); self.baseline_dir(folder_id, rel, &uid)?; Ok(Applied::Dir(rel.clone(), uid)) } @@ -1295,6 +1246,10 @@ impl Core { ) .await .map_err(|e| format!("upload {rel}: {e}"))?; + // The parent has gained a child that no cached listing of it knows + // about. Without this, a folder later switched to on-demand serves a + // snapshot that predates this upload (B86). + self.invalidate_children_of(parent); warn_if_torn(rel, &path, streamed); self.record_file_baseline(folder_id, rel, Some(streamed), &uid) .await @@ -1546,6 +1501,95 @@ fn removed_locally(result: std::io::Result<()>) -> std::io::Result<()> { } } +/// The body of [`Core::walk_local`], without a `Core`: recursively collect the +/// local tree under `dir` into `out`, reporting each entry to `scanned`. +/// +/// **Every failure here is fatal to the pass, deliberately.** A `read_dir` that +/// fails, an entry that cannot be stat-ed, a name that is not UTF-8 — each of +/// them would, if skipped, leave `out` describing a *smaller* tree than the one +/// on disk. Reconciliation reads a path present in the baseline and absent from +/// the scan as "the user deleted this", and propagates that to Drive. So an +/// unreadable subdirectory would trash its remote counterpart, and the deeper +/// the failure, the more it takes with it (`docs/BUGS.md` B55). Refusing the +/// whole pass keeps the previous baseline and leaves the remote alone; the next +/// pass retries once the transient condition clears. +/// +/// Split out from the method so exactly that can be tested against a real tree +/// with a real unreadable directory in it, which needs no `Core` at all. +fn walk_local_tree( + root: &Path, + dir: &Path, + rules: &IgnoreRules, + writing: &HashSet, + out: &mut HashMap, + scanned: &mut dyn FnMut(&str), +) -> Result<(), String> { + let entries = std::fs::read_dir(dir).map_err(|e| format!("read {}: {e}", dir.display()))?; + for entry in entries { + let entry = entry.map_err(|e| format!("read entry in {}: {e}", dir.display()))?; + let path = entry.path(); + let meta = std::fs::symlink_metadata(&path) + .map_err(|e| format!("stat {}: {e}", path.display()))?; + let stripped = path.strip_prefix(root).map_err(|e| { + format!( + "local path {} escaped sync root {}: {e}", + path.display(), + root.display() + ) + })?; + let rel = stripped.to_str().ok_or_else(|| { + format!( + "local path {} is not valid UTF-8; refusing a destructive sync pass", + path.display() + ) + })?; + // Ignore our own in-flight download temp files. + if rel.contains(".pdfs-tmp-") { + continue; + } + // Ignored paths are dropped here as well as from the classification + // union — not for correctness (the union filter is what makes this + // safe) but so an ignored `node_modules/` is never descended into or + // counted against scan progress. + if rules.is_ignored(rel, meta.is_dir()) { + continue; + } + if meta.is_dir() { + out.insert( + rel.to_string(), + LocalItem { + is_dir: true, + mtime: 0, + mtime_ns: None, + size: 0, + open_for_write: false, + }, + ); + scanned(rel); + walk_local_tree(root, &path, rules, writing, out, scanned)?; + } else if meta.is_file() { + // Check whether any process holds this file open for writing. + // The canonical path is used because /proc/*/fd links resolve + // to canonical targets. + let ofw = std::fs::canonicalize(&path) + .map(|canon| writing.contains(&canon)) + .unwrap_or(false); + out.insert( + rel.to_string(), + LocalItem { + is_dir: false, + mtime: system_mtime(&meta), + mtime_ns: system_mtime_ns(&meta), + size: meta.len() as i64, + open_for_write: ofw, + }, + ); + scanned(rel); + } + } + Ok(()) +} + /// Publish a conflict copy without overwriting an earlier conflict bearing the /// same timestamp. The original is removed only after the new copy is durable. fn preserve_conflict_copy(path: &Path, stamp: i64) -> std::io::Result { @@ -1579,6 +1623,26 @@ fn preserve_conflict_copy(path: &Path, stamp: i64) -> std::io::Result { unreachable!("u32 conflict suffix space exhausted") } +/// The remote uid of `rel`'s parent folder: from the pass's folder-uid map +/// first, and from the baseline second for a parent the map never covered. +/// +/// Used to name the folder whose cached listing a remote mutation has just +/// invalidated (B86). `None` is not an error — it only means this pass cannot +/// say which folder to invalidate, and the caller skips it. +fn parent_uid_of( + rel: &str, + remote_dirs: &HashMap, + baseline: &HashMap, +) -> Option { + let parent = parent_rel(rel); + remote_dirs.get(parent).cloned().or_else(|| { + baseline + .get(parent) + .and_then(|e| e.remote_uid.as_deref()) + .and_then(parse_uid) + }) +} + fn sync_parent_dir(path: &Path) -> std::io::Result<()> { match path.parent() { Some(parent) => std::fs::File::open(parent)?.sync_all(), @@ -1925,6 +1989,44 @@ mod tests { assert!(guard_local_wipe(&base, &fresh).is_err()); } + /// B54's remaining case, end to end: a folder whose local root is still + /// *readable* but is no longer the tree it was — an empty directory left + /// where a mount used to be, or a different device mounted over it. + /// + /// Nothing fails here. The scan succeeds and honestly reports an empty + /// directory, so the only thing standing between the user and a trashed + /// remote folder is the guard, and it has to fire for a one-file baseline + /// exactly as it does for a large one. + #[test] + fn a_replaced_mountpoint_trips_the_guard_at_every_baseline_size() { + let root = sync_test_dir("wipe-replaced-mount"); + let rules = rules_for(&[]); + let writing = HashSet::new(); + let mut local: HashMap = HashMap::new(); + let mut progress = no_progress(); + walk_local_tree(&root, &root, &rules, &writing, &mut local, &mut progress).unwrap(); + assert!(local.is_empty(), "the replacement root is genuinely empty"); + + for size in 1..=3 { + let baseline: HashMap = + (0..size).map(|i| (format!("file{i}.txt"), ())).collect(); + assert!( + guard_local_wipe(&filter_baseline(&baseline, &rules), &local).is_err(), + "a {size}-path baseline against an empty root is not a deletion" + ); + } + + // And the guard steps aside the moment the real tree is back, so a + // remounted folder resumes rather than staying wedged. + std::fs::write(root.join("file0.txt"), b"back").unwrap(); + let mut local: HashMap = HashMap::new(); + walk_local_tree(&root, &root, &rules, &writing, &mut local, &mut progress).unwrap(); + let baseline: HashMap = [("file0.txt".to_string(), ())].into_iter().collect(); + assert!(guard_local_wipe(&filter_baseline(&baseline, &rules), &local).is_ok()); + + std::fs::remove_dir_all(&root).unwrap(); + } + /// A restored folder (features.md 5.2) starts with an empty baseline against /// an empty local directory and a full remote. That must reconcile as /// "download everything" rather than tripping the wipe guard — otherwise @@ -2126,6 +2228,204 @@ mod tests { std::fs::remove_dir_all(dir).unwrap(); } + /// B49's remaining cases. Preservation is the step that makes a download + /// safe to publish, so every way it can fail has to leave the local file + /// where it was — the download is aborted, not the file replaced. + #[test] + fn conflict_preservation_failures_all_keep_the_local_file() { + use std::os::unix::fs::PermissionsExt as _; + + // An unwritable directory: the conflict copy cannot be created at all. + let dir = sync_test_dir("conflict-eacces"); + let source = dir.join("notes.txt"); + std::fs::write(&source, b"the only copy").unwrap(); + std::fs::set_permissions(&dir, std::fs::Permissions::from_mode(0o500)).unwrap(); + let denied = preserve_conflict_copy(&source, 1700); + std::fs::set_permissions(&dir, std::fs::Permissions::from_mode(0o700)).unwrap(); + assert!(denied.is_err(), "a read-only directory must fail the copy"); + assert_eq!( + std::fs::read(&source).unwrap(), + b"the only copy", + "the local file survives byte-identical" + ); + std::fs::remove_dir_all(&dir).unwrap(); + + // A source that vanished between planning and preservation. The pass + // must report it rather than publish the download over the gap. + let dir = sync_test_dir("conflict-enoent"); + assert!(preserve_conflict_copy(&dir.join("gone.txt"), 1700).is_err()); + std::fs::remove_dir_all(&dir).unwrap(); + + // A write error partway through the copy — induced by making the + // destination name a directory, so `create_new` fails as something + // other than `AlreadyExists`. + let dir = sync_test_dir("conflict-eisdir"); + let source = dir.join("notes.txt"); + std::fs::write(&source, b"still the only copy").unwrap(); + std::fs::create_dir(conflict_path(&source, 1700)).unwrap(); + // Suffix 0 is taken by a directory; `create_new` on it reports + // `AlreadyExists`, so preservation moves on to suffix 1 and succeeds. + // What must not happen is the source being removed without a copy. + let preserved = preserve_conflict_copy(&source, 1700).unwrap(); + assert!(preserved.is_file()); + assert_eq!(std::fs::read(&preserved).unwrap(), b"still the only copy"); + assert!(!source.exists()); + std::fs::remove_dir_all(&dir).unwrap(); + } + + /// A conflict copy keeps the bytes *and* the metadata that identifies them, + /// so the next pass uploads the user's version rather than a file that looks + /// freshly written. + #[test] + fn a_conflict_copy_is_byte_and_mtime_identical() { + use std::os::unix::fs::PermissionsExt as _; + + let dir = sync_test_dir("conflict-identity"); + let source = dir.join("notes.txt"); + std::fs::write(&source, b"local edit that must survive").unwrap(); + let before = std::fs::metadata(&source).unwrap(); + + let preserved = preserve_conflict_copy(&source, 1700).unwrap(); + let after = std::fs::metadata(&preserved).unwrap(); + + assert_eq!( + std::fs::read(&preserved).unwrap(), + b"local edit that must survive" + ); + assert_eq!(after.len(), before.len()); + assert_eq!(after.modified().unwrap(), before.modified().unwrap()); + assert_eq!(after.permissions().mode(), before.permissions().mode()); + std::fs::remove_dir_all(&dir).unwrap(); + } + + // ---- B55: an incomplete local scan is never a set of deletions --------- + + fn no_progress() -> impl FnMut(&str) { + |_: &str| {} + } + + /// A directory the pass cannot read must fail the pass. Omitting it would + /// hand reconciliation a baseline-minus-subtree, which it reads as the user + /// having deleted that subtree and propagates to Drive (B55). + #[test] + fn an_unreadable_subdirectory_fails_the_scan_instead_of_shrinking_it() { + use std::os::unix::fs::PermissionsExt as _; + + let root = sync_test_dir("scan-eacces"); + std::fs::write(root.join("keep.txt"), b"a").unwrap(); + let locked = root.join("locked"); + std::fs::create_dir(&locked).unwrap(); + std::fs::write(locked.join("inner.txt"), b"b").unwrap(); + std::fs::set_permissions(&locked, std::fs::Permissions::from_mode(0o000)).unwrap(); + + let rules = rules_for(&[]); + let writing = HashSet::new(); + let mut out = HashMap::new(); + let mut progress = no_progress(); + let result = walk_local_tree(&root, &root, &rules, &writing, &mut out, &mut progress); + + std::fs::set_permissions(&locked, std::fs::Permissions::from_mode(0o700)).unwrap(); + std::fs::remove_dir_all(&root).unwrap(); + + let error = result.expect_err("an unreadable directory must abort the pass"); + assert!( + error.contains("locked"), + "the error should name the subtree it could not read: {error}" + ); + } + + /// The same rule one level up: if the sync root itself cannot be read, the + /// pass sees nothing, and "nothing" must never mean "delete everything". + #[test] + fn an_unreadable_root_fails_the_scan() { + use std::os::unix::fs::PermissionsExt as _; + + let root = sync_test_dir("scan-root-eacces"); + std::fs::write(root.join("keep.txt"), b"a").unwrap(); + std::fs::set_permissions(&root, std::fs::Permissions::from_mode(0o000)).unwrap(); + + let rules = rules_for(&[]); + let writing = HashSet::new(); + let mut out = HashMap::new(); + let mut progress = no_progress(); + let result = walk_local_tree(&root, &root, &rules, &writing, &mut out, &mut progress); + + std::fs::set_permissions(&root, std::fs::Permissions::from_mode(0o700)).unwrap(); + std::fs::remove_dir_all(&root).unwrap(); + + assert!(result.is_err()); + assert!(out.is_empty()); + } + + /// A root that is gone entirely — an unmounted device folder, a removed + /// directory — is the sharpest version of the same case: every baseline path + /// is "missing", and the scan must refuse rather than report an empty tree. + #[test] + fn an_absent_root_fails_the_scan() { + let root = sync_test_dir("scan-absent"); + std::fs::remove_dir_all(&root).unwrap(); + + let rules = rules_for(&[]); + let writing = HashSet::new(); + let mut out = HashMap::new(); + let mut progress = no_progress(); + assert!(walk_local_tree(&root, &root, &rules, &writing, &mut out, &mut progress).is_err()); + assert!(out.is_empty()); + } + + /// A name that is not valid UTF-8 cannot be keyed into the scan, so it + /// cannot be compared against the baseline either. Refusing the pass is the + /// only safe reading: skipping it would classify the file as deleted. + #[test] + fn a_non_utf8_name_fails_the_scan() { + use std::ffi::OsStr; + use std::os::unix::ffi::OsStrExt as _; + + let root = sync_test_dir("scan-non-utf8"); + std::fs::write(root.join(OsStr::from_bytes(b"bad\xffname")), b"x").unwrap(); + + let rules = rules_for(&[]); + let writing = HashSet::new(); + let mut out = HashMap::new(); + let mut progress = no_progress(); + let result = walk_local_tree(&root, &root, &rules, &writing, &mut out, &mut progress); + std::fs::remove_dir_all(&root).unwrap(); + + let error = result.expect_err("a name the scan cannot key must abort the pass"); + assert!(error.contains("UTF-8"), "{error}"); + } + + /// The complement: a scan that *can* read everything reports everything, so + /// the strictness above does not come at the cost of a working pass. A + /// symlink is skipped by design rather than by failure, and a dangling one + /// stats fine through `symlink_metadata`. + #[test] + fn a_readable_tree_scans_completely() { + let root = sync_test_dir("scan-complete"); + std::fs::write(root.join("a.txt"), b"aa").unwrap(); + std::fs::create_dir(root.join("sub")).unwrap(); + std::fs::write(root.join("sub/b.txt"), b"bbb").unwrap(); + std::os::unix::fs::symlink("nowhere", root.join("dangling")).unwrap(); + + let rules = rules_for(&[]); + let writing = HashSet::new(); + let mut out = HashMap::new(); + let mut seen: Vec = Vec::new(); + let result = walk_local_tree(&root, &root, &rules, &writing, &mut out, &mut |rel| { + seen.push(rel.to_string()) + }); + std::fs::remove_dir_all(&root).unwrap(); + + result.unwrap(); + let mut keys: Vec<&str> = out.keys().map(String::as_str).collect(); + keys.sort_unstable(); + assert_eq!(keys, vec!["a.txt", "sub", "sub/b.txt"]); + assert_eq!(out["a.txt"].size, 2); + assert_eq!(out["sub/b.txt"].size, 3); + assert!(out["sub"].is_dir); + assert_eq!(seen.len(), 3, "every reported entry ticks scan progress"); + } + fn sync_test_dir(label: &str) -> PathBuf { let nonce = SystemTime::now() .duration_since(SystemTime::UNIX_EPOCH) diff --git a/docs/BUGS.md b/docs/BUGS.md index 3923ce6..ac58a8a 100644 --- a/docs/BUGS.md +++ b/docs/BUGS.md @@ -12,9 +12,93 @@ Conventions: --- +## B89 — The recorded SDK dependency is three minor versions out of date in the working memory + +**Status:** Open — documentation only, but it misdirects every reader of the SDK code. +**Found:** 2026-08-17, while checking B85's evidence against the reader implementation. +**Where:** `proton-dev/CLAUDE.md` (the container-level file, not in this repo). + +`CLAUDE.md` states that `proton-drive-linux/Cargo.toml` declares `proton-sdk = "0.1"` / +`proton-drive-rs = "0.1"` and that `Cargo.lock` pins **0.1.11** from the registry, against a local +`proton-sdk-rs/` at 0.1.12. Neither half is true any more: the lock pins **0.6.0**, bumped by +1.7.1 (see its changelog entry, "Workspace deps bumped to `proton-sdk` / `proton-drive-rs` 0.6.0") +and never reflected in the working memory. + +**Why it matters beyond tidiness.** The whole point of that section is to tell you which source +you are reasoning about when you read SDK code. Verifying B85 meant establishing what `read_at` +does in the version the client actually builds against; following `CLAUDE.md` sends you to +`proton-drive-rs-0.1.11` in the registry cache, and to a local checkout described as one patch +ahead of it when it is in fact five minor versions behind. The conclusion happened to be the same +in every version — but that was luck, and the next question of this shape may not be. + +**Fix:** correct the versions, and prefer describing where to *look* (`grep -A2 'name = +"proton-drive-rs"' Cargo.lock`) over restating a number that goes stale on every dependency bump. + +--- + +## B88 — `pdfs pin`, `rename`, `move`, `delete` and sharing still cannot name a path under a secondary mount + +**Status:** Open — B86's fix covers `ls` and `refresh` only. +**Found:** 2026-08-17, while fixing B86. +**Where:** `crates/pdfs-fuse/src/control.rs`, every handler still calling `rel_to_mount`. + +B86's second half added `Core::rooted_at` and `control.rs::route_to_mount`, which resolve a path +against whichever mount the daemon owns it under and answer the request in that mount's inode +space. Only `ListDir` and `RefreshScope::Dir` were routed through it, because those two were what +the B86 repro needed: an escape hatch for a folder serving a stale listing. + +Every other path-addressed control request — `Pin`, `Unpin`, `Rename`, `Move`, `Delete`, +`CreateFolder`, `OpenFile`, the sharing and public-link requests, `ListRevisions` — still uses the +primary-only `rel_to_mount`, so an absolute path inside a secondary on-demand mount is rejected +with "is not under the mountpoint". The user can now *see* such a folder's contents and cannot act +on them by path. + +**Why it was left.** Routing a mutating request is not the same change as routing a read. Each of +those handlers resolves an inode, checks access, performs a remote operation and then invalidates +local state, and several of them invalidate `self.state()` — the primary — explicitly. Routing +them means auditing each for which mount's state it is really talking about, which is a larger and +riskier change than the one B86 needed. Doing it half-way would be worse than not doing it. + +**Required fix/test:** move the remaining handlers onto `route_to_mount`, auditing each for +per-mount state assumptions as it goes; the `for_each_state`/uid-keyed helpers are the pattern for +the ones that touch more than their own inode space. Test: a `pdfs pin` and a `pdfs rename` by +absolute path inside a secondary on-demand mount, plus a path under no mount still erroring. + +--- + +## B87 — A revision's block geometry is learned a read too late + +**Status:** Open — a bounded, self-correcting cost, accepted deliberately in 1.8.2. +**Found:** 2026-08-17, while fixing B85. This is the residue that fix leaves. +**Where:** `crates/pdfs-fuse/src/reads.rs` (`Core::block_geometry`, `read_block`). + +B85's fix takes block boundaries from the revision instead of assuming 4 MiB. The geometry comes +from `RevisionReader::block_sizes()`, and a reader is expensive to open — link details, ancestor +keys, an S2K node-key unlock and the block table, which is exactly the per-read work the block +cache exists to avoid (B12). So the geometry is *learned* as a side effect of the first read, +which has to open a reader anyway, and written to a `.geom` sidecar for later reads. + +The first read of a file therefore still plans on `BlockGeometry::uniform`. On a file whose blocks +are not 4 MiB that plan is wrong, and it costs exactly one straddling fetch: the reader fetches +two of the revision's blocks to answer for one client block, and the block written to the cache is +at an offset the *next* read — now planning correctly — will not ask for, so it ages out unused. +Every read after the first is right. + +**Why it is not being fixed now.** The alternatives are all worse than the cost. Resolving a +reader before planning reintroduces B12's per-read key derivation on every file. Persisting the +geometry at upload only covers files this client wrote, which are the ones the fallback is already +correct for. Fetching the block table alone still costs the node-key unlock. + +**Worth doing if:** non-4-MiB revisions turn out to be common on real accounts rather than a +handful of files on one, or if the block table becomes reachable without unlocking the node key. +Measure before deciding — one wasted fetch per file per revision is not obviously worth paying +anything to remove. + +--- + ## B86 — An on-demand mount serves a listing that predates the mirror engine's own uploads -**Status:** Open, live-reproduced 2026-08-16 on 1.8.1. +**Status:** Fixed in code (2026-08-17); live re-run of the repro pending. **Found:** 2026-08-16, while trying to drive B42's remote-delete path from one machine. **Where:** `crates/pdfs-fuse/src/sync.rs` upload path (no listing invalidation) meets the cached children snapshot the on-demand mount enumerates from. @@ -48,6 +132,35 @@ creates a remote node, the same way the FUSE write path does; and let `refresh`/ path under any mount the daemon owns, not just the primary. Test: the repro above, asserting the new files appear immediately after the mode switch. +### Fix (2026-08-17) + +Two halves, matching the two halves of the report. + +**The stale listing.** `Core::invalidate_children_of` (`pdfs-fuse/src/lib.rs`) drops a folder's +cached children *by uid*. The existing `invalidate_parent_listing` could not serve here for two +reasons: it names the folder by inode, and a mirror folder has no inode space at all at the moment +the engine writes to it; and the listing that outlived the pass was not a hot-cache map but the +`listed` flag on the folder's DB row — which is why restarting the daemon did not help. The new +helper clears that flag unconditionally and then invalidates whichever mounts happen to hold the +folder resident. The sync engine calls it after every remote mutation it makes: folder creation +and file upload (`apply_one`, `upload_new`), and the three remote-trash sites, whose parent uid +comes from the pass's folder-uid map or the baseline (`parent_uid_of`). + +**The missing escape hatch.** `StateRegistry` now carries each mount's `size_upgrades` map as well +as its state, notifier and liveness flag — that set is exactly the per-mount half of a `Core` +(everything `fork_state` replaces) — and `StateRegistry::covering_parts` returns it for the mount +that most specifically covers a path. `Core::rooted_at` swaps those fields into a clone of the +`Core`, giving a view rooted at the owning mount plus the path relative to *its* root. `pdfs ls` +and `pdfs refresh` route through it (`control.rs::route_to_mount`), so a secondary on-demand mount +can be named at last; a path under nothing this daemon has mounted still gets the familiar "is not +under the mountpoint" error. The other path-addressed control requests deliberately still use the +primary-only `rel_to_mount` — routing the mutating ones is a larger change and was not what +blocked the repro. + +Test: `covering_parts_route_a_path_to_the_mount_that_owns_it` covers the routing rule (nested +mount wins, suffix relative to it, its own inode space and upgrade batches, a path under no mount +routes nowhere). The listing-invalidation half is the live repro, which is what is still pending. + --- ## B84 — A block read that comes back short truncates the file, silently @@ -166,7 +279,9 @@ download on first read. ## B85 — The read path assumes 4 MiB blocks; some revisions do not have them -**Status:** Open — **data correctness**, currently masked by B84's repair path +**Status:** Fixed in code (2026-08-17) — the geometry now comes from the revision. See +"Correction to the diagnosis" below: the premise that this was *losing* data does not survive +reading the SDK, and the fix is a cost fix, not a correctness one. **Found:** 2026-08-16, while live-verifying B84 (see its journal evidence) **Where:** `crates/pdfs-fuse/src/reads.rs`, `crates/pdfs-core/src/cache.rs` (`BLOCK_SIZE`, `block_len`) @@ -210,6 +325,65 @@ size recorded per block after some transformation. Worth answering before choosi "blocks are whatever the revision says" and "blocks are a different constant" lead to different designs. +### Correction to the diagnosis (2026-08-17) + +**The logged numbers cannot have come from `read_at`, and the "this client asks for the wrong +range" reading is wrong.** `RevisionReader::read_at(offset, length)` does not assume any geometry: +it computes `end = min(offset + length, file_size)`, plans over the *real* block sizes +(`plan_blocks`), and splices each block clamped to `[offset, end)` (`splice_block`). Both are +present and identical in the version the client actually builds against — `Cargo.lock` pins +`proton-drive-rs` **0.6.0**, not the 0.1.11 the top-level `CLAUDE.md` still describes — and in +0.1.11 as well. So a `read_at(bstart, blen)` can return at most `blen` bytes, and every quoted +line has `got > blen`: + +``` +bidx=0 blen=4194304 got=4939506 # got exceeds the length requested +bidx=1 blen=4194304 got=5670228 # …as does this one +``` + +The only site in the client that logs this shape (`reads.rs`, the `blen`/`got`/`fsize` warning) +reaches it solely on the `bytes.len() != blen` branch of a `read_at(bstart, blen)`. Those two +facts cannot both hold. Either the numbers were transcribed with `blen` and `got` swapped, or they +came from a build that is not in the history. **This wants re-capturing from the journal before +anything is concluded from it** — and in particular, the entry's claim that B84's repair path is +"currently masking data loss" here is not supported. + +What *is* true, and is what got fixed: the client assumed the geometry. `idx * BLOCK_SIZE` and +`block_len(size, idx)` encoded "4 MiB blocks with a remainder" in the range planner, the in-memory +ring, the in-flight fetch key and the on-disk block cache. When a revision is not shaped that way, +nothing breaks — but every client block straddles two server blocks, so the reader fetches and +decrypts two to answer for one, and the block written to disk lines up with nothing the next read +asks for. That is a real and unbounded cost on exactly the files the entry found, and it defeats +the property the block cache was built around. + +### Fix (2026-08-17) + +`BlockGeometry` (`pdfs-core/src/cache.rs`) is a prefix sum over a revision's block sizes, with +`BlockGeometry::uniform(size)` reproducing the old constant-block arithmetic exactly. Blocks are +now addressed by `BlockSpan { idx, start, len }` rather than by index, everywhere: the range +planner (`spans(offset, end)`), the in-memory ring, the in-flight-fetch key, and the on-disk block +cache. The span is the identity because the index alone stopped naming one byte range — block 1 +under the fallback and block 1 under the real geometry are different bytes — so the block +sidecar records the block's plaintext `start` and a re-planned read misses rather than being +served the wrong offset. Sidecars written before that field existed read as "wherever the 4 MiB +assumption put it", which keeps the existing cache intact for every file this client uploaded. + +The geometry itself is learned as a side effect of the first read (which has to open a +`RevisionReader` anyway) and written to a `.geom` sidecar under the block cache; +`Core::block_geometry` reads it back and falls back to `uniform` when it is absent. Resolving a +reader just to *plan* would reintroduce the per-read key derivation the block cache exists to +avoid (B12), so the first read of a file still plans on the fallback and self-corrects. A recorded +table whose sizes do not sum to the node's size is refused outright, and eviction takes the +geometry with the blocks it describes. + +The open question above is still open, and no longer blocks anything: "blocks are whatever the +revision says" is now what the code does, so a different constant is just another table. + +Tests: `the_uniform_geometry_is_the_old_block_arithmetic`, +`a_non_uniform_geometry_places_blocks_by_prefix_sum` (built on the exact sizes from the log +above), `a_cached_block_is_not_served_to_a_different_geometry`, +`a_recorded_geometry_is_returned_only_for_the_revision_it_describes`. + --- ## B83 — An access-deferred op retries forever and is reported as an ordinary queued upload @@ -2611,7 +2785,8 @@ reused, making a session-long tombstone safe. ## B49 — Failed Conflict Preservation Still Overwrites the Local File (CRIT-05) -**Status:** Fixed in code with deterministic regressions; live/fault verification pending +**Status:** Fixed, with the injected-failure suite the entry asked for (2026-08-17). Only the +end-to-end live conflict remains unexercised. **Found:** 2026-07-22, 1.0 data-safety audit **Where:** `crates/pdfs-fuse/src/sync.rs`, download/conflict handling @@ -2624,11 +2799,23 @@ use collision-resistant names. Test existing names, permissions, `ENOSPC`, `EIO`, and cross-filesystem behavior. This is B39's failure path; B25 remains the separate weak-baseline problem. +**Verification (2026-08-17).** `preserve_conflict_copy` was already correct — it copies to a +`create_new` destination, syncs data and directory, and removes the source last, so every failure +returns `Err` and `apply_one` never reaches the download. What was missing was proof, so: +`conflict_preservation_failures_all_keep_the_local_file` drives an unwritable directory +(`EACCES`), a source that vanished between planning and preservation (`ENOENT`), and a destination +name already taken by a directory — and asserts the local file is still there, byte-identical, +after each. `a_conflict_copy_is_byte_and_mtime_identical` covers the other half: the copy carries +the bytes, mtime and mode, so the next pass uploads the user's version rather than something that +looks freshly written. The existing collision test covers the suffix search. + +Not covered: a genuine `ENOSPC` or `EIO` from the storage layer, and the end-to-end live conflict. + --- ## B50 — Failed Orphan Staging Deletes the Sole Scratch Copy (CRIT-06) -**Status:** Fixed in code; injected filesystem-failure verification pending +**Status:** Fixed, with injected filesystem-failure verification (2026-08-17). **Found:** 2026-07-22, 1.0 data-safety audit **Where:** `crates/pdfs-fuse/src/lib.rs`, `Core::stage_orphaned_write` @@ -2640,11 +2827,29 @@ metadata are committed. Inject `ENOSPC`, `EIO`, `EROFS`, permission, and cross-device failures and assert a byte-identical copy remains discoverable. Related: B23, B27, and B28. +**Verification (2026-08-17).** Four new tests in `pdfs-core/src/cache.rs`. +`orphan_preservation_keeps_the_scratch_copy_when_staging_is_unwritable` makes staging mode `0500` +so both the rename and the copy fallback fail, and asserts the scratch file survives +byte-identical, is the path the error names, and carries its recovery marker. +`a_retained_orphan_write_is_discoverable_after_a_restart` reopens the cache after that failure and +asserts `recovered_writes` finds the bytes *with their metadata* — present is not the same as +discoverable, and discoverable is what the entry asked for. +`orphan_preservation_across_a_device_boundary_publishes_before_removing` and +`a_failed_cross_device_preservation_keeps_the_marked_source` drive the copy fallback across a real +device boundary (`/dev/shm` vs `/tmp`), asserting the ordering — destination and sidecar published +and synced first, source removed last — and that a publication which does not finish leaves the +marked source in place. The cross-device pair skips with a printed note on a machine that has only +one filesystem, rather than passing silently. + +Not covered: a genuine `ENOSPC`/`EIO` from the block layer; `EROFS` is approximated by the +unwritable-directory case rather than a read-only mount. + --- ## B51 — Scratch Sidecar Does Not Establish a Durable Recovery Record (CRIT-07) -**Status:** Fixed in code; power-loss verification pending +**Status:** Fixed, with crash-boundary and corrupt-sidecar verification (2026-08-17). True +power-loss testing (a machine actually losing power mid-write) is still not covered. **Found:** 2026-07-22, 1.0 crash-consistency audit **Where:** `crates/pdfs-core/src/cache.rs`, scratch durability markers @@ -2657,11 +2862,24 @@ a sparse partial write as complete and replace untouched ranges with zeros. then sync the directory before acknowledging durability. Crash-test every boundary and corrupt/partial sidecars. B20 and B23 do not cover this invariant. +**Verification (2026-08-17).** `a_half_written_scratch_sidecar_is_never_read_as_a_record` puts on +disk exactly what a crash inside `mark_scratch_durable` leaves — a `.json.tmp` and no published +sidecar — and asserts the next open recovers nothing from it, so an unpublished temporary can +never be replayed as a write. `a_corrupt_scratch_sidecar_still_rescues_its_bytes` truncates a +published sidecar to a JSON prefix and asserts the blob is still moved to `recovery/` +unaccompanied: metadata loss must not take the user's bytes with it. +`remarking_a_scratch_file_replaces_its_record_atomically` covers the repeated-`fsync` case — each +call leaves one complete parseable record and no debris. + +Not covered: real power loss. Nothing short of a crashing VM tests that, and the entry should not +be closed as if it did. + --- ## B52 — Staged-Write Publication Is Not Crash-Safe (CRIT-08) -**Status:** Fixed in code; cross-filesystem and power-loss verification pending +**Status:** Fixed, with cross-filesystem verification against a real device boundary +(2026-08-17). True power-loss testing is still not covered. **Found:** 2026-07-22, 1.0 crash-consistency audit **Where:** `crates/pdfs-core/src/cache.rs`, staged-write publication @@ -2674,11 +2892,24 @@ renames and directory syncs; delete the source only after verification. Inject failure after every write, sync, rename, DB enqueue, and deletion. Related: B50 and B51. +**Verification (2026-08-17).** `stage_write` gained a `stage_write_at` seam — the public entry +point picks a timestamped destination, which is right for production and impossible to arrange a +failure at, the same reason `preserve_write_at` exists. +`staging_across_a_device_boundary_removes_the_source_last` drives the copy path across a real +device boundary and asserts the destination and sidecar are both on disk before the source goes. +`a_failed_staging_sidecar_keeps_the_cross_device_source` blocks the sidecar's temporary name and +asserts the source survives — a staged blob nothing describes is not a substitute for the bytes. +`an_unaccompanied_staged_blob_survives_a_restart` asserts startup leaves an undescribed staging +blob (and a leftover `.json.tmp`) alone rather than tidying user data away. + +Not covered: real power loss, and injected failure at *every* one of the listed points — the +sidecar and source-removal boundaries are covered, the intermediate `fsync` calls are not. + --- ## B53 — Superseding a Pending Operation Is Not Transactional (CRIT-09) -**Status:** Fixed with rollback regression; crash/ENOSPC verification pending +**Status:** Fixed, with rollback, restart and full-database verification (2026-08-17). **Found:** 2026-07-22, 1.0 SQLite durability audit **Where:** `crates/pdfs-core/src/db/ops.rs`, `Db::enqueue_op` @@ -2691,11 +2922,22 @@ blob until commit and make cleanup recoverable. Inject insert/commit failures and prove that old or new operation—never neither—remains queued. B27 concerns a retained failing row; this issue erases the row. +**Verification (2026-08-17).** The existing `failed_superseding_insert_keeps_the_old_pending_op` +proves the row rolls back. Two tests were added for the parts that make the rollback usable. +`a_rolled_back_supersede_keeps_its_blob_and_survives_reopen` asserts the failed call does *not* +report the old blob as superseded — that value is the caller's instruction to delete those bytes, +and the row that owns them is still queued — and that the rollback is still there after closing +and reopening a file-backed database, so it is not an artefact of the connection that failed. +`a_full_database_rolls_the_supersede_back_rather_than_erasing_it` caps `max_page_count` at the +current size, which is the same `SQLITE_FULL` an out-of-space filesystem produces, and asserts the +acknowledged upload is still queued and still owns its bytes. + --- ## B54 — Total-Wipe Guard Does Not Protect a One-Entry Baseline (CRIT-10) -**Status:** Fixed with one-entry regression; live mode-matrix verification pending +**Status:** Fixed, with one-entry, absent-root, unreadable-root and replaced-mountpoint +regressions (2026-08-17). The live mode matrix is still the only thing unexercised. **Found:** 2026-07-22, 1.0 mirror-sync audit **Where:** `crates/pdfs-fuse/src/sync/planner.rs` @@ -2708,11 +2950,31 @@ independently complete scan plus an explicit deliberate-delete signal for a total wipe. Test absent roots, replaced mountpoints, permissions, and one-file folders in every mode combination. +**Verification (2026-08-17).** The guard reads `!baseline.is_empty()`, so every non-empty baseline +is protected; `guard_local_wipe_blocks_only_a_total_disappearance` already carried the one-entry +case. The "independently complete scan" half is B55's fix — an absent, unreadable or partially +readable root now fails the pass before the guard is reached, which those tests cover. + +What was missing is the case where nothing fails: a root that is readable but is no longer the +tree it was, an empty directory left where a mount used to be. +`a_replaced_mountpoint_trips_the_guard_at_every_baseline_size` walks a real empty root, confirms +the scan honestly reports nothing, and asserts the guard fires for baselines of one, two and three +paths — then that it steps aside the moment the tree is back, so a remounted folder resumes +instead of staying wedged. + +**Deliberate divergence from the required fix:** there is no explicit deliberate-delete signal. +The guard refuses a total wipe outright rather than offering a way to confirm one. That is +stricter than asked for, and it means a user who really did delete everything must remove and +re-add the sync folder. Worth revisiting if anyone hits it; not worth building a confirmation path +nobody has asked for yet. + +Not covered: the live mode matrix (`scripts/fuse-acceptance.sh --managed-live`). + --- ## B55 — Incomplete Local Scans Are Interpreted as Deletions (CRIT-11) -**Status:** Fixed in code; injected iterator/stat failure tests pending +**Status:** Fixed, with injected iterator/stat failure tests (2026-08-17). **Found:** 2026-07-22, 1.0 mirror-sync audit **Where:** `crates/pdfs-fuse/src/sync.rs`, local scan/reconciliation @@ -2725,6 +2987,23 @@ subtree must make the pass non-destructive and retain its prior baseline. Test `readdir`, `stat`, permission, transient-I/O, and disappearing-entry failures. B42 covers the inverse local-delete failure. +**Verification (2026-08-17).** The scan body was lifted out of `Core::walk_local` into a free +`walk_local_tree`, so it can be driven against a real tree with a real unreadable directory in it +and no `Core` at all. Every failure in it is fatal to the pass by construction — a `read_dir` that +fails, an entry that cannot be stat-ed, a name that is not UTF-8 — because a skipped path is a +path reconciliation reads as a deletion and propagates to Drive. + +Tests: `an_unreadable_subdirectory_fails_the_scan_instead_of_shrinking_it` (chmod `000` on a +subdirectory holding a file, asserting the error names the subtree), +`an_unreadable_root_fails_the_scan`, `an_absent_root_fails_the_scan` (the unmounted-device case), +`a_non_utf8_name_fails_the_scan`, and `a_readable_tree_scans_completely` as the complement — the +strictness must not cost a working pass, so a healthy tree still reports every entry and still +skips symlinks by design. + +Not covered: transient I/O errors, and an entry that disappears between `read_dir` and `stat`. +Both are races that would need a filesystem shim to force; the code path they take is the same +`symlink_metadata` failure the permission test drives. + --- ## B56 — Database Schema Version Parsing Fails Open (HIGH-13) diff --git a/docs/CHANGELOG.md b/docs/CHANGELOG.md index 6b980b5..5384099 100644 --- a/docs/CHANGELOG.md +++ b/docs/CHANGELOG.md @@ -9,6 +9,84 @@ release ships. Migrations are forward-only — a database written by a newer cli refuse-to-open, not a downgrade, so rolling back a release means restoring the cache from scratch (user data in `staging/` and `recovery/` is never touched by this). +## [1.8.2] — 2026-08-17 + +Closes the two findings 1.8.1's live verification produced (B85, B86) and finishes the data-safety +audit entries that had been sitting as "fixed in code, verification pending" since 1.0 (B49–B55). +No schema change. Schema: **28**. + +The read path no longer assumes a revision is a series of 4 MiB blocks; it takes the geometry from +the revision. The B85 investigation also **contradicted its own bug report** — see Notes, because +the conclusion recorded there was that reads of such files were losing data, and they were not. + +### Fixed +- A folder switched from mirror to on-demand now shows the files the mirror pass itself uploaded. + The sync engine created nodes on the remote without invalidating the parent's cached child + listing, and that listing is a database flag rather than an in-memory cache — so it survived a + daemon restart, and the files stayed invisible until the folder was switched back. Every remote + mutation the engine makes (folder create, file upload, and the three trash paths) now drops the + parent's cached listing by uid, which reaches a folder that no mount has resident — the case a + mirror folder always is (`docs/BUGS.md` B86). +- `pdfs ls` and `pdfs refresh` accept a path under any mount the daemon owns, not only the primary + one. Both previously answered "is not under the mountpoint" for anything inside a secondary + on-demand mount, which left a folder serving a stale listing with no manual way to re-enumerate + it. A path under nothing this daemon has mounted still gets the same error as before. +- A file whose blocks are not 4 MiB is no longer read twice over. Block offsets came from + `index * 4 MiB`, so on a revision laid out differently every block the client asked for + straddled two of the revision's own, and each was fetched and decrypted to answer for one — then + cached at boundaries the next read would not ask about. Offsets now come from the revision's + recorded block sizes, and the on-disk cache, the in-memory ring and the in-flight fetch key + address a block by the byte range it covers rather than by its index (`docs/BUGS.md` B85). + +### Changed +- The block cache records each block's plaintext start alongside its `(mtime, size)` tag, and each + revision's block table in a small sidecar beside its blocks. Blocks cached by an earlier release + carry no start; they are honoured wherever the old 4 MiB assumption and the new plan agree on + the offset, which is every file this client uploaded — so the existing cache survives the + upgrade intact and is quietly refetched only for the files it was placing wrongly. + +### Notes +- **B85's diagnosis was wrong, and the entry has been corrected rather than closed.** It read a + set of journal lines as proof that this client was requesting wrong byte ranges and being saved + only by B84's repair path. `RevisionReader::read_at` does not work that way: it plans over the + revision's real block sizes and clamps every block to the range asked for, identically in the + 0.6.0 the client builds against and in every earlier version. A read can therefore never return + more bytes than it requested — yet every quoted line shows exactly that (`blen=4194304 + got=4939506`), and the one place in the client that logs that shape can only reach it on the + opposite condition. The numbers were either transcribed with two fields swapped or captured from + a build not in the history; they want re-capturing from the journal before anything is concluded + from them. What was real is the assumption itself, and that is what this release removes — a + cost fix, not a correctness one. Nothing here changes what a read returns. +- The first read of a file still plans on the 4 MiB assumption, and learns the real geometry as a + side effect of opening the reader it needs anyway. Resolving a reader merely to *plan* would + reintroduce the per-read key derivation the block cache exists to avoid, so the first read of a + non-4-MiB file pays one straddling fetch and then self-corrects. +- Seven data-safety audit entries (B49–B55) moved from "fixed in code, verification pending" to + verified. The fixes were all genuinely present and needed no change; what they lacked was the + injected-failure testing each entry asked for, and that is what was written — unwritable + directories and `chmod 000` subtrees for permission failures, a real `/dev/shm`↔`/tmp` device + boundary for the cross-filesystem fallbacks, SQLite's `max_page_count` for a full disk, an abort + trigger for a failed insert, and truncated and unpublished sidecars for the crash boundaries. + Two seams were added to make failure arrangeable at all: `stage_write_at`, mirroring the + `preserve_write_at` that already existed, and `walk_local_tree` lifted out of the sync engine so + the local scan can be driven against a real tree without a daemon. +- **Two of those entries are not fully closed, and say so.** B51 and B52 concern crash consistency, + and nothing short of a machine actually losing power tests that; the crash *boundaries* are + covered by reconstructing on disk exactly what a crash at each point leaves. B54 also diverges + from its required fix on purpose: it has no "deliberate delete" confirmation path, so a total + local wipe is refused outright rather than offered for confirmation. That is stricter than asked + for, and the cost is that a user who really did delete everything must remove and re-add the + sync folder. +- B86's fix routes only `ls` and `refresh` to the owning mount. The other path-addressed control + requests — pin, rename, move, delete, sharing — still resolve against the primary mount only. + That is a larger change and was not what blocked the repro; it is recorded as an open follow-up + (`docs/BUGS.md` B88). +- Three findings were opened rather than fixed: the geometry above is learned one read late, so a + non-4-MiB file pays a single wasted fetch on its first read (B87); the remaining path-addressed + control requests are still primary-only (B88); and the container-level `CLAUDE.md` still + describes the SDK dependency as 0.1.11 when 1.7.1 bumped it to 0.6.0, which is what sent the + B85 investigation to the wrong source tree to begin with (B89). + ## [1.8.1] — 2026-08-16 Some files read short through the mount and reported success — an application got a truncated @@ -361,6 +439,9 @@ First stable release: FUSE files-on-demand mount, sync daemon under `proton-driv - The outstanding FUSE defects tracked in `docs/BUGS.md`, plus a truncate defect, validated by a new POSIX compliance suite for the filesystem. +[1.8.2]: https://github.com/narrrl/proton-drive-linux/releases/tag/v1.8.2 +[1.8.1]: https://github.com/narrrl/proton-drive-linux/releases/tag/v1.8.1 +[1.8.0]: https://github.com/narrrl/proton-drive-linux/releases/tag/v1.8.0 [1.7.1]: https://github.com/narrrl/proton-drive-linux/releases/tag/v1.7.1 [1.7.0]: https://github.com/narrrl/proton-drive-linux/releases/tag/v1.7.0 [1.6.0]: https://github.com/narrrl/proton-drive-linux/releases/tag/v1.6.0 diff --git a/packaging/PKGBUILD b/packaging/PKGBUILD index 655ec0b..114a9c0 100644 --- a/packaging/PKGBUILD +++ b/packaging/PKGBUILD @@ -9,7 +9,7 @@ # Build + install: cd packaging && makepkg -fi pkgname=proton-drive-linux -pkgver=1.8.1 +pkgver=1.8.2 pkgrel=1 pkgdesc="Proton Drive client for Linux: FUSE files-on-demand mount, CLI, GTK4 app + tray" arch=('x86_64')