diff --git a/crates/pdfs-fuse/src/lib.rs b/crates/pdfs-fuse/src/lib.rs index 1eea574..86bd628 100644 --- a/crates/pdfs-fuse/src/lib.rs +++ b/crates/pdfs-fuse/src/lib.rs @@ -80,15 +80,18 @@ pub use filesystem::ProtonFs; mod mount; mod photos; mod profile; +mod reads; mod sharing; mod state; mod sync; mod transfers; +mod upload; mod workers; use background::{run_event_sync, run_local_index}; pub(crate) use mount::is_stale_mount; pub use mount::{MountOutcome, mount}; use mount::{SecondaryMount, clear_stale_mount, fuse_connection_id, umount_session_unblocked}; +use reads::{ReaderSlot, STREAM_BYPASS_MIN, StreamRing}; use state::{Entry, Intervals, PendingRevision, State, WriteHandle}; use tracing::{debug, error, info, warn}; use transfers::{CountingWriter, JobGuard, OwnedCountingReader, TransferRegistry}; @@ -149,38 +152,6 @@ const PHOTOS_SYNCED_MS: &str = "photos_synced_ms"; const PHOTOS_AVAILABLE: &str = "photos_available"; const TRASH_SYNCED_MS: &str = "trash_synced_ms"; -/// Longest edge, in px, of a thumbnail generated locally for a photo the server -/// has none for. Matches the server's own thumbnail scale closely enough that a -/// tile can't tell them apart. -const THUMB_EDGE: u32 = 512; -/// JPEG quality of a locally generated thumbnail. -const THUMB_QUALITY: u8 = 82; -/// How many photos may be downloaded at once to generate their missing -/// thumbnails. Bounded: a screenful of 20 MB digicam JPEGs would otherwise -/// saturate the link and starve the rest of the daemon. -const THUMB_GEN_CONCURRENCY: usize = 4; - -/// A read of an unpinned video at least this large streams *without* persisting -/// its blocks to the on-disk cache. Playing a 2 GB film would otherwise pour it -/// through the block LRU and evict everything else the user actually wants kept — -/// and it re-streams cheaply enough on a rewatch that keeping it was never worth -/// that. Pinned videos (kept offline on purpose) and anything smaller cache as -/// usual. -const STREAM_BYPASS_MIN: u64 = 256 * 1024 * 1024; - -/// How many bytes of streamed blocks the in-memory ring keeps. Bypassing the -/// on-disk cache must not mean re-fetching: the kernel asks for a streamed file -/// in reads far smaller than [`BLOCK_SIZE`], so without a ring every 128 KiB the -/// player consumes would download and decrypt a whole 4 MiB block again. Sized -/// for a handful of blocks per concurrently-streamed file. -const STREAM_RING_BYTES: u64 = 128 * 1024 * 1024; - -/// Blocks fetched *past* the one a sequential streaming read needed, warmed into -/// the ring in the background so the player's next read is already in memory -/// instead of waiting a round-trip. Kept small so it cannot crowd out the -/// in-flight block governor. -const STREAM_READAHEAD: u64 = 4; - /// How stale the local-file index may get before the background scanner rebuilds /// it. A rescan is a full walk of `$HOME`, so this trades index freshness against /// disk churn; the scanner also always rebuilds once per daemon start when the @@ -215,222 +186,6 @@ const MAX_THUMBNAIL_MISSES: usize = 8192; /// delay the waiters this chunking exists to release (bugs.md B14). const SIZE_UPGRADE_CHUNK: usize = 150; -/// How many files the bulk uploader ships at once. Overlaps the per-file network -/// round-trips without letting an unbounded number of block buffers pile up. -const UPLOAD_CONCURRENCY: usize = 4; - -/// One file queued for bulk upload, resolved during the directory walk so the -/// concurrent phase carries everything it needs (no shared state, no `block_on`). -struct UploadTask { - /// Inode of the (already-created) remote parent folder, for interning the - /// uploaded node afterwards. - parent_ino: u64, - parent_uid: NodeUid, - name: String, - /// Local filesystem path to stream from. - path: PathBuf, - size: u64, -} - -/// Tally of a completed [`Core::upload_paths`] batch, for the daemon log. -#[derive(Default)] -struct UploadStats { - uploaded: usize, - failed: usize, - /// Total plaintext bytes of the files that uploaded successfully. - bytes: u64, - /// Folders created (or reused) to mirror the local tree. - folders: usize, -} - -/// Upload every [`UploadTask`] with at most `limit` in flight at once, each -/// streamed straight from disk and ticking its own transfer-registry guard. -/// Returns, per file, either `(parent_ino, new_uid)` for the caller to intern or -/// `(name, error)` to log — one failure never sinks the batch. -/// -/// `job` counts files finished (either way: a failure is still one file the batch -/// no longer waits on), so a front-end can show "12 of 40" over the per-file bars. -async fn run_uploads( - core: Core, - tasks: Vec, - limit: usize, - job: Arc, -) -> Vec> { - let sem = Arc::new(tokio::sync::Semaphore::new(limit)); - let mut set = tokio::task::JoinSet::new(); - for t in tasks { - let core = core.clone(); - let sem = sem.clone(); - set.spawn(async move { - let _permit = sem.acquire_owned().await.expect("semaphore not closed"); - let file = match std::fs::File::open(&t.path) { - Ok(f) => f, - Err(e) => return Err((t.name, format!("open {}: {e}", t.path.display()))), - }; - let mtime = std::fs::metadata(&t.path) - .and_then(|m| m.modified()) - .ok() - .and_then(|mt| mt.duration_since(std::time::UNIX_EPOCH).ok()) - .map(|d| d.as_secs() as i64); - let guard = core - .transfers - .begin(&t.name, "", TransferDirection::Upload, t.size); - let reader = OwnedCountingReader::new(file, guard); - match core - .client - .upload_file_from( - &t.parent_uid, - &t.name, - media_type_for(&t.name), - reader, - t.size as i64, - Vec::new(), - mtime, - false, - ) - .await - { - Ok(uid) => Ok((t.parent_ino, uid, t.size)), - Err(e) => Err((t.name, format!("upload: {e}"))), - } - }); - } - let mut out = Vec::new(); - while let Some(joined) = set.join_next().await { - job.step(); - match joined { - Ok(result) => out.push(result), - Err(e) => warn!(error = %e, "upload task panicked"), - } - } - out -} - -/// Shared engine behind both the FUSE callbacks and the control socket: the -/// How many [`RevisionReader`]s stay open at once. -/// -/// A reader holds its revision's content key and block table — a few KB even for -/// a large file, so this is bounded for tidiness and staleness rather than -/// memory. Evicted least-recently-used; a dropped reader costs one re-open -/// (two API calls and a node-key unlock) the next time that file is read. -const MAX_OPEN_READERS: usize = 64; - -/// An open reader plus the node metadata it was opened against. -/// -/// The SDK pins a reader to the revision that was active at `open_revision`, so -/// a reader is only reusable while the node still reports the same -/// `(mtime, size)` — the same validity pair the content cache uses (a new -/// revision bumps mtime). On a mismatch the reader is dropped and reopened. -struct CachedReader { - reader: Arc, - mtime: i64, - size: u64, - /// For LRU eviction. - last_used: Instant, -} - -/// What the `readers` map holds for a node: an open reader, or the fact that -/// someone is already opening one. -/// -/// The `Pending` arm is what makes the map single-flight rather than merely a -/// cache. Opening resolves link details, ancestor keys, an S2K node-key unlock -/// and the block table — five round-trips — and read-ahead on a cold file puts -/// several workers into the open at the same moment, so without this each -/// racer would redo all of it and discard every result but one. -enum ReaderSlot { - Ready(CachedReader), - Pending(PendingOpen), -} - -/// An `open_revision` in flight. Racers clone `rx` and await the leader's result -/// instead of opening their own. -struct PendingOpen { - /// Identifies this attempt so the leader can tell "my slot is still there" - /// from "someone evicted or replaced it while I was on the network", and - /// only publish its reader in the first case. - id: u64, - /// What the leader is opening against. A racer wanting a different revision - /// must not join this open — it would get a reader for the wrong bytes. - mtime: i64, - size: u64, - rx: tokio::sync::watch::Receiver, Errno>>>, -} - -/// Distinguishes `PendingOpen`s. Only uniqueness matters, so a plain counter -/// does; ids are never compared for order. -static NEXT_OPEN_ID: AtomicU64 = AtomicU64::new(0); - -/// Drive client, a Tokio handle to bridge the synchronous FUSE/socket threads -/// to the async SDK, the inode bookkeeping, and the on-disk content cache. -/// -/// Cheaply cloneable (every field is a handle/`Arc`), so the control-socket task -/// gets its own copy while the FUSE session keeps another. -/// In-memory LRU of blocks belonging to a file streaming past the on-disk cache -/// (see [`STREAM_BYPASS_MIN`]). Bypassing the disk stops one film evicting the -/// block LRU; it must not also mean the same 4 MiB block is downloaded once per -/// 128 KiB the player reads out of it. -/// -/// 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. -#[derive(Default)] -struct StreamRing { - blocks: HashMap<(NodeUid, u64), Arc>>, - /// Keys oldest-first, for eviction. - order: VecDeque<(NodeUid, u64)>, - /// Per-node validity tag, `(mtime, size)`. - tags: HashMap, - bytes: u64, -} - -impl StreamRing { - fn get(&mut self, uid: &NodeUid, mtime: i64, size: u64, idx: u64) -> Option>> { - if self.tags.get(uid) != Some(&(mtime, size)) { - return None; - } - self.blocks.get(&(uid.clone(), idx)).cloned() - } - - fn insert(&mut self, uid: &NodeUid, mtime: i64, size: u64, idx: u64, 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); - if self.blocks.contains_key(&key) { - return; - } - self.bytes += bytes.len() as u64; - self.blocks.insert(key.clone(), bytes); - self.order.push_back(key); - while self.bytes > STREAM_RING_BYTES { - let Some(victim) = self.order.pop_front() else { - break; - }; - if let Some(dropped) = self.blocks.remove(&victim) { - self.bytes -= dropped.len() as u64; - } - } - } - - fn drop_node(&mut self, uid: &NodeUid) { - self.order.retain(|(u, _)| u != uid); - let mut freed = 0u64; - self.blocks.retain(|(u, _), bytes| { - if u == uid { - freed += bytes.len() as u64; - false - } else { - true - } - }); - self.bytes -= freed; - self.tags.remove(uid); - } -} - #[derive(Clone)] struct Core { client: ProtonDriveClient, @@ -1314,501 +1069,6 @@ impl Core { } } - /// An open [`RevisionReader`] for `uid`, reusing the cached one when it is - /// still valid for `(mtime, fsize)` and opening a fresh one otherwise. - /// - /// Opening resolves the file's link details, ancestor keys, node key (an S2K - /// unlock) and block table. Doing that once per file rather than once per - /// block is the whole point: a cold 100 MB read is 25 block misses, which - /// used to mean 25 full resolutions (50 API calls, 25 unlocks) and now means - /// one. - /// - /// Racing reads of one cold file open it **once**: the first caller claims - /// the slot and the rest await its result. That matters now that `read` is - /// served off the dispatch loop, because read-ahead on a first touch issues - /// several parallel reads of the same file as a matter of course. - async fn revision_reader( - &self, - uid: &NodeUid, - mtime: i64, - fsize: u64, - ) -> Result, Errno> { - loop { - // Decide under the lock, act outside it — network I/O must never run - // under a std `Mutex`, and an await must never hold one. - enum Act { - Hit(Arc), - Join(tokio::sync::watch::Receiver, Errno>>>), - Lead( - u64, - tokio::sync::watch::Sender, Errno>>>, - ), - } - let act = { - let mut readers = self.readers.lock(); - match readers.get_mut(uid) { - Some(ReaderSlot::Ready(entry)) - if entry.mtime == mtime && entry.size == fsize => - { - entry.last_used = Instant::now(); - Act::Hit(entry.reader.clone()) - } - Some(ReaderSlot::Pending(p)) if p.mtime == mtime && p.size == fsize => { - Act::Join(p.rx.clone()) - } - // Either nothing here, or something for a revision we no - // longer want (a stale reader, or an open for one). Ours - // replaces it: the newer `(mtime, size)` is the truth. - _ => { - let id = NEXT_OPEN_ID.fetch_add(1, Ordering::Relaxed); - let (tx, rx) = tokio::sync::watch::channel(None); - readers.insert( - uid.clone(), - ReaderSlot::Pending(PendingOpen { - id, - mtime, - size: fsize, - rx, - }), - ); - Act::Lead(id, tx) - } - } - }; - - match act { - Act::Hit(reader) => return Ok(reader), - Act::Join(mut rx) => { - // `changed()` errors only if the leader dropped its sender - // without publishing — it panicked. Retry from the top - // rather than fail: the slot it left is gone, so this pass - // leads its own open. - loop { - if let Some(result) = rx.borrow_and_update().clone() { - return result; - } - if rx.changed().await.is_err() { - break; - } - } - continue; - } - Act::Lead(id, tx) => { - debug!(%uid, mtime, fsize, "opening revision"); - let result = match self.client.open_revision(uid).await { - Ok(reader) => Ok(Arc::new(reader)), - Err(e) => { - warn!(%uid, error = %e, "open_revision failed"); - Err(Errno::EIO) - } - }; - { - let mut readers = self.readers.lock(); - // Publish only into the slot we claimed. If it is gone - // or belongs to a later attempt, an eviction or a newer - // revision overtook us: our reader still answers the - // callers waiting on it, but it must not be cached. - let ours = matches!( - readers.get(uid), - Some(ReaderSlot::Pending(p)) if p.id == id - ); - if ours { - match &result { - Ok(reader) => { - readers.insert( - uid.clone(), - ReaderSlot::Ready(CachedReader { - reader: reader.clone(), - mtime, - size: fsize, - last_used: Instant::now(), - }), - ); - } - // Never cache a failure: the next read retries. - Err(_) => { - readers.remove(uid); - } - } - } - Self::trim_readers(&mut readers); - } - // After the map is updated, so a racer that wakes and - // re-enters finds the finished slot rather than racing us. - let _ = tx.send(Some(result.clone())); - return result; - } - } - } - } - - /// Bound the reader map: drop least-recently-used entries until it is back - /// at [`MAX_OPEN_READERS`]. Only `Ready` slots are eviction candidates — a - /// `Pending` has callers waiting on it and no reader to drop yet. - fn trim_readers(readers: &mut HashMap) { - while readers.len() > MAX_OPEN_READERS { - let victim = readers - .iter() - .filter_map(|(uid, slot)| match slot { - ReaderSlot::Ready(entry) => Some((uid, entry.last_used)), - ReaderSlot::Pending(_) => None, - }) - .min_by_key(|(_, last_used)| *last_used) - .map(|(uid, _)| uid.clone()); - let Some(victim) = victim else { - break; // every slot is an open in flight; nothing to drop - }; - readers.remove(&victim); - } - } - - /// Drop any open reader for `uid`, so the next read reopens against the - /// current revision. Called wherever cached content is evicted. - /// - /// Drops an in-flight open too: its leader finds the slot gone and declines - /// to cache what it opened, which is what we want — the eviction says that - /// revision is no longer the truth. - fn evict_reader(&self, uid: &NodeUid) { - self.readers.lock().remove(uid); - } - - /// 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. - fn read_range( - &self, - uid: &NodeUid, - mtime: i64, - fsize: u64, - offset: u64, - len: u64, - cache_blocks: bool, - ) -> Result, Errno> { - // A queued write has not reached the remote yet, so the remote's current - // revision is stale and the staged blob is the truth. Serve from it until - // the drain worker lands the upload (offline.md Phase 3). - if let Some(pending) = self.pending.lock().get(uid).cloned() { - return self.read_pending(&pending, offset, len); - } - // A node created offline and never written has no blob and no remote: it - // is an empty file, and asking the API about a `local~` uid would only - // earn a 404 (offline.md Phase 3b). - if is_local_uid(uid) { - return Ok(Vec::new()); - } - self.read_range_remote(uid, mtime, fsize, offset, len, cache_blocks) - } - - /// Serve a read from a staged blob, falling back to the remote base for any - /// range the write did not author (an incomplete [`StagedWrite`] holds zeros - /// there, which must never be handed out as content). - fn read_pending( - &self, - pending: &PendingRevision, - offset: u64, - len: u64, - ) -> Result, Errno> { - let m = &pending.meta; - if offset >= m.len || len == 0 { - return Ok(Vec::new()); - } - let uid = parse_node_uid(&m.uid).ok_or(Errno::EIO)?; - let file = File::open(&pending.path).map_err(|e| { - error!(%uid, path = %pending.path.display(), error = %e, "open staged blob failed"); - Errno::EIO - })?; - let mut written = Intervals::default(); - for &(s, e) in &m.authored { - written.add(s, e); - } - // Same merge as `serve_open_read`, but resolving gaps against the remote - // rather than through `read_range` — going through `read_range` would find - // this very pending op and recurse. - let end = offset.saturating_add(len).min(m.len); - let mut out = Vec::with_capacity((end - offset) as usize); - for (s, e, authored) in written.segments(offset, end) { - if authored { - let mut buf = vec![0u8; (e - s) as usize]; - file.read_exact_at(&mut buf, s).map_err(|err| { - warn!(%uid, error = %err, "staged blob read failed"); - Errno::EIO - })?; - out.extend_from_slice(&buf); - continue; - } - let bend = e.min(m.base_size); - if s < bend { - out.extend_from_slice(&self.read_range_remote( - &uid, - m.base_mtime, - m.base_size, - s, - bend - s, - true, - )?); - } - // Past the base: a hole the write extended over. - out.resize(out.len() + e.saturating_sub(s.max(m.base_size)) as usize, 0); - } - Ok(out) - } - - /// Warm the [`STREAM_READAHEAD`] blocks after `last` into the ring, in the - /// background, so a player reading forward finds its next block already in - /// memory rather than paying a fetch + decrypt round-trip at each boundary. - /// - /// Fire-and-forget: a failure here is not an error, it only means the read - /// that needs those bytes will fetch them itself. - fn stream_readahead(&self, uid: &NodeUid, mtime: i64, fsize: u64, last: u64) { - let last_block = (fsize.saturating_sub(1)) / BLOCK_SIZE; - let wanted: Vec = ((last + 1)..=(last + STREAM_READAHEAD).min(last_block)) - .filter(|&bidx| { - self.stream_ring - .lock() - .get(uid, mtime, fsize, bidx) - .is_none() - }) - .collect(); - if wanted.is_empty() { - return; - } - let core = self.clone(); - let uid = uid.clone(); - self.rt.spawn(async move { - let Ok(reader) = core.revision_reader(&uid, mtime, fsize).await else { - return; - }; - // Concurrently, for the same reason the demand path fetches its - // misses concurrently: serially awaited blocks cost the sum of their - // round-trips, and the player catches up with the readahead before it - // finishes. The SDK's in-flight block governor bounds the fan-out. - let mut set = tokio::task::JoinSet::new(); - for bidx in wanted { - let reader = reader.clone(); - let bstart = bidx * BLOCK_SIZE; - let blen = BLOCK_SIZE.min(fsize - bstart); - set.spawn(async move { (bidx, reader.read_at(bstart, blen).await) }); - } - while let Some(joined) = set.join_next().await { - let Ok((bidx, read)) = joined else { continue }; - match read { - Ok(bytes) => { - core.stream_ring - .lock() - .insert(&uid, mtime, fsize, bidx, Arc::new(bytes)) - } - Err(e) => debug!(%uid, bidx, error = %e, "stream readahead failed"), - } - } - }); - } - - /// Read from the content cache, else the remote. The base-content path, with - /// no awareness of queued writes — callers wanting the file's *current* - /// content want [`Core::read_range`]. Gap-filling a staged blob is the one - /// caller that genuinely means "the base", since that is what its zeroed - /// ranges have to be filled from. - fn read_range_remote( - &self, - uid: &NodeUid, - mtime: i64, - fsize: u64, - offset: u64, - len: u64, - cache_blocks: bool, - ) -> Result, Errno> { - if let Some(bytes) = self.cache.read_range(uid, mtime, fsize, offset, len) { - return Ok(bytes); - } - if offset >= fsize || len == 0 { - return Ok(Vec::new()); - } - 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; - - // Collect the blocks overlapping the request, serving any already cached - // and fetching the rest concurrently. A multi-block read (e.g. a media - // player buffering, or a large sequential read split into one FUSE call) - // 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 { - // A streaming read keeps its blocks only in memory, so the ring is - // the only cache it has — check it before the disk (which will not - // hold them) and before the network (which would refetch the same - // 4 MiB block for every 128 KiB the player consumes). - let hit = if cache_blocks { - self.cache - .cached_block(uid, mtime, fsize, bidx) - .map(Arc::new) - } else { - self.stream_ring.lock().get(uid, mtime, fsize, bidx) - }; - match hit { - Some(b) => blocks.push(Some(b)), - None => { - blocks.push(None); - misses.push(bidx); - } - } - } - - if !misses.is_empty() { - let fetched = self.rt.block_on(async { - // Resolve the file's keys and block table once, then read every - // missing block through the shared reader. Previously each block - // called `download_range`, which redid that resolution per block. - let reader = self.revision_reader(uid, mtime, fsize).await?; - - let mut set = tokio::task::JoinSet::new(); - for &bidx in &misses { - let reader = reader.clone(); - let uid = uid.clone(); - let bstart = bidx * BLOCK_SIZE; - let blen = BLOCK_SIZE.min(fsize - bstart); - set.spawn(async move { - reader - .read_at(bstart, blen) - .await - .map(|bytes| (bidx, bytes)) - .map_err(|e| { - warn!(%uid, bstart, blen, error = %e, "block read failed"); - Errno::EIO - }) - }); - } - let mut out = Vec::with_capacity(misses.len()); - while let Some(joined) = set.join_next().await { - // A join error means the task panicked; surface it as EIO. - out.push(joined.map_err(|_| Errno::EIO)??); - } - Ok::<_, Errno>(out) - })?; - for (bidx, bytes) in fetched { - // A streaming read (large unpinned video) skips the on-disk cache - // so it can't evict the rest of it; it keeps the block in the - // in-memory ring instead, which is bounded and dies with the read. - let bytes = Arc::new(bytes); - if cache_blocks { - let _ = self.cache.store_block(uid, mtime, fsize, bidx, &bytes); - } else { - self.stream_ring - .lock() - .insert(uid, mtime, fsize, bidx, bytes.clone()); - } - blocks[(bidx - first) as usize] = Some(bytes); - } - if !cache_blocks { - self.stream_readahead(uid, mtime, fsize, last); - } - } - - for (i, block) in blocks.into_iter().enumerate() { - let bidx = first + i as u64; - let bstart = bidx * BLOCK_SIZE; - // Every slot is populated: cache hits up front, misses by the fetch above. - let block = block.expect("block fetched or cached"); - let s = (offset.max(bstart) - bstart) as usize; - let e = (end.min(bstart + block.len() as u64) - bstart) as usize; - if s < e { - out.extend_from_slice(&block[s..e]); - } - } - Ok(out) - } - - /// Serve a read against an open write handle: stitch authored ranges (from - /// the scratch file) and untouched ranges (from the remote base via the - /// block cache) in order, zero-filling any region past the base. - #[allow(clippy::too_many_arguments)] - fn serve_open_read( - &self, - file: &Arc, - len: u64, - uid: &NodeUid, - base_mtime: i64, - base_size: u64, - written: &Intervals, - offset: u64, - size: u64, - ) -> Result, Errno> { - if offset >= len || size == 0 { - return Ok(Vec::new()); - } - let end = offset.saturating_add(size).min(len); - let mut out = Vec::with_capacity((end - offset) as usize); - for (s, e, authored) in written.segments(offset, end) { - if authored { - let mut buf = vec![0u8; (e - s) as usize]; - file.read_exact_at(&mut buf, s).map_err(|err| { - warn!(%uid, error = %err, "scratch read failed"); - Errno::EIO - })?; - out.extend_from_slice(&buf); - } else { - let bend = e.min(base_size); - if s < bend { - out.extend_from_slice(&self.read_range( - uid, - base_mtime, - base_size, - s, - bend - s, - true, - )?); - } - // Anything past the base is a hole: zero-fill. - let zeros = e.saturating_sub(s.max(base_size)); - out.resize(out.len() + zeros as usize, 0); - } - } - Ok(out) - } - - /// Fill every unauthored range of a scratch/staged file that overlaps its - /// base with the base's bytes, so the file becomes the complete new content. - /// - /// This is the step a partial overwrite cannot skip: only the authored bytes - /// were ever written to disk, and a revision upload sends the whole file. - /// The gaps come from the *remote base* (through the block cache, so a small - /// edit of a large file does not pull all of it), which is exactly why this - /// can fail with no network — see `StagedWrite`. - fn fill_gaps( - &self, - uid: &NodeUid, - file: &File, - len: u64, - base_mtime: i64, - base_size: u64, - written: &Intervals, - ) -> Result<(), Errno> { - file.set_len(len).map_err(|e| { - error!(%uid, error = %e, "resize scratch file failed"); - Errno::EIO - })?; - for (s, e, authored) in written.segments(0, len) { - if authored { - continue; - } - let bend = e.min(base_size); - if s >= bend { - continue; // wholly past the base: already zero-filled on disk - } - let bytes = self.read_range_remote(uid, base_mtime, base_size, s, bend - s, true)?; - file.write_all_at(&bytes, s).map_err(|err| { - error!(%uid, error = %err, "scratch gap-fill write failed"); - Errno::EIO - })?; - } - Ok(()) - } - /// Accept a released write handle's bytes and queue their upload /// (offline.md Phase 3). /// @@ -2954,205 +2214,6 @@ impl Core { } Ok(name.to_string()) } - - /// Bulk-upload local files and directory trees under `sources` into the - /// mountpoint-relative `parent_rel` folder. Directories are recreated (or - /// merged into an existing same-named folder) and walked; the resulting flat - /// set of files is uploaded with bounded concurrency, each ticking the - /// transfer registry so a front-end sees live progress. Runs on a background - /// thread — a large tree far outlasts the control socket's read timeout — so - /// it reports only a summary for the log. Individual failures are counted and - /// logged rather than aborting the whole batch. - fn upload_paths(&self, parent_rel: &Path, sources: &[PathBuf]) -> CoreResult { - let (pino, parent_uid) = self.resolve(parent_rel)?; - self.ensure_children(pino) - .map_err(|e| self.errno_error(e, "enumerate"))?; - - // Phase 1 (sequential): build the remote folder skeleton and collect the - // flat list of files to upload. Folders must exist before their children, - // so this can't be parallelised. On a deep tree this is a folder-creation - // round-trip per directory before a single byte moves — long enough that - // it needs a job of its own, or the daemon looks idle for minutes. - let mut tasks = Vec::new(); - let mut folders = 0usize; - { - let job = self.transfers.begin_job("Preparing upload"); - for src in sources { - if let Err(e) = - self.collect_uploads(pino, &parent_uid, src, &mut tasks, &mut folders, &job) - { - warn!(source = %src.display(), error = %e, "skipping source"); - } - } - } - - // Phase 2 (concurrent): upload the files, up to UPLOAD_CONCURRENCY at once. - // Each file reports its own bytes; this job is the batch's "N of M files", - // which is the number a user actually waits on. - let job = Arc::new(self.transfers.begin_job(match tasks.len() { - 1 => "Uploading 1 file".to_string(), - n => format!("Uploading {n} files"), - })); - job.set_total(tasks.len() as u64); - let outcomes = self.rt.block_on(run_uploads( - self.clone(), - tasks, - UPLOAD_CONCURRENCY, - job.clone(), - )); - drop(job); - - // Phase 3 (sequential): intern each uploaded node so it shows up in the - // listing without a re-enumeration. fetch_node uses `block_on`, so it must - // run here rather than inside the async batch — and it is a round-trip per - // file, so it too gets a job rather than a silent tail. - let mut stats = UploadStats { - folders, - ..UploadStats::default() - }; - let job = self.transfers.begin_job("Finishing upload"); - job.set_total(outcomes.len() as u64); - for outcome in outcomes { - job.step(); - match outcome { - Ok((parent_ino, uid, size)) => { - stats.uploaded += 1; - stats.bytes += size; - match self.fetch_node(&uid) { - Ok(node) => { - let mut st = self.state.lock(); - let ino = st.intern(parent_ino, node); - if let Some(kids) = st.children.get_mut(&parent_ino) - && !kids.contains(&ino) - { - kids.push(ino); - } - } - // Uploaded fine but the metadata refresh failed; it will - // appear on the next directory enumeration regardless. - Err(e) => warn!(%uid, error = ?e, "uploaded node metadata refresh failed"), - } - } - Err((name, msg)) => { - stats.failed += 1; - warn!(name, error = %msg, "file upload failed"); - } - } - } - info!( - uploaded = stats.uploaded, - failed = stats.failed, - "bulk upload finished" - ); - Ok(stats) - } - - /// Resolve a remote child folder named `name` under `pino`, creating it if it - /// doesn't exist, and return its `(inode, uid)`. Reusing an existing same-named - /// folder makes re-uploading a directory merge into it rather than fail on a - /// duplicate name. - fn ensure_remote_folder( - &self, - pino: u64, - parent_uid: &NodeUid, - name: &str, - ) -> CoreResult<(u64, NodeUid)> { - if name.is_empty() || name.contains('/') { - return Err(CoreError::invalid(format!("invalid folder name: {name:?}"))); - } - self.ensure_children(pino) - .map_err(|e| self.errno_error(e, "enumerate"))?; - { - let st = self.state.lock(); - if let Some(kids) = st.children.get(&pino) { - for &ino in kids { - if let Some(e) = st.entries.get(&ino) - && e.node.is_folder() - && e.node.name == name - { - return Ok((ino, e.uid.clone())); - } - } - } - } - let new_uid = self - .rt - .block_on( - self.client - .create_folder(parent_uid, name, Some(now_secs())), - ) - .map_err(|e| CoreError::from_api(&e, &format!("create folder {name}")))?; - let node = self - .fetch_node(&new_uid) - .map_err(|e| self.errno_error(e, "fetch node"))?; - let mut st = self.state.lock(); - let ino = st.intern(pino, node); - if let Some(kids) = st.children.get_mut(&pino) - && !kids.contains(&ino) - { - kids.push(ino); - } - Ok((ino, new_uid)) - } - - /// Walk one local source path, appending its files to `tasks`. A file becomes - /// one task; a directory is recreated remotely and recursed into (children - /// sorted for a stable order). Symlinks and other special files are skipped. - /// - /// `job` narrates the walk with the folder currently being mirrored. It stays - /// indeterminate: nothing knows the size of the tree until the walk has ended. - fn collect_uploads( - &self, - pino: u64, - parent_uid: &NodeUid, - src: &Path, - tasks: &mut Vec, - folders: &mut usize, - job: &JobGuard, - ) -> CoreResult<()> { - let meta = std::fs::symlink_metadata(src) - .map_err(|e| CoreError::internal(format!("stat {}: {e}", src.display())))?; - let name = src - .file_name() - .and_then(|n| n.to_str()) - .ok_or_else(|| CoreError::invalid(format!("unusable name: {}", src.display())))? - .to_string(); - - if meta.is_dir() { - job.detail(&name); - let (child_ino, child_uid) = self.ensure_remote_folder(pino, parent_uid, &name)?; - *folders += 1; - let mut entries: Vec = std::fs::read_dir(src) - .map_err(|e| CoreError::internal(format!("read dir {}: {e}", src.display())))? - .filter_map(|e| e.ok()) - .map(|e| e.path()) - .collect(); - entries.sort(); - for entry in entries { - if let Err(e) = - self.collect_uploads(child_ino, &child_uid, &entry, tasks, folders, job) - { - warn!(source = %entry.display(), error = %e, "skipping entry"); - } - } - // Deeper folders have retitled the job by now; put this one back so the - // line tracks the walk's position rather than its deepest leaf. - job.detail(&name); - } else if meta.is_file() { - if name.contains('/') { - return Err(CoreError::invalid(format!("invalid file name: {name:?}"))); - } - tasks.push(UploadTask { - parent_ino: pino, - parent_uid: parent_uid.clone(), - name, - path: src.to_path_buf(), - size: meta.len(), - }); - } - Ok(()) - } - // ---- activity log ----------------------------------------------------- /// Append one entry to the activity log. Callable from any thread (the sync @@ -3677,65 +2738,6 @@ fn now_ms() -> i64 { .map(|d| d.as_millis() as i64) .unwrap_or(0) } - -/// The aspect ratio (w/h) of an encoded image, read from its header alone — no -/// pixels are decoded. `None` when the format is unknown or the header is torn. -fn ratio_of(bytes: &[u8]) -> Option { - let (width, height) = image::ImageReader::new(std::io::Cursor::new(bytes)) - .with_guessed_format() - .ok()? - .into_dimensions() - .ok()?; - (height > 0).then(|| f64::from(width) / f64::from(height)) -} - -/// A thumbnail the daemon made itself, and the aspect ratio of the photo it was -/// made from (free at that point — the full image had to be decoded anyway). -struct GeneratedThumb { - bytes: Vec, - ratio: f64, -} - -/// How one attempt at generating a missing thumbnail ended. The distinction that -/// matters is *permanent* versus *transient*: only bytes we cannot decode prove -/// the photo will never have a thumbnail, and only that verdict is persisted. -enum ThumbAttempt { - Made(GeneratedThumb), - /// Decoded nothing — a format this build has no decoder for. Permanent. - Undecodable, - /// The photo couldn't be downloaded. Transient: try again next time. - Unavailable, -} - -/// Scale a full-size photo down to a thumbnail: at most [`THUMB_EDGE`] on its -/// longest side, JPEG, aspect ratio preserved. `None` when the bytes aren't an -/// image this build can decode — the caller then writes the photo off as -/// un-thumbnailable. -/// -/// CPU-bound (a 20 MP JPEG is real work), so callers run it on the blocking pool. -fn scale_thumbnail(bytes: &[u8]) -> Option { - let image = image::ImageReader::new(std::io::Cursor::new(bytes)) - .with_guessed_format() - .ok()? - .decode() - .ok()?; - let (width, height) = (image.width(), image.height()); - if width == 0 || height == 0 { - return None; - } - let ratio = f64::from(width) / f64::from(height); - - // `thumbnail` fits the image *inside* the box, so the longest edge lands on - // THUMB_EDGE and the ratio is untouched. - let thumb = image.thumbnail(THUMB_EDGE, THUMB_EDGE).to_rgb8(); - let mut bytes = Vec::new(); - image::codecs::jpeg::JpegEncoder::new_with_quality(&mut bytes, THUMB_QUALITY) - .encode_image(&thumb) - .ok()?; - Some(GeneratedThumb { bytes, ratio }) -} - -/// Convert a Unix timestamp (seconds, possibly negative) to a `SystemTime`. fn unix_secs(secs: i64) -> SystemTime { if secs >= 0 { UNIX_EPOCH + Duration::from_secs(secs as u64) @@ -3910,64 +2912,6 @@ mod replace_tests { assert!(check_replaceable(false, false, false).is_ok()); } } - -#[cfg(test)] -mod thumb_tests { - use super::{THUMB_EDGE, ratio_of, scale_thumbnail}; - - /// A `width`×`height` JPEG, standing in for a camera photo the server never - /// generated a thumbnail for. - fn jpeg(width: u32, height: u32) -> Vec { - let image = image::RgbImage::from_fn(width, height, |x, y| { - image::Rgb([(x % 256) as u8, (y % 256) as u8, 128]) - }); - let mut bytes = Vec::new(); - image::codecs::jpeg::JpegEncoder::new(&mut bytes) - .encode_image(&image) - .unwrap(); - bytes - } - - #[test] - fn scaling_fits_the_long_edge_and_keeps_the_aspect_ratio() { - let photo = jpeg(4000, 3000); - let thumb = scale_thumbnail(&photo).expect("a JPEG scales"); - - let (width, height) = image::ImageReader::new(std::io::Cursor::new(&thumb.bytes)) - .with_guessed_format() - .unwrap() - .into_dimensions() - .unwrap(); - assert_eq!(width, THUMB_EDGE, "the long edge lands on the target"); - assert_eq!(height, THUMB_EDGE * 3 / 4, "and nothing is cropped"); - assert!((thumb.ratio - 4.0 / 3.0).abs() < 1e-6); - assert!( - thumb.bytes.len() < photo.len(), - "a thumbnail that isn't smaller than its photo is no thumbnail" - ); - } - - #[test] - fn a_portrait_photo_fits_its_long_edge_too() { - let thumb = scale_thumbnail(&jpeg(1000, 2000)).expect("a JPEG scales"); - assert!(thumb.ratio < 1.0, "portrait stays portrait"); - assert_eq!(ratio_of(&thumb.bytes).map(|r| r < 1.0), Some(true)); - } - - #[test] - fn undecodable_bytes_are_not_a_thumbnail() { - // What a photo in a format this build has no decoder for looks like: the - // caller writes it off as un-thumbnailable rather than retrying forever. - assert!(scale_thumbnail(b"not an image at all").is_none()); - assert!(ratio_of(b"not an image at all").is_none()); - } - - #[test] - fn ratio_is_read_from_the_header_alone() { - assert_eq!(ratio_of(&jpeg(300, 200)), Some(1.5)); - } -} - #[cfg(test)] mod local_uid_tests { use super::*; diff --git a/crates/pdfs-fuse/src/photos.rs b/crates/pdfs-fuse/src/photos.rs index f345e2a..f70fe17 100644 --- a/crates/pdfs-fuse/src/photos.rs +++ b/crates/pdfs-fuse/src/photos.rs @@ -18,10 +18,78 @@ use proton_drive_rs::{NodeKind, ThumbnailType}; use tracing::{info, warn}; use super::{ - Core, PHOTOS_AVAILABLE, PHOTOS_SYNCED_MS, THUMB_GEN_CONCURRENCY, TIMELINE_ENRICH_CHUNK, - TIMELINE_TTL, ThumbAttempt, node_size, now_ms, parse_uid, ratio_of, scale_thumbnail, + Core, PHOTOS_AVAILABLE, PHOTOS_SYNCED_MS, TIMELINE_ENRICH_CHUNK, TIMELINE_TTL, node_size, + now_ms, parse_uid, }; +/// Longest edge, in px, of a thumbnail generated locally for a photo the server +/// has none for. Matches the server's own thumbnail scale closely enough that a +/// tile can't tell them apart. +const THUMB_EDGE: u32 = 512; +/// JPEG quality of a locally generated thumbnail. +const THUMB_QUALITY: u8 = 82; +/// How many photos may be downloaded at once to generate their missing +/// thumbnails. Bounded: a screenful of 20 MB digicam JPEGs would otherwise +/// saturate the link and starve the rest of the daemon. +const THUMB_GEN_CONCURRENCY: usize = 4; + +/// The aspect ratio (w/h) of an encoded image, read from its header alone — no +/// pixels are decoded. `None` when the format is unknown or the header is torn. +fn ratio_of(bytes: &[u8]) -> Option { + let (width, height) = image::ImageReader::new(std::io::Cursor::new(bytes)) + .with_guessed_format() + .ok()? + .into_dimensions() + .ok()?; + (height > 0).then(|| f64::from(width) / f64::from(height)) +} + +/// A thumbnail the daemon made itself, and the aspect ratio of the photo it was +/// made from (free at that point — the full image had to be decoded anyway). +struct GeneratedThumb { + bytes: Vec, + ratio: f64, +} + +/// How one attempt at generating a missing thumbnail ended. The distinction that +/// matters is *permanent* versus *transient*: only bytes we cannot decode prove +/// the photo will never have a thumbnail, and only that verdict is persisted. +enum ThumbAttempt { + Made(GeneratedThumb), + /// Decoded nothing — a format this build has no decoder for. Permanent. + Undecodable, + /// The photo couldn't be downloaded. Transient: try again next time. + Unavailable, +} + +/// Scale a full-size photo down to a thumbnail: at most [`THUMB_EDGE`] on its +/// longest side, JPEG, aspect ratio preserved. `None` when the bytes aren't an +/// image this build can decode — the caller then writes the photo off as +/// un-thumbnailable. +/// +/// CPU-bound (a 20 MP JPEG is real work), so callers run it on the blocking pool. +fn scale_thumbnail(bytes: &[u8]) -> Option { + let image = image::ImageReader::new(std::io::Cursor::new(bytes)) + .with_guessed_format() + .ok()? + .decode() + .ok()?; + let (width, height) = (image.width(), image.height()); + if width == 0 || height == 0 { + return None; + } + let ratio = f64::from(width) / f64::from(height); + + // `thumbnail` fits the image *inside* the box, so the longest edge lands on + // THUMB_EDGE and the ratio is untouched. + let thumb = image.thumbnail(THUMB_EDGE, THUMB_EDGE).to_rgb8(); + let mut bytes = Vec::new(); + image::codecs::jpeg::JpegEncoder::new_with_quality(&mut bytes, THUMB_QUALITY) + .encode_image(&thumb) + .ok()?; + Some(GeneratedThumb { bytes, ratio }) +} + impl Core { pub(crate) fn photos_timeline( &self, @@ -361,3 +429,60 @@ impl Core { Ok(self.cache.content_path(uid)) } } + +#[cfg(test)] +mod thumb_tests { + use super::{THUMB_EDGE, ratio_of, scale_thumbnail}; + + /// A `width`×`height` JPEG, standing in for a camera photo the server never + /// generated a thumbnail for. + fn jpeg(width: u32, height: u32) -> Vec { + let image = image::RgbImage::from_fn(width, height, |x, y| { + image::Rgb([(x % 256) as u8, (y % 256) as u8, 128]) + }); + let mut bytes = Vec::new(); + image::codecs::jpeg::JpegEncoder::new(&mut bytes) + .encode_image(&image) + .unwrap(); + bytes + } + + #[test] + fn scaling_fits_the_long_edge_and_keeps_the_aspect_ratio() { + let photo = jpeg(4000, 3000); + let thumb = scale_thumbnail(&photo).expect("a JPEG scales"); + + let (width, height) = image::ImageReader::new(std::io::Cursor::new(&thumb.bytes)) + .with_guessed_format() + .unwrap() + .into_dimensions() + .unwrap(); + assert_eq!(width, THUMB_EDGE, "the long edge lands on the target"); + assert_eq!(height, THUMB_EDGE * 3 / 4, "and nothing is cropped"); + assert!((thumb.ratio - 4.0 / 3.0).abs() < 1e-6); + assert!( + thumb.bytes.len() < photo.len(), + "a thumbnail that isn't smaller than its photo is no thumbnail" + ); + } + + #[test] + fn a_portrait_photo_fits_its_long_edge_too() { + let thumb = scale_thumbnail(&jpeg(1000, 2000)).expect("a JPEG scales"); + assert!(thumb.ratio < 1.0, "portrait stays portrait"); + assert_eq!(ratio_of(&thumb.bytes).map(|r| r < 1.0), Some(true)); + } + + #[test] + fn undecodable_bytes_are_not_a_thumbnail() { + // What a photo in a format this build has no decoder for looks like: the + // caller writes it off as un-thumbnailable rather than retrying forever. + assert!(scale_thumbnail(b"not an image at all").is_none()); + assert!(ratio_of(b"not an image at all").is_none()); + } + + #[test] + fn ratio_is_read_from_the_header_alone() { + assert_eq!(ratio_of(&jpeg(300, 200)), Some(1.5)); + } +} diff --git a/crates/pdfs-fuse/src/reads.rs b/crates/pdfs-fuse/src/reads.rs new file mode 100644 index 0000000..71a1ca0 --- /dev/null +++ b/crates/pdfs-fuse/src/reads.rs @@ -0,0 +1,645 @@ +//! Revision-reader cache and streamed range-read subsystem. + +use super::*; + +/// A read of an unpinned video at least this large streams *without* persisting +/// its blocks to the on-disk cache. Playing a 2 GB film would otherwise pour it +/// through the block LRU and evict everything else the user actually wants kept — +/// and it re-streams cheaply enough on a rewatch that keeping it was never worth +/// that. Pinned videos (kept offline on purpose) and anything smaller cache as +/// usual. +pub(super) const STREAM_BYPASS_MIN: u64 = 256 * 1024 * 1024; + +/// How many bytes of streamed blocks the in-memory ring keeps. Bypassing the +/// on-disk cache must not mean re-fetching: the kernel asks for a streamed file +/// in reads far smaller than [`BLOCK_SIZE`], so without a ring every 128 KiB the +/// player consumes would download and decrypt a whole 4 MiB block again. Sized +/// for a handful of blocks per concurrently-streamed file. +const STREAM_RING_BYTES: u64 = 128 * 1024 * 1024; + +/// Blocks fetched *past* the one a sequential streaming read needed, warmed into +/// the ring in the background so the player's next read is already in memory +/// instead of waiting a round-trip. Kept small so it cannot crowd out the +/// in-flight block governor. +const STREAM_READAHEAD: u64 = 4; + +/// How many [`RevisionReader`]s stay open at once. +/// +/// A reader holds its revision's content key and block table — a few KB even for +/// a large file, so this is bounded for tidiness and staleness rather than +/// memory. Evicted least-recently-used; a dropped reader costs one re-open +/// (two API calls and a node-key unlock) the next time that file is read. +const MAX_OPEN_READERS: usize = 64; + +/// An open reader plus the node metadata it was opened against. +/// +/// The SDK pins a reader to the revision that was active at `open_revision`, so +/// a reader is only reusable while the node still reports the same +/// `(mtime, size)` — the same validity pair the content cache uses (a new +/// revision bumps mtime). On a mismatch the reader is dropped and reopened. +pub(super) struct CachedReader { + reader: Arc, + mtime: i64, + size: u64, + /// For LRU eviction. + last_used: Instant, +} + +/// What the `readers` map holds for a node: an open reader, or the fact that +/// someone is already opening one. +/// +/// The `Pending` arm is what makes the map single-flight rather than merely a +/// cache. Opening resolves link details, ancestor keys, an S2K node-key unlock +/// and the block table — five round-trips — and read-ahead on a cold file puts +/// several workers into the open at the same moment, so without this each +/// racer would redo all of it and discard every result but one. +pub(super) enum ReaderSlot { + Ready(CachedReader), + Pending(PendingOpen), +} + +/// An `open_revision` in flight. Racers clone `rx` and await the leader's result +/// instead of opening their own. +pub(super) struct PendingOpen { + /// Identifies this attempt so the leader can tell "my slot is still there" + /// from "someone evicted or replaced it while I was on the network", and + /// only publish its reader in the first case. + id: u64, + /// What the leader is opening against. A racer wanting a different revision + /// must not join this open — it would get a reader for the wrong bytes. + mtime: i64, + size: u64, + rx: tokio::sync::watch::Receiver, Errno>>>, +} + +/// Distinguishes `PendingOpen`s. Only uniqueness matters, so a plain counter +/// does; ids are never compared for order. +static NEXT_OPEN_ID: AtomicU64 = AtomicU64::new(0); + +/// Drive client, a Tokio handle to bridge the synchronous FUSE/socket threads +/// to the async SDK, the inode bookkeeping, and the on-disk content cache. +/// +/// Cheaply cloneable (every field is a handle/`Arc`), so the control-socket task +/// gets its own copy while the FUSE session keeps another. +/// In-memory LRU of blocks belonging to a file streaming past the on-disk cache +/// (see [`STREAM_BYPASS_MIN`]). Bypassing the disk stops one film evicting the +/// block LRU; it must not also mean the same 4 MiB block is downloaded once per +/// 128 KiB the player reads out of it. +/// +/// 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. +#[derive(Default)] +pub(super) struct StreamRing { + blocks: HashMap<(NodeUid, u64), Arc>>, + /// Keys oldest-first, for eviction. + order: VecDeque<(NodeUid, u64)>, + /// Per-node validity tag, `(mtime, size)`. + tags: HashMap, + bytes: u64, +} + +impl StreamRing { + fn get(&mut self, uid: &NodeUid, mtime: i64, size: u64, idx: u64) -> Option>> { + if self.tags.get(uid) != Some(&(mtime, size)) { + return None; + } + self.blocks.get(&(uid.clone(), idx)).cloned() + } + + fn insert(&mut self, uid: &NodeUid, mtime: i64, size: u64, idx: u64, 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); + if self.blocks.contains_key(&key) { + return; + } + self.bytes += bytes.len() as u64; + self.blocks.insert(key.clone(), bytes); + self.order.push_back(key); + while self.bytes > STREAM_RING_BYTES { + let Some(victim) = self.order.pop_front() else { + break; + }; + if let Some(dropped) = self.blocks.remove(&victim) { + self.bytes -= dropped.len() as u64; + } + } + } + + fn drop_node(&mut self, uid: &NodeUid) { + self.order.retain(|(u, _)| u != uid); + let mut freed = 0u64; + self.blocks.retain(|(u, _), bytes| { + if u == uid { + freed += bytes.len() as u64; + false + } else { + true + } + }); + self.bytes -= freed; + self.tags.remove(uid); + } +} + +impl Core { + /// An open [`RevisionReader`] for `uid`, reusing the cached one when it is + /// still valid for `(mtime, fsize)` and opening a fresh one otherwise. + /// + /// Opening resolves the file's link details, ancestor keys, node key (an S2K + /// unlock) and block table. Doing that once per file rather than once per + /// block is the whole point: a cold 100 MB read is 25 block misses, which + /// used to mean 25 full resolutions (50 API calls, 25 unlocks) and now means + /// one. + /// + /// Racing reads of one cold file open it **once**: the first caller claims + /// the slot and the rest await its result. That matters now that `read` is + /// served off the dispatch loop, because read-ahead on a first touch issues + /// several parallel reads of the same file as a matter of course. + async fn revision_reader( + &self, + uid: &NodeUid, + mtime: i64, + fsize: u64, + ) -> Result, Errno> { + loop { + // Decide under the lock, act outside it — network I/O must never run + // under a std `Mutex`, and an await must never hold one. + enum Act { + Hit(Arc), + Join(tokio::sync::watch::Receiver, Errno>>>), + Lead( + u64, + tokio::sync::watch::Sender, Errno>>>, + ), + } + let act = { + let mut readers = self.readers.lock(); + match readers.get_mut(uid) { + Some(ReaderSlot::Ready(entry)) + if entry.mtime == mtime && entry.size == fsize => + { + entry.last_used = Instant::now(); + Act::Hit(entry.reader.clone()) + } + Some(ReaderSlot::Pending(p)) if p.mtime == mtime && p.size == fsize => { + Act::Join(p.rx.clone()) + } + // Either nothing here, or something for a revision we no + // longer want (a stale reader, or an open for one). Ours + // replaces it: the newer `(mtime, size)` is the truth. + _ => { + let id = NEXT_OPEN_ID.fetch_add(1, Ordering::Relaxed); + let (tx, rx) = tokio::sync::watch::channel(None); + readers.insert( + uid.clone(), + ReaderSlot::Pending(PendingOpen { + id, + mtime, + size: fsize, + rx, + }), + ); + Act::Lead(id, tx) + } + } + }; + + match act { + Act::Hit(reader) => return Ok(reader), + Act::Join(mut rx) => { + // `changed()` errors only if the leader dropped its sender + // without publishing — it panicked. Retry from the top + // rather than fail: the slot it left is gone, so this pass + // leads its own open. + loop { + if let Some(result) = rx.borrow_and_update().clone() { + return result; + } + if rx.changed().await.is_err() { + break; + } + } + continue; + } + Act::Lead(id, tx) => { + debug!(%uid, mtime, fsize, "opening revision"); + let result = match self.client.open_revision(uid).await { + Ok(reader) => Ok(Arc::new(reader)), + Err(e) => { + warn!(%uid, error = %e, "open_revision failed"); + Err(Errno::EIO) + } + }; + { + let mut readers = self.readers.lock(); + // Publish only into the slot we claimed. If it is gone + // or belongs to a later attempt, an eviction or a newer + // revision overtook us: our reader still answers the + // callers waiting on it, but it must not be cached. + let ours = matches!( + readers.get(uid), + Some(ReaderSlot::Pending(p)) if p.id == id + ); + if ours { + match &result { + Ok(reader) => { + readers.insert( + uid.clone(), + ReaderSlot::Ready(CachedReader { + reader: reader.clone(), + mtime, + size: fsize, + last_used: Instant::now(), + }), + ); + } + // Never cache a failure: the next read retries. + Err(_) => { + readers.remove(uid); + } + } + } + Self::trim_readers(&mut readers); + } + // After the map is updated, so a racer that wakes and + // re-enters finds the finished slot rather than racing us. + let _ = tx.send(Some(result.clone())); + return result; + } + } + } + } + + /// Bound the reader map: drop least-recently-used entries until it is back + /// at [`MAX_OPEN_READERS`]. Only `Ready` slots are eviction candidates — a + /// `Pending` has callers waiting on it and no reader to drop yet. + fn trim_readers(readers: &mut HashMap) { + while readers.len() > MAX_OPEN_READERS { + let victim = readers + .iter() + .filter_map(|(uid, slot)| match slot { + ReaderSlot::Ready(entry) => Some((uid, entry.last_used)), + ReaderSlot::Pending(_) => None, + }) + .min_by_key(|(_, last_used)| *last_used) + .map(|(uid, _)| uid.clone()); + let Some(victim) = victim else { + break; // every slot is an open in flight; nothing to drop + }; + readers.remove(&victim); + } + } + + /// Drop any open reader for `uid`, so the next read reopens against the + /// current revision. Called wherever cached content is evicted. + /// + /// Drops an in-flight open too: its leader finds the slot gone and declines + /// to cache what it opened, which is what we want — the eviction says that + /// revision is no longer the truth. + pub(super) fn evict_reader(&self, uid: &NodeUid) { + self.readers.lock().remove(uid); + } + + /// 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. + pub(super) fn read_range( + &self, + uid: &NodeUid, + mtime: i64, + fsize: u64, + offset: u64, + len: u64, + cache_blocks: bool, + ) -> Result, Errno> { + // A queued write has not reached the remote yet, so the remote's current + // revision is stale and the staged blob is the truth. Serve from it until + // the drain worker lands the upload (offline.md Phase 3). + if let Some(pending) = self.pending.lock().get(uid).cloned() { + return self.read_pending(&pending, offset, len); + } + // A node created offline and never written has no blob and no remote: it + // is an empty file, and asking the API about a `local~` uid would only + // earn a 404 (offline.md Phase 3b). + if is_local_uid(uid) { + return Ok(Vec::new()); + } + self.read_range_remote(uid, mtime, fsize, offset, len, cache_blocks) + } + + /// Serve a read from a staged blob, falling back to the remote base for any + /// range the write did not author (an incomplete [`StagedWrite`] holds zeros + /// there, which must never be handed out as content). + fn read_pending( + &self, + pending: &PendingRevision, + offset: u64, + len: u64, + ) -> Result, Errno> { + let m = &pending.meta; + if offset >= m.len || len == 0 { + return Ok(Vec::new()); + } + let uid = parse_node_uid(&m.uid).ok_or(Errno::EIO)?; + let file = File::open(&pending.path).map_err(|e| { + error!(%uid, path = %pending.path.display(), error = %e, "open staged blob failed"); + Errno::EIO + })?; + let mut written = Intervals::default(); + for &(s, e) in &m.authored { + written.add(s, e); + } + // Same merge as `serve_open_read`, but resolving gaps against the remote + // rather than through `read_range` — going through `read_range` would find + // this very pending op and recurse. + let end = offset.saturating_add(len).min(m.len); + let mut out = Vec::with_capacity((end - offset) as usize); + for (s, e, authored) in written.segments(offset, end) { + if authored { + let mut buf = vec![0u8; (e - s) as usize]; + file.read_exact_at(&mut buf, s).map_err(|err| { + warn!(%uid, error = %err, "staged blob read failed"); + Errno::EIO + })?; + out.extend_from_slice(&buf); + continue; + } + let bend = e.min(m.base_size); + if s < bend { + out.extend_from_slice(&self.read_range_remote( + &uid, + m.base_mtime, + m.base_size, + s, + bend - s, + true, + )?); + } + // Past the base: a hole the write extended over. + out.resize(out.len() + e.saturating_sub(s.max(m.base_size)) as usize, 0); + } + Ok(out) + } + + /// Warm the [`STREAM_READAHEAD`] blocks after `last` into the ring, in the + /// background, so a player reading forward finds its next block already in + /// memory rather than paying a fetch + decrypt round-trip at each boundary. + /// + /// Fire-and-forget: a failure here is not an error, it only means the read + /// that needs those bytes will fetch them itself. + fn stream_readahead(&self, uid: &NodeUid, mtime: i64, fsize: u64, last: u64) { + let last_block = (fsize.saturating_sub(1)) / BLOCK_SIZE; + let wanted: Vec = ((last + 1)..=(last + STREAM_READAHEAD).min(last_block)) + .filter(|&bidx| { + self.stream_ring + .lock() + .get(uid, mtime, fsize, bidx) + .is_none() + }) + .collect(); + if wanted.is_empty() { + return; + } + let core = self.clone(); + let uid = uid.clone(); + self.rt.spawn(async move { + let Ok(reader) = core.revision_reader(&uid, mtime, fsize).await else { + return; + }; + // Concurrently, for the same reason the demand path fetches its + // misses concurrently: serially awaited blocks cost the sum of their + // round-trips, and the player catches up with the readahead before it + // finishes. The SDK's in-flight block governor bounds the fan-out. + let mut set = tokio::task::JoinSet::new(); + for bidx in wanted { + let reader = reader.clone(); + let bstart = bidx * BLOCK_SIZE; + let blen = BLOCK_SIZE.min(fsize - bstart); + set.spawn(async move { (bidx, reader.read_at(bstart, blen).await) }); + } + while let Some(joined) = set.join_next().await { + let Ok((bidx, read)) = joined else { continue }; + match read { + Ok(bytes) => { + core.stream_ring + .lock() + .insert(&uid, mtime, fsize, bidx, Arc::new(bytes)) + } + Err(e) => debug!(%uid, bidx, error = %e, "stream readahead failed"), + } + } + }); + } + + /// Read from the content cache, else the remote. The base-content path, with + /// no awareness of queued writes — callers wanting the file's *current* + /// content want [`Core::read_range`]. Gap-filling a staged blob is the one + /// caller that genuinely means "the base", since that is what its zeroed + /// ranges have to be filled from. + fn read_range_remote( + &self, + uid: &NodeUid, + mtime: i64, + fsize: u64, + offset: u64, + len: u64, + cache_blocks: bool, + ) -> Result, Errno> { + if let Some(bytes) = self.cache.read_range(uid, mtime, fsize, offset, len) { + return Ok(bytes); + } + if offset >= fsize || len == 0 { + return Ok(Vec::new()); + } + 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; + + // Collect the blocks overlapping the request, serving any already cached + // and fetching the rest concurrently. A multi-block read (e.g. a media + // player buffering, or a large sequential read split into one FUSE call) + // 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 { + // A streaming read keeps its blocks only in memory, so the ring is + // the only cache it has — check it before the disk (which will not + // hold them) and before the network (which would refetch the same + // 4 MiB block for every 128 KiB the player consumes). + let hit = if cache_blocks { + self.cache + .cached_block(uid, mtime, fsize, bidx) + .map(Arc::new) + } else { + self.stream_ring.lock().get(uid, mtime, fsize, bidx) + }; + match hit { + Some(b) => blocks.push(Some(b)), + None => { + blocks.push(None); + misses.push(bidx); + } + } + } + + if !misses.is_empty() { + let fetched = self.rt.block_on(async { + // Resolve the file's keys and block table once, then read every + // missing block through the shared reader. Previously each block + // called `download_range`, which redid that resolution per block. + let reader = self.revision_reader(uid, mtime, fsize).await?; + + let mut set = tokio::task::JoinSet::new(); + for &bidx in &misses { + let reader = reader.clone(); + let uid = uid.clone(); + let bstart = bidx * BLOCK_SIZE; + let blen = BLOCK_SIZE.min(fsize - bstart); + set.spawn(async move { + reader + .read_at(bstart, blen) + .await + .map(|bytes| (bidx, bytes)) + .map_err(|e| { + warn!(%uid, bstart, blen, error = %e, "block read failed"); + Errno::EIO + }) + }); + } + let mut out = Vec::with_capacity(misses.len()); + while let Some(joined) = set.join_next().await { + // A join error means the task panicked; surface it as EIO. + out.push(joined.map_err(|_| Errno::EIO)??); + } + Ok::<_, Errno>(out) + })?; + for (bidx, bytes) in fetched { + // A streaming read (large unpinned video) skips the on-disk cache + // so it can't evict the rest of it; it keeps the block in the + // in-memory ring instead, which is bounded and dies with the read. + let bytes = Arc::new(bytes); + if cache_blocks { + let _ = self.cache.store_block(uid, mtime, fsize, bidx, &bytes); + } else { + self.stream_ring + .lock() + .insert(uid, mtime, fsize, bidx, bytes.clone()); + } + blocks[(bidx - first) as usize] = Some(bytes); + } + if !cache_blocks { + self.stream_readahead(uid, mtime, fsize, last); + } + } + + for (i, block) in blocks.into_iter().enumerate() { + let bidx = first + i as u64; + let bstart = bidx * BLOCK_SIZE; + // Every slot is populated: cache hits up front, misses by the fetch above. + let block = block.expect("block fetched or cached"); + let s = (offset.max(bstart) - bstart) as usize; + let e = (end.min(bstart + block.len() as u64) - bstart) as usize; + if s < e { + out.extend_from_slice(&block[s..e]); + } + } + Ok(out) + } + + /// Serve a read against an open write handle: stitch authored ranges (from + /// the scratch file) and untouched ranges (from the remote base via the + /// block cache) in order, zero-filling any region past the base. + #[allow(clippy::too_many_arguments)] + pub(super) fn serve_open_read( + &self, + file: &Arc, + len: u64, + uid: &NodeUid, + base_mtime: i64, + base_size: u64, + written: &Intervals, + offset: u64, + size: u64, + ) -> Result, Errno> { + if offset >= len || size == 0 { + return Ok(Vec::new()); + } + let end = offset.saturating_add(size).min(len); + let mut out = Vec::with_capacity((end - offset) as usize); + for (s, e, authored) in written.segments(offset, end) { + if authored { + let mut buf = vec![0u8; (e - s) as usize]; + file.read_exact_at(&mut buf, s).map_err(|err| { + warn!(%uid, error = %err, "scratch read failed"); + Errno::EIO + })?; + out.extend_from_slice(&buf); + } else { + let bend = e.min(base_size); + if s < bend { + out.extend_from_slice(&self.read_range( + uid, + base_mtime, + base_size, + s, + bend - s, + true, + )?); + } + // Anything past the base is a hole: zero-fill. + let zeros = e.saturating_sub(s.max(base_size)); + out.resize(out.len() + zeros as usize, 0); + } + } + Ok(out) + } + + /// Fill every unauthored range of a scratch/staged file that overlaps its + /// base with the base's bytes, so the file becomes the complete new content. + /// + /// This is the step a partial overwrite cannot skip: only the authored bytes + /// were ever written to disk, and a revision upload sends the whole file. + /// The gaps come from the *remote base* (through the block cache, so a small + /// edit of a large file does not pull all of it), which is exactly why this + /// can fail with no network — see `StagedWrite`. + pub(super) fn fill_gaps( + &self, + uid: &NodeUid, + file: &File, + len: u64, + base_mtime: i64, + base_size: u64, + written: &Intervals, + ) -> Result<(), Errno> { + file.set_len(len).map_err(|e| { + error!(%uid, error = %e, "resize scratch file failed"); + Errno::EIO + })?; + for (s, e, authored) in written.segments(0, len) { + if authored { + continue; + } + let bend = e.min(base_size); + if s >= bend { + continue; // wholly past the base: already zero-filled on disk + } + let bytes = self.read_range_remote(uid, base_mtime, base_size, s, bend - s, true)?; + file.write_all_at(&bytes, s).map_err(|err| { + error!(%uid, error = %err, "scratch gap-fill write failed"); + Errno::EIO + })?; + } + Ok(()) + } +} diff --git a/crates/pdfs-fuse/src/sync.rs b/crates/pdfs-fuse/src/sync.rs index 7727daf..c1dd82e 100644 --- a/crates/pdfs-fuse/src/sync.rs +++ b/crates/pdfs-fuse/src/sync.rs @@ -42,613 +42,23 @@ use pdfs_core::control::{ActivityKind, TransferDirection}; use pdfs_core::db::{StoredSyncEntry, StoredSyncFolder}; use pdfs_core::syncignore::IgnoreRules; -/// How long the watcher must go quiet before a burst counts as finished. -/// -/// Measured from the *last* event, not the first: applications do not write a -/// file once. An editor saves by writing a temp file and renaming it over the -/// target; an export or a database dump writes continuously for as long as it -/// takes. Settling a fixed interval after the first event walks the tree while -/// the file is still growing, and uploads a torn snapshot as a real revision — -/// corrected on the next pass, but only after spending the encrypt-and-upload -/// on a file that was never in that state. -const DEBOUNCE: Duration = Duration::from_secs(2); -/// Ceiling on the settle wait, so a directory under sustained change still -/// syncs. Without it, copying a large tree in — events arriving forever — -/// would postpone the reconcile for as long as the copy ran. -const MAX_SETTLE: Duration = Duration::from_secs(30); -/// How often to re-walk remotes so changes made on other clients are pulled in. -const POLL_INTERVAL: Duration = Duration::from_secs(120); -/// How many uploads/downloads/folder-creations a reconcile runs at once. The -/// per-item work is a network round-trip, so a folder with thousands of files is -/// unusable done one-at-a-time; this bounds concurrency like the bulk-upload -/// engine ([`crate::run_uploads`]) does. -const SYNC_CONCURRENCY: usize = 8; - -/// A message to the sync engine's control thread. -pub(crate) enum SyncMsg { - /// Reconcile one folder by id (a local change, or a targeted `SyncNow`). - Reconcile(i64), - /// Reconcile every mirror folder (periodic poll, or startup). - ReconcileAll, - /// Re-read the folder set and adjust the filesystem watches (after add/remove). - Rewatch, -} - -/// Start the sync engine: a control thread that owns the filesystem watcher and -/// serialises reconcile passes, plus a periodic poll thread. The engine receives -/// on `rx`; senders live in [`Core::sync_tx`]. -pub(crate) fn spawn(core: Core, rx: Receiver) { - if let Err(e) = std::thread::Builder::new() - .name("pdfs-sync".into()) - .spawn(move || engine_loop(core, rx)) - { - warn!(error = %e, "failed to start sync engine"); - } -} - -/// The engine control loop. Single-threaded so reconcile passes never overlap. -fn engine_loop(core: Core, rx: Receiver) { - // Paths the watcher currently covers, mapped to their folder id, so an event - // path can be resolved back to the folder it belongs to. - let watched: Mutex> = Mutex::new(Vec::new()); - - let tx_events = core.sync_tx.clone(); - let mut watcher = - match notify::recommended_watcher(move |res: notify::Result| { - let Ok(event) = res else { return }; - // Ignore pure metadata/access noise; content changes are what matter. - if matches!(event.kind, notify::EventKind::Access(_)) { - return; - } - let _ = tx_events.send(SyncMsg::ReconcileAll); - }) { - Ok(w) => w, - Err(e) => { - warn!(error = %e, "filesystem watcher unavailable; sync is poll-only"); - // Still run the loop so polling and explicit SyncNow work. - poll_only_loop(&core, rx); - return; - } - }; - - // Periodic remote poll. - { - let tx = core.sync_tx.clone(); - std::thread::Builder::new() - .name("pdfs-sync-poll".into()) - .spawn(move || { - loop { - std::thread::sleep(POLL_INTERVAL); - if tx.send(SyncMsg::ReconcileAll).is_err() { - break; - } - } - }) - .ok(); - } - - rewatch(&core, &mut watcher, &watched); - reconcile_all(&core); - - while let Ok(msg) = rx.recv() { - let mut ids: HashSet = HashSet::new(); - let mut all = false; - let mut do_rewatch = false; - classify(msg, &mut ids, &mut all, &mut do_rewatch); - settle(&rx, &mut ids, &mut all, &mut do_rewatch); - - if do_rewatch { - rewatch(&core, &mut watcher, &watched); - } - if all { - reconcile_all(&core); - } else { - for id in ids { - if let Ok(Some(folder)) = core.db.sync_folder_get(id) { - core.reconcile_folder(&folder); - } - } - } - } -} - -/// Fallback loop when no filesystem watcher could be created: poll + SyncNow only. -fn poll_only_loop(core: &Core, rx: Receiver) { - { - let tx = core.sync_tx.clone(); - std::thread::spawn(move || { - loop { - std::thread::sleep(POLL_INTERVAL); - if tx.send(SyncMsg::ReconcileAll).is_err() { - break; - } - } - }); - } - reconcile_all(core); - while let Ok(msg) = rx.recv() { - let mut ids: HashSet = HashSet::new(); - let mut all = false; - let mut do_rewatch = false; - classify(msg, &mut ids, &mut all, &mut do_rewatch); - settle(&rx, &mut ids, &mut all, &mut do_rewatch); - if all { - reconcile_all(core); - } else { - for id in ids { - if let Ok(Some(folder)) = core.db.sync_folder_get(id) { - core.reconcile_folder(&folder); - } - } - } - } -} - -/// Absorb the rest of an event burst, returning once the watcher has been quiet -/// for `quiet` or `cap` has elapsed since the first event — whichever comes -/// first. The caller has already classified the event that opened the burst. -/// -/// Split out with its timings as parameters so the settling behaviour can be -/// tested in milliseconds rather than in the tens of seconds the real constants -/// describe. -fn settle_with( - rx: &Receiver, - ids: &mut HashSet, - all: &mut bool, - do_rewatch: &mut bool, - quiet: Duration, - cap: Duration, -) { - let deadline = Instant::now() + cap; - loop { - let now = Instant::now(); - if now >= deadline { - break; - } - // Never wait past the cap, however quiet the watcher goes. - let wait = quiet.min(deadline - now); - match rx.recv_timeout(wait) { - // Another event: the burst is still going, so the quiet window - // starts again from here. - Ok(m) => classify(m, ids, all, do_rewatch), - // Quiet for a full window: whatever was being written has stopped. - Err(RecvTimeoutError::Timeout) => break, - // Every sender is gone; the caller's loop is about to end anyway. - Err(RecvTimeoutError::Disconnected) => break, - } - } -} - -/// [`settle_with`] at the real timings. -fn settle(rx: &Receiver, ids: &mut HashSet, all: &mut bool, do_rewatch: &mut bool) { - settle_with(rx, ids, all, do_rewatch, DEBOUNCE, MAX_SETTLE); -} - -fn classify(msg: SyncMsg, ids: &mut HashSet, all: &mut bool, rewatch: &mut bool) { - match msg { - SyncMsg::Reconcile(id) => { - ids.insert(id); - } - SyncMsg::ReconcileAll => *all = true, - SyncMsg::Rewatch => *rewatch = true, - } -} - -/// Reconcile every mirror folder in turn. -fn reconcile_all(core: &Core) { - let folders = match core.db.sync_folder_list() { - Ok(f) => f, - Err(e) => { - warn!(error = ?e, "sync: cannot list folders"); - return; - } - }; - for folder in folders { - core.reconcile_folder(&folder); - } -} - -/// Bring the filesystem watches in line with the current mirror-folder set: -/// watch newly-added folders, drop removed ones. -fn rewatch( - core: &Core, - watcher: &mut notify::RecommendedWatcher, - watched: &Mutex>, -) { - let want: Vec<(PathBuf, i64)> = match core.db.sync_folder_list() { - Ok(f) => f - .into_iter() - .filter(|f| f.mode == "mirror") - .map(|f| (PathBuf::from(f.local_path), f.id)) - .collect(), - Err(e) => { - warn!(error = ?e, "sync: cannot list folders to watch"); - return; - } - }; - let mut have = watched.lock(); - // Drop watches no longer wanted. - have.retain(|(path, _)| { - if want.iter().any(|(p, _)| p == path) { - true - } else { - let _ = watcher.unwatch(path); - false - } - }); - // Add newly wanted watches. - for (path, id) in &want { - if have.iter().any(|(p, _)| p == path) { - continue; - } - match watcher.watch(path, RecursiveMode::Recursive) { - Ok(()) => have.push((path.clone(), *id)), - Err(e) => warn!(path = %path.display(), error = %e, "sync: watch failed"), - } - } -} - -// ---- reconcile ------------------------------------------------------------ - -/// One item found while walking a local tree. -struct LocalItem { - is_dir: bool, - mtime: i64, - size: i64, - /// True when another process holds this file open for writing (detected via - /// `/proc/*/fd`). An open-for-write file is kept in the map — so it is not - /// misread as a deletion — but treated as unchanged for upload purposes, - /// deferring the upload until the writer closes the file. - open_for_write: bool, -} - -/// One item found while walking a remote tree. -struct RemoteItem { - uid: NodeUid, - is_dir: bool, - mtime: i64, - size: i64, -} - -/// The result of a reconcile pass: what it moved, how many paths were kept as -/// conflict copies, and how many failed to apply (and so still need another -/// pass). The counts drive both the folder's state and its activity summary. -#[derive(Default)] -struct Outcome { - uploaded: usize, - downloaded: usize, - created: usize, - deleted: usize, - conflicts: usize, - errors: usize, - /// Files skipped because another process held them open for writing. - deferred: usize, -} - -impl Outcome { - /// Fold in one applied op. - fn record(&mut self, applied: &Applied) { - match applied { - Applied::Dir(..) => self.created += 1, - Applied::Uploaded => self.uploaded += 1, - Applied::Downloaded => self.downloaded += 1, - Applied::Conflict => self.conflicts += 1, - } - } - - /// Whether the pass moved nothing at all — the common case on a poll of an - /// unchanged folder, which should not add a line to the activity feed. - fn is_empty(&self) -> bool { - self.uploaded == 0 - && self.downloaded == 0 - && self.created == 0 - && self.deleted == 0 - && self.conflicts == 0 - && self.errors == 0 - && self.deferred == 0 - } - - /// A human summary of the pass: "3 uploaded, 1 downloaded, 2 failed". - fn summary(&self) -> String { - let mut parts = Vec::new(); - for (n, label) in [ - (self.uploaded, "uploaded"), - (self.downloaded, "downloaded"), - (self.created, "folder(s) created"), - (self.deleted, "deleted"), - (self.conflicts, "conflicted"), - (self.deferred, "deferred (open for write)"), - (self.errors, "failed"), - ] { - if n > 0 { - parts.push(format!("{n} {label}")); - } - } - parts.join(", ") - } -} - -/// A network operation queued during classification and run concurrently in a -/// per-depth batch. Parent uids are resolved up front (the parent folder is one -/// depth shallower and already created), so tasks share nothing mutable. -enum Pending { - /// Create a new remote folder under `parent`. - CreateDir { rel: String, parent: NodeUid }, - /// Upload a brand-new local file into `parent`. - UploadNew { rel: String, parent: NodeUid }, - /// Upload a changed local file as a new revision of `uid`. - UploadRevision { rel: String, uid: NodeUid }, - /// Download remote `uid` to the local path, stamping `mtime`. `size` is the - /// remote's reported size, used as the transfer's expected total. - Download { - rel: String, - uid: NodeUid, - mtime: i64, - size: i64, - }, - /// Both sides changed: set the local copy aside, then download remote `uid`. - Conflict { - rel: String, - uid: NodeUid, - mtime: i64, - size: i64, - }, - /// Both sides changed during a [`push pass`](Core::push_pass): upload the local - /// copy into `parent` under a conflict name, leaving the remote file as it is. - PushConflict { rel: String, parent: NodeUid }, -} - -impl Pending { - /// The path this op acts on, relative to the folder root. - fn rel(&self) -> &str { - match self { - Pending::CreateDir { rel, .. } - | Pending::UploadNew { rel, .. } - | Pending::UploadRevision { rel, .. } - | Pending::Download { rel, .. } - | Pending::Conflict { rel, .. } - | Pending::PushConflict { rel, .. } => rel, - } - } - - /// How this op reads in the activity feed. - fn kind(&self) -> ActivityKind { - match self { - Pending::CreateDir { .. } => ActivityKind::CreateFolder, - Pending::UploadNew { .. } - | Pending::UploadRevision { .. } - | Pending::PushConflict { .. } => ActivityKind::Upload, - Pending::Download { .. } | Pending::Conflict { .. } => ActivityKind::Download, - } - } - - /// The activity line's detail, which distinguishes ops that share a kind. - fn detail(&self) -> &'static str { - match self { - Pending::CreateDir { .. } => "on Drive", - Pending::UploadNew { .. } => "new file", - Pending::UploadRevision { .. } => "new version", - Pending::Download { .. } => "from Drive", - Pending::Conflict { .. } => "local changes kept as a conflict copy", - Pending::PushConflict { .. } => "local changes uploaded as a conflict copy", - } - } -} - -/// Why a full reconcile pass stopped without a result. -enum PassAbort { - /// An on-demand switch was queued while the pass was running. Everything left - /// to do is work on a local copy about to be evicted, so the pass gives up its - /// remaining work to a [`push pass`](Core::push_pass) instead. Not a failure — - /// nothing is left half-applied, because each step is applied whole. - Interrupted, - /// The pass could not establish its diff (a walk or the baseline load failed). - Failed(String), -} - -impl From for PassAbort { - fn from(e: String) -> Self { - PassAbort::Failed(e) - } -} - -/// Refuse to run a pass whose local side has vanished in its entirety. -/// -/// A path in the baseline with no local copy is classified as "the user deleted -/// this", and the pass answers by trashing it on Drive. That is right for one -/// file and right for a handful. When it is true of *every* path the baseline -/// knows, the likely cause is not a user: it is an unmounted external disk, a -/// mountpoint whose FUSE mount died, a home directory that was not yet decrypted -/// when the daemon started, or a bug in this client — and the answer to all of -/// those is to trash the folder's entire contents on Drive. -/// -/// So this pass abort exists to cap the blast radius rather than to catch a -/// specific defect. Losing a pass to it costs a retry; the failure it prevents -/// costs the folder. A user who really did empty the folder can express that by -/// removing it from sync and re-adding it — the baseline goes with it. -/// -/// One surviving path is enough to clear the guard, since the classification is -/// only ever wrong about all of them at once. -/// The paths a reconcile pass will classify: the union of the local, remote, -/// and baseline states, minus anything the ignore rules exclude, shallowest -/// first so a parent folder is created before its children are placed in it. -/// -/// The ignore filter belongs *here*, on the union, rather than on the local walk -/// alone — filtering the walk alone would be actively destructive. A file synced -/// before it became ignored is absent from `local` but present in `remote` and -/// `baseline`, which is precisely the "local deleted, remote untouched" shape -/// `do_reconcile` responds to by trashing the remote copy. Adding a line to -/// `.pdfsignore` must never delete anything from Drive. -/// -/// Ignored baseline rows are deliberately left in the database rather than -/// removed: with the row intact, un-ignoring the path later re-adopts the -/// existing remote file instead of reading as a brand-new local one. -fn classification_order( - local: &HashMap, - remote: &HashMap, - baseline: &HashMap, - rules: &IgnoreRules, -) -> Vec -where - L: HasKind, - R: HasKind, -{ - let mut paths: HashSet = HashSet::new(); - paths.extend(local.keys().cloned()); - paths.extend(remote.keys().cloned()); - paths.extend(baseline.keys().cloned()); - let mut order: Vec = paths.into_iter().collect(); - if !rules.is_empty() { - order.retain(|rel| { - // A baseline-only path has no kind recorded, so it is tested as - // both; see `filter_baseline` for why erring towards "ignored" is - // the safe direction here. - let kind = local - .get(rel) - .map(HasKind::is_dir) - .or_else(|| remote.get(rel).map(HasKind::is_dir)); - match kind { - Some(is_dir) => !rules.is_ignored(rel, is_dir), - None => !rules.is_ignored(rel, false) && !rules.is_ignored(rel, true), - } - }); - } - order.sort_by_key(|p| p.matches('/').count()); - order -} - -/// Lets [`classification_order`] ask either walk's item whether it is a -/// directory without knowing which walk it came from. -trait HasKind { - fn is_dir(&self) -> bool; -} - -impl HasKind for LocalItem { - fn is_dir(&self) -> bool { - self.is_dir - } -} - -// ---- open-for-write detection --------------------------------------------- - -/// Scan `/proc/*/fd` once and return the set of **canonical** paths under -/// `root` that any process currently holds open for writing (`O_WRONLY` or -/// `O_RDWR`). The scan is best-effort: unreadable `/proc/*/fd` directories -/// (other users' processes, or kernel threads) are silently skipped. -/// -/// Cost: one `readdir` of `/proc`, then for every process one `readdir` of its -/// `/proc//fd/`, and one `readlink` + one `open` of `/proc//fdinfo/` -/// per open fd that resolves under `root`. For a typical desktop with a few -/// hundred processes and a handful of files inside the sync root, this -/// completes in low single-digit milliseconds. -fn open_for_write_set(root: &Path) -> HashSet { - let mut result = HashSet::new(); - let Ok(root_canonical) = std::fs::canonicalize(root) else { - return result; - }; - let Ok(proc_entries) = std::fs::read_dir("/proc") else { - return result; - }; - for proc_entry in proc_entries.filter_map(|e| e.ok()) { - let pid_name = proc_entry.file_name(); - // Skip non-numeric entries (kernel threads, /proc/self, etc.) - if !pid_name - .to_str() - .is_some_and(|s| s.bytes().all(|b| b.is_ascii_digit())) - { - continue; - } - let fd_dir = PathBuf::from("/proc").join(&pid_name).join("fd"); - let Ok(fds) = std::fs::read_dir(&fd_dir) else { - continue; - }; - for fd_entry in fds.filter_map(|e| e.ok()) { - // readlink on /proc//fd/ gives the target path. - let link = fd_dir.join(fd_entry.file_name()); - let Ok(target) = std::fs::read_link(&link) else { - continue; - }; - // Quick prefix check before the more expensive fdinfo read. - if !target.starts_with(&root_canonical) { - continue; - } - // Read the fd's flags from /proc//fdinfo/ to determine - // the access mode. The first line is "pos:\t", the second - // is "flags:\t". We only need the low two bits of flags: - // 0 = O_RDONLY, 1 = O_WRONLY, 2 = O_RDWR - let fdinfo_path = PathBuf::from("/proc") - .join(&pid_name) - .join("fdinfo") - .join(fd_entry.file_name()); - if let Ok(contents) = std::fs::read_to_string(&fdinfo_path) - && is_write_mode(&contents) - { - result.insert(target); - } - } - } - result -} - -/// Parse the `flags:` line out of a `/proc//fdinfo/` file and return -/// true if the low two bits indicate write access (`O_WRONLY = 1` or -/// `O_RDWR = 2`). -fn is_write_mode(fdinfo: &str) -> bool { - for line in fdinfo.lines() { - if let Some(rest) = line.strip_prefix("flags:\t") { - if let Ok(flags) = u32::from_str_radix(rest.trim_start_matches('0'), 8) { - let access = flags & 0o3; // O_ACCMODE - return access == 1 || access == 2; // O_WRONLY or O_RDWR - } - // An empty octal string ("0" stripped to "") means flags == 0 == O_RDONLY. - return false; - } - } - false -} - -impl HasKind for RemoteItem { - fn is_dir(&self) -> bool { - self.is_dir - } -} - -/// The baseline minus its ignored paths, for [`guard_local_wipe`]. -/// -/// The guard asks "did every synced path vanish locally?", and an ignored path -/// is absent from the local walk by rule rather than by loss. Left in, a rule -/// covering the whole tree would trip the guard on every pass and wedge that -/// folder's sync for good. -/// -/// A baseline row does not record whether it was a file or a folder, so a path -/// counts as ignored if it matches as either. Erring towards "ignored" only ever -/// shrinks the set the guard checks, which weakens a safety net rather than -/// causing a deletion — the wrong direction to be wrong in is the other one. -fn filter_baseline<'a, B>( - baseline: &'a HashMap, - rules: &IgnoreRules, -) -> HashMap { - baseline - .iter() - .filter(|(rel, _)| !rules.is_ignored(rel, false) && !rules.is_ignored(rel, true)) - .map(|(rel, entry)| (rel.clone(), entry)) - .collect() -} - -fn guard_local_wipe( - baseline: &HashMap, - local: &HashMap, -) -> Result<(), String> { - if baseline.len() >= 2 && baseline.keys().all(|rel| !local.contains_key(rel)) { - return Err(format!( - "every one of the {} synced paths is missing locally; refusing to trash \ - them on Drive. Check that the folder is mounted and readable.", - baseline.len() - )); - } - Ok(()) -} - +mod engine; +mod local_scan; +mod model; +mod planner; +use engine::SYNC_CONCURRENCY; +pub(crate) use engine::{SyncMsg, spawn}; +#[cfg(test)] +use engine::{classify, settle_with}; +#[cfg(test)] +use local_scan::is_write_mode; +use local_scan::open_for_write_set; +use model::{LocalItem, Outcome, PassAbort, Pending, RemoteItem}; +pub(crate) use planner::base_name; +use planner::{ + classification_order, conflict_path, filter_baseline, guard_local_wipe, join_rel, parent_rel, + rel_to_path, remote_sig, unchanged_remote_size, +}; /// What a [`Pending`] op did, so the reconcile can update shared state on the /// engine thread (never inside a task). enum Applied { @@ -1992,40 +1402,6 @@ fn remote_file_sig(node: &Node) -> (i64, i64) { (node.modification_time, size) } -/// The size the baseline recorded for `rel`, if its remote signature's mtime -/// still matches `mtime` — meaning the file has not changed on Drive, so the -/// recorded size stands and there is no need to decrypt it again. -/// -/// `None` means the size has to be read for real: either the mtime moved, or -/// there is no baseline signature to trust (a file the sync has not seen). -fn unchanged_remote_size( - baseline: &HashMap, - rel: &str, - mtime: i64, -) -> Option { - match baseline.get(rel).and_then(remote_sig) { - Some((recorded, size)) if recorded == mtime => Some(size), - _ => None, - } -} - -/// Join a child `name` onto a walk's `prefix`, giving a rel path. -fn join_rel(prefix: &str, name: &str) -> String { - if prefix.is_empty() { - name.to_string() - } else { - format!("{prefix}/{name}") - } -} - -/// The stored remote signature of a baseline row, if it has one. -fn remote_sig(e: &StoredSyncEntry) -> Option<(i64, i64)> { - match (&e.remote_rev, &e.remote_hash) { - (Some(m), Some(s)) => Some((m.parse().ok()?, s.parse().ok()?)), - _ => None, - } -} - /// A file's modification time as epoch seconds (0 if unavailable). fn system_mtime(meta: &std::fs::Metadata) -> i64 { meta.modified() @@ -2034,43 +1410,6 @@ fn system_mtime(meta: &std::fs::Metadata) -> i64 { .map(|d| d.as_secs() as i64) .unwrap_or(0) } - -/// The parent of a `/`-joined relative path (`""` for a top-level entry). -fn parent_rel(rel: &str) -> &str { - match rel.rfind('/') { - Some(i) => &rel[..i], - None => "", - } -} - -/// The final component of a `/`-joined relative path. -pub(crate) fn base_name(rel: &str) -> &str { - match rel.rfind('/') { - Some(i) => &rel[i + 1..], - None => rel, - } -} - -/// Turn a `/`-joined relative path into an OS path (`/` is already the separator -/// on Linux, this keeps the intent explicit). -fn rel_to_path(rel: &str) -> PathBuf { - rel.split('/').collect() -} - -/// The name for a conflict copy of `path`, e.g. `notes (sync-conflict 1700000000).txt`. -fn conflict_path(path: &Path, stamp: i64) -> PathBuf { - let stem = path.file_stem().and_then(|s| s.to_str()).unwrap_or("file"); - let ext = path.extension().and_then(|s| s.to_str()); - let name = match ext { - Some(ext) => format!("{stem} (sync-conflict {stamp}).{ext}"), - None => format!("{stem} (sync-conflict {stamp})"), - }; - match path.parent() { - Some(dir) => dir.join(name), - None => PathBuf::from(name), - } -} - #[cfg(test)] mod tests { use super::*; diff --git a/crates/pdfs-fuse/src/sync/engine.rs b/crates/pdfs-fuse/src/sync/engine.rs new file mode 100644 index 0000000..d1e2ca5 --- /dev/null +++ b/crates/pdfs-fuse/src/sync/engine.rs @@ -0,0 +1,251 @@ +//! Filesystem watcher, polling, and debounce control plane. + +use super::*; + +/// How long the watcher must go quiet before a burst counts as finished. +/// +/// Measured from the *last* event, not the first: applications do not write a +/// file once. An editor saves by writing a temp file and renaming it over the +/// target; an export or a database dump writes continuously for as long as it +/// takes. Settling a fixed interval after the first event walks the tree while +/// the file is still growing, and uploads a torn snapshot as a real revision — +/// corrected on the next pass, but only after spending the encrypt-and-upload +/// on a file that was never in that state. +const DEBOUNCE: Duration = Duration::from_secs(2); +/// Ceiling on the settle wait, so a directory under sustained change still +/// syncs. Without it, copying a large tree in — events arriving forever — +/// would postpone the reconcile for as long as the copy ran. +const MAX_SETTLE: Duration = Duration::from_secs(30); +/// How often to re-walk remotes so changes made on other clients are pulled in. +const POLL_INTERVAL: Duration = Duration::from_secs(120); +/// How many uploads/downloads/folder-creations a reconcile runs at once. The +/// per-item work is a network round-trip, so a folder with thousands of files is +/// unusable done one-at-a-time; this bounds concurrency like the bulk-upload +/// engine does. +pub(super) const SYNC_CONCURRENCY: usize = 8; + +/// A message to the sync engine's control thread. +pub(crate) enum SyncMsg { + /// Reconcile one folder by id (a local change, or a targeted `SyncNow`). + Reconcile(i64), + /// Reconcile every mirror folder (periodic poll, or startup). + ReconcileAll, + /// Re-read the folder set and adjust the filesystem watches (after add/remove). + Rewatch, +} + +/// Start the sync engine: a control thread that owns the filesystem watcher and +/// serialises reconcile passes, plus a periodic poll thread. The engine receives +/// on `rx`; senders live in [`Core::sync_tx`]. +pub(crate) fn spawn(core: Core, rx: Receiver) { + if let Err(e) = std::thread::Builder::new() + .name("pdfs-sync".into()) + .spawn(move || engine_loop(core, rx)) + { + warn!(error = %e, "failed to start sync engine"); + } +} + +/// The engine control loop. Single-threaded so reconcile passes never overlap. +fn engine_loop(core: Core, rx: Receiver) { + // Paths the watcher currently covers, mapped to their folder id, so an event + // path can be resolved back to the folder it belongs to. + let watched: Mutex> = Mutex::new(Vec::new()); + + let tx_events = core.sync_tx.clone(); + let mut watcher = + match notify::recommended_watcher(move |res: notify::Result| { + let Ok(event) = res else { return }; + // Ignore pure metadata/access noise; content changes are what matter. + if matches!(event.kind, notify::EventKind::Access(_)) { + return; + } + let _ = tx_events.send(SyncMsg::ReconcileAll); + }) { + Ok(w) => w, + Err(e) => { + warn!(error = %e, "filesystem watcher unavailable; sync is poll-only"); + // Still run the loop so polling and explicit SyncNow work. + poll_only_loop(&core, rx); + return; + } + }; + + // Periodic remote poll. + { + let tx = core.sync_tx.clone(); + std::thread::Builder::new() + .name("pdfs-sync-poll".into()) + .spawn(move || { + loop { + std::thread::sleep(POLL_INTERVAL); + if tx.send(SyncMsg::ReconcileAll).is_err() { + break; + } + } + }) + .ok(); + } + + rewatch(&core, &mut watcher, &watched); + reconcile_all(&core); + + while let Ok(msg) = rx.recv() { + let mut ids: HashSet = HashSet::new(); + let mut all = false; + let mut do_rewatch = false; + classify(msg, &mut ids, &mut all, &mut do_rewatch); + settle(&rx, &mut ids, &mut all, &mut do_rewatch); + + if do_rewatch { + rewatch(&core, &mut watcher, &watched); + } + if all { + reconcile_all(&core); + } else { + for id in ids { + if let Ok(Some(folder)) = core.db.sync_folder_get(id) { + core.reconcile_folder(&folder); + } + } + } + } +} + +/// Fallback loop when no filesystem watcher could be created: poll + SyncNow only. +fn poll_only_loop(core: &Core, rx: Receiver) { + { + let tx = core.sync_tx.clone(); + std::thread::spawn(move || { + loop { + std::thread::sleep(POLL_INTERVAL); + if tx.send(SyncMsg::ReconcileAll).is_err() { + break; + } + } + }); + } + reconcile_all(core); + while let Ok(msg) = rx.recv() { + let mut ids: HashSet = HashSet::new(); + let mut all = false; + let mut do_rewatch = false; + classify(msg, &mut ids, &mut all, &mut do_rewatch); + settle(&rx, &mut ids, &mut all, &mut do_rewatch); + if all { + reconcile_all(core); + } else { + for id in ids { + if let Ok(Some(folder)) = core.db.sync_folder_get(id) { + core.reconcile_folder(&folder); + } + } + } + } +} + +/// Absorb the rest of an event burst, returning once the watcher has been quiet +/// for `quiet` or `cap` has elapsed since the first event — whichever comes +/// first. The caller has already classified the event that opened the burst. +/// +/// Split out with its timings as parameters so the settling behaviour can be +/// tested in milliseconds rather than in the tens of seconds the real constants +/// describe. +pub(super) fn settle_with( + rx: &Receiver, + ids: &mut HashSet, + all: &mut bool, + do_rewatch: &mut bool, + quiet: Duration, + cap: Duration, +) { + let deadline = Instant::now() + cap; + loop { + let now = Instant::now(); + if now >= deadline { + break; + } + // Never wait past the cap, however quiet the watcher goes. + let wait = quiet.min(deadline - now); + match rx.recv_timeout(wait) { + // Another event: the burst is still going, so the quiet window + // starts again from here. + Ok(m) => classify(m, ids, all, do_rewatch), + // Quiet for a full window: whatever was being written has stopped. + Err(RecvTimeoutError::Timeout) => break, + // Every sender is gone; the caller's loop is about to end anyway. + Err(RecvTimeoutError::Disconnected) => break, + } + } +} + +/// [`settle_with`] at the real timings. +fn settle(rx: &Receiver, ids: &mut HashSet, all: &mut bool, do_rewatch: &mut bool) { + settle_with(rx, ids, all, do_rewatch, DEBOUNCE, MAX_SETTLE); +} + +pub(super) fn classify(msg: SyncMsg, ids: &mut HashSet, all: &mut bool, rewatch: &mut bool) { + match msg { + SyncMsg::Reconcile(id) => { + ids.insert(id); + } + SyncMsg::ReconcileAll => *all = true, + SyncMsg::Rewatch => *rewatch = true, + } +} + +/// Reconcile every mirror folder in turn. +fn reconcile_all(core: &Core) { + let folders = match core.db.sync_folder_list() { + Ok(f) => f, + Err(e) => { + warn!(error = ?e, "sync: cannot list folders"); + return; + } + }; + for folder in folders { + core.reconcile_folder(&folder); + } +} + +/// Bring the filesystem watches in line with the current mirror-folder set: +/// watch newly-added folders, drop removed ones. +fn rewatch( + core: &Core, + watcher: &mut notify::RecommendedWatcher, + watched: &Mutex>, +) { + let want: Vec<(PathBuf, i64)> = match core.db.sync_folder_list() { + Ok(f) => f + .into_iter() + .filter(|f| f.mode == "mirror") + .map(|f| (PathBuf::from(f.local_path), f.id)) + .collect(), + Err(e) => { + warn!(error = ?e, "sync: cannot list folders to watch"); + return; + } + }; + let mut have = watched.lock(); + // Drop watches no longer wanted. + have.retain(|(path, _)| { + if want.iter().any(|(p, _)| p == path) { + true + } else { + let _ = watcher.unwatch(path); + false + } + }); + // Add newly wanted watches. + for (path, id) in &want { + if have.iter().any(|(p, _)| p == path) { + continue; + } + match watcher.watch(path, RecursiveMode::Recursive) { + Ok(()) => have.push((path.clone(), *id)), + Err(e) => warn!(path = %path.display(), error = %e, "sync: watch failed"), + } + } +} + +// ---- reconcile ------------------------------------------------------------ diff --git a/crates/pdfs-fuse/src/sync/local_scan.rs b/crates/pdfs-fuse/src/sync/local_scan.rs new file mode 100644 index 0000000..0658aca --- /dev/null +++ b/crates/pdfs-fuse/src/sync/local_scan.rs @@ -0,0 +1,81 @@ +//! Local open-for-write detection used to defer torn uploads. + +use super::*; + +// ---- open-for-write detection --------------------------------------------- + +/// Scan `/proc/*/fd` once and return the set of **canonical** paths under +/// `root` that any process currently holds open for writing (`O_WRONLY` or +/// `O_RDWR`). The scan is best-effort: unreadable `/proc/*/fd` directories +/// (other users' processes, or kernel threads) are silently skipped. +/// +/// Cost: one `readdir` of `/proc`, then for every process one `readdir` of its +/// `/proc//fd/`, and one `readlink` + one `open` of `/proc//fdinfo/` +/// per open fd that resolves under `root`. For a typical desktop with a few +/// hundred processes and a handful of files inside the sync root, this +/// completes in low single-digit milliseconds. +pub(super) fn open_for_write_set(root: &Path) -> HashSet { + let mut result = HashSet::new(); + let Ok(root_canonical) = std::fs::canonicalize(root) else { + return result; + }; + let Ok(proc_entries) = std::fs::read_dir("/proc") else { + return result; + }; + for proc_entry in proc_entries.filter_map(|e| e.ok()) { + let pid_name = proc_entry.file_name(); + // Skip non-numeric entries (kernel threads, /proc/self, etc.) + if !pid_name + .to_str() + .is_some_and(|s| s.bytes().all(|b| b.is_ascii_digit())) + { + continue; + } + let fd_dir = PathBuf::from("/proc").join(&pid_name).join("fd"); + let Ok(fds) = std::fs::read_dir(&fd_dir) else { + continue; + }; + for fd_entry in fds.filter_map(|e| e.ok()) { + // readlink on /proc//fd/ gives the target path. + let link = fd_dir.join(fd_entry.file_name()); + let Ok(target) = std::fs::read_link(&link) else { + continue; + }; + // Quick prefix check before the more expensive fdinfo read. + if !target.starts_with(&root_canonical) { + continue; + } + // Read the fd's flags from /proc//fdinfo/ to determine + // the access mode. The first line is "pos:\t", the second + // is "flags:\t". We only need the low two bits of flags: + // 0 = O_RDONLY, 1 = O_WRONLY, 2 = O_RDWR + let fdinfo_path = PathBuf::from("/proc") + .join(&pid_name) + .join("fdinfo") + .join(fd_entry.file_name()); + if let Ok(contents) = std::fs::read_to_string(&fdinfo_path) + && is_write_mode(&contents) + { + result.insert(target); + } + } + } + result +} + +/// Parse the `flags:` line out of a `/proc//fdinfo/` file and return +/// true if the low two bits indicate write access (`O_WRONLY = 1` or +/// `O_RDWR = 2`). +pub(super) fn is_write_mode(fdinfo: &str) -> bool { + for line in fdinfo.lines() { + if let Some(rest) = line.strip_prefix("flags:\t") { + if let Ok(flags) = u32::from_str_radix(rest.trim_start_matches('0'), 8) { + let access = flags & 0o3; // O_ACCMODE + return access == 1 || access == 2; // O_WRONLY or O_RDWR + } + // An empty octal string ("0" stripped to "") means flags == 0 == O_RDONLY. + return false; + } + } + false +} diff --git a/crates/pdfs-fuse/src/sync/model.rs b/crates/pdfs-fuse/src/sync/model.rs new file mode 100644 index 0000000..5d8d6bf --- /dev/null +++ b/crates/pdfs-fuse/src/sync/model.rs @@ -0,0 +1,164 @@ +//! Reconciliation snapshots, queued operations, and pass outcomes. + +use super::*; + +pub(super) struct LocalItem { + pub(super) is_dir: bool, + pub(super) mtime: i64, + pub(super) size: i64, + /// True when another process holds this file open for writing (detected via + /// `/proc/*/fd`). An open-for-write file is kept in the map — so it is not + /// misread as a deletion — but treated as unchanged for upload purposes, + /// deferring the upload until the writer closes the file. + pub(super) open_for_write: bool, +} + +/// One item found while walking a remote tree. +pub(super) struct RemoteItem { + pub(super) uid: NodeUid, + pub(super) is_dir: bool, + pub(super) mtime: i64, + pub(super) size: i64, +} + +/// The result of a reconcile pass: what it moved, how many paths were kept as +/// conflict copies, and how many failed to apply (and so still need another +/// pass). The counts drive both the folder's state and its activity summary. +#[derive(Default)] +pub(super) struct Outcome { + pub(super) uploaded: usize, + pub(super) downloaded: usize, + pub(super) created: usize, + pub(super) deleted: usize, + pub(super) conflicts: usize, + pub(super) errors: usize, + /// Files skipped because another process held them open for writing. + pub(super) deferred: usize, +} + +impl Outcome { + /// Fold in one applied op. + pub(super) fn record(&mut self, applied: &Applied) { + match applied { + Applied::Dir(..) => self.created += 1, + Applied::Uploaded => self.uploaded += 1, + Applied::Downloaded => self.downloaded += 1, + Applied::Conflict => self.conflicts += 1, + } + } + + /// Whether the pass moved nothing at all — the common case on a poll of an + /// unchanged folder, which should not add a line to the activity feed. + pub(super) fn is_empty(&self) -> bool { + self.uploaded == 0 + && self.downloaded == 0 + && self.created == 0 + && self.deleted == 0 + && self.conflicts == 0 + && self.errors == 0 + && self.deferred == 0 + } + + /// A human summary of the pass: "3 uploaded, 1 downloaded, 2 failed". + pub(super) fn summary(&self) -> String { + let mut parts = Vec::new(); + for (n, label) in [ + (self.uploaded, "uploaded"), + (self.downloaded, "downloaded"), + (self.created, "folder(s) created"), + (self.deleted, "deleted"), + (self.conflicts, "conflicted"), + (self.deferred, "deferred (open for write)"), + (self.errors, "failed"), + ] { + if n > 0 { + parts.push(format!("{n} {label}")); + } + } + parts.join(", ") + } +} + +/// A network operation queued during classification and run concurrently in a +/// per-depth batch. Parent uids are resolved up front (the parent folder is one +/// depth shallower and already created), so tasks share nothing mutable. +pub(super) enum Pending { + /// Create a new remote folder under `parent`. + CreateDir { rel: String, parent: NodeUid }, + /// Upload a brand-new local file into `parent`. + UploadNew { rel: String, parent: NodeUid }, + /// Upload a changed local file as a new revision of `uid`. + UploadRevision { rel: String, uid: NodeUid }, + /// Download remote `uid` to the local path, stamping `mtime`. `size` is the + /// remote's reported size, used as the transfer's expected total. + Download { + rel: String, + uid: NodeUid, + mtime: i64, + size: i64, + }, + /// Both sides changed: set the local copy aside, then download remote `uid`. + Conflict { + rel: String, + uid: NodeUid, + mtime: i64, + size: i64, + }, + /// Both sides changed during a [`push pass`](Core::push_pass): upload the local + /// copy into `parent` under a conflict name, leaving the remote file as it is. + PushConflict { rel: String, parent: NodeUid }, +} + +impl Pending { + /// The path this op acts on, relative to the folder root. + pub(super) fn rel(&self) -> &str { + match self { + Pending::CreateDir { rel, .. } + | Pending::UploadNew { rel, .. } + | Pending::UploadRevision { rel, .. } + | Pending::Download { rel, .. } + | Pending::Conflict { rel, .. } + | Pending::PushConflict { rel, .. } => rel, + } + } + + /// How this op reads in the activity feed. + pub(super) fn kind(&self) -> ActivityKind { + match self { + Pending::CreateDir { .. } => ActivityKind::CreateFolder, + Pending::UploadNew { .. } + | Pending::UploadRevision { .. } + | Pending::PushConflict { .. } => ActivityKind::Upload, + Pending::Download { .. } | Pending::Conflict { .. } => ActivityKind::Download, + } + } + + /// The activity line's detail, which distinguishes ops that share a kind. + pub(super) fn detail(&self) -> &'static str { + match self { + Pending::CreateDir { .. } => "on Drive", + Pending::UploadNew { .. } => "new file", + Pending::UploadRevision { .. } => "new version", + Pending::Download { .. } => "from Drive", + Pending::Conflict { .. } => "local changes kept as a conflict copy", + Pending::PushConflict { .. } => "local changes uploaded as a conflict copy", + } + } +} + +/// Why a full reconcile pass stopped without a result. +pub(super) enum PassAbort { + /// An on-demand switch was queued while the pass was running. Everything left + /// to do is work on a local copy about to be evicted, so the pass gives up its + /// remaining work to a [`push pass`](Core::push_pass) instead. Not a failure — + /// nothing is left half-applied, because each step is applied whole. + Interrupted, + /// The pass could not establish its diff (a walk or the baseline load failed). + Failed(String), +} + +impl From for PassAbort { + fn from(e: String) -> Self { + PassAbort::Failed(e) + } +} diff --git a/crates/pdfs-fuse/src/sync/planner.rs b/crates/pdfs-fuse/src/sync/planner.rs new file mode 100644 index 0000000..8efa54c --- /dev/null +++ b/crates/pdfs-fuse/src/sync/planner.rs @@ -0,0 +1,176 @@ +//! Pure reconciliation ordering, safety guards, signatures, and path rules. + +use super::*; + +/// The paths a reconcile pass will classify: the union of the local, remote, +/// and baseline states, minus anything the ignore rules exclude, shallowest +/// first so a parent folder is created before its children are placed in it. +/// +/// The ignore filter belongs *here*, on the union, rather than on the local walk +/// alone — filtering the walk alone would be actively destructive. A file synced +/// before it became ignored is absent from `local` but present in `remote` and +/// `baseline`, which is precisely the "local deleted, remote untouched" shape +/// `do_reconcile` responds to by trashing the remote copy. Adding a line to +/// `.pdfsignore` must never delete anything from Drive. +/// +/// Ignored baseline rows are deliberately left in the database rather than +/// removed: with the row intact, un-ignoring the path later re-adopts the +/// existing remote file instead of reading as a brand-new local one. +pub(super) fn classification_order( + local: &HashMap, + remote: &HashMap, + baseline: &HashMap, + rules: &IgnoreRules, +) -> Vec +where + L: HasKind, + R: HasKind, +{ + let mut paths: HashSet = HashSet::new(); + paths.extend(local.keys().cloned()); + paths.extend(remote.keys().cloned()); + paths.extend(baseline.keys().cloned()); + let mut order: Vec = paths.into_iter().collect(); + if !rules.is_empty() { + order.retain(|rel| { + // A baseline-only path has no kind recorded, so it is tested as + // both; see `filter_baseline` for why erring towards "ignored" is + // the safe direction here. + let kind = local + .get(rel) + .map(HasKind::is_dir) + .or_else(|| remote.get(rel).map(HasKind::is_dir)); + match kind { + Some(is_dir) => !rules.is_ignored(rel, is_dir), + None => !rules.is_ignored(rel, false) && !rules.is_ignored(rel, true), + } + }); + } + order.sort_by_key(|p| p.matches('/').count()); + order +} + +/// Lets [`classification_order`] ask either walk's item whether it is a +/// directory without knowing which walk it came from. +pub(super) trait HasKind { + fn is_dir(&self) -> bool; +} + +impl HasKind for LocalItem { + fn is_dir(&self) -> bool { + self.is_dir + } +} + +impl HasKind for RemoteItem { + fn is_dir(&self) -> bool { + self.is_dir + } +} + +/// The baseline minus its ignored paths, for [`guard_local_wipe`]. +/// +/// The guard asks "did every synced path vanish locally?", and an ignored path +/// is absent from the local walk by rule rather than by loss. Left in, a rule +/// covering the whole tree would trip the guard on every pass and wedge that +/// folder's sync for good. +/// +/// A baseline row does not record whether it was a file or a folder, so a path +/// counts as ignored if it matches as either. Erring towards "ignored" only ever +/// shrinks the set the guard checks, which weakens a safety net rather than +/// causing a deletion — the wrong direction to be wrong in is the other one. +pub(super) fn filter_baseline<'a, B>( + baseline: &'a HashMap, + rules: &IgnoreRules, +) -> HashMap { + baseline + .iter() + .filter(|(rel, _)| !rules.is_ignored(rel, false) && !rules.is_ignored(rel, true)) + .map(|(rel, entry)| (rel.clone(), entry)) + .collect() +} + +/// Refuse to run a pass whose local side has vanished in its entirety. +/// +/// When every baseline path is absent locally, the likely cause is an unavailable +/// mount or unreadable folder rather than a deliberate whole-tree deletion. +/// One surviving path is enough to clear the guard. +pub(super) fn guard_local_wipe( + baseline: &HashMap, + local: &HashMap, +) -> Result<(), String> { + if baseline.len() >= 2 && baseline.keys().all(|rel| !local.contains_key(rel)) { + return Err(format!( + "every one of the {} synced paths is missing locally; refusing to trash \ + them on Drive. Check that the folder is mounted and readable.", + baseline.len() + )); + } + Ok(()) +} + +/// The size the baseline recorded for `rel`, if its remote signature's mtime +/// still matches `mtime`. +pub(super) fn unchanged_remote_size( + baseline: &HashMap, + rel: &str, + mtime: i64, +) -> Option { + match baseline.get(rel).and_then(remote_sig) { + Some((recorded, size)) if recorded == mtime => Some(size), + _ => None, + } +} + +/// Join a child `name` onto a walk's `prefix`, giving a rel path. +pub(super) fn join_rel(prefix: &str, name: &str) -> String { + if prefix.is_empty() { + name.to_string() + } else { + format!("{prefix}/{name}") + } +} + +/// The stored remote signature of a baseline row, if it has one. +pub(super) fn remote_sig(e: &StoredSyncEntry) -> Option<(i64, i64)> { + match (&e.remote_rev, &e.remote_hash) { + (Some(m), Some(s)) => Some((m.parse().ok()?, s.parse().ok()?)), + _ => None, + } +} + +/// The parent of a `/`-joined relative path (`""` for a top-level entry). +pub(super) fn parent_rel(rel: &str) -> &str { + match rel.rfind('/') { + Some(i) => &rel[..i], + None => "", + } +} + +/// The final component of a `/`-joined relative path. +pub(crate) fn base_name(rel: &str) -> &str { + match rel.rfind('/') { + Some(i) => &rel[i + 1..], + None => rel, + } +} + +/// Turn a `/`-joined relative path into an OS path (`/` is already the separator +/// on Linux, this keeps the intent explicit). +pub(super) fn rel_to_path(rel: &str) -> PathBuf { + rel.split('/').collect() +} + +/// The name for a conflict copy of `path`, e.g. `notes (sync-conflict 1700000000).txt`. +pub(super) fn conflict_path(path: &Path, stamp: i64) -> PathBuf { + let stem = path.file_stem().and_then(|s| s.to_str()).unwrap_or("file"); + let ext = path.extension().and_then(|s| s.to_str()); + let name = match ext { + Some(ext) => format!("{stem} (sync-conflict {stamp}).{ext}"), + None => format!("{stem} (sync-conflict {stamp})"), + }; + match path.parent() { + Some(dir) => dir.join(name), + None => PathBuf::from(name), + } +} diff --git a/crates/pdfs-fuse/src/upload.rs b/crates/pdfs-fuse/src/upload.rs new file mode 100644 index 0000000..8211732 --- /dev/null +++ b/crates/pdfs-fuse/src/upload.rs @@ -0,0 +1,298 @@ +//! Bulk local-to-Drive upload workflow. + +use super::*; + +/// How many files the bulk uploader ships at once. Overlaps the per-file network +/// round-trips without letting an unbounded number of block buffers pile up. +const UPLOAD_CONCURRENCY: usize = 4; + +/// One file queued for bulk upload, resolved during the directory walk so the +/// concurrent phase carries everything it needs (no shared state, no `block_on`). +struct UploadTask { + /// Inode of the (already-created) remote parent folder, for interning the + /// uploaded node afterwards. + parent_ino: u64, + parent_uid: NodeUid, + name: String, + /// Local filesystem path to stream from. + path: PathBuf, + size: u64, +} + +/// Tally of a completed [`Core::upload_paths`] batch, for the daemon log. +#[derive(Default)] +pub(super) struct UploadStats { + pub(super) uploaded: usize, + pub(super) failed: usize, + /// Total plaintext bytes of the files that uploaded successfully. + pub(super) bytes: u64, + /// Folders created (or reused) to mirror the local tree. + pub(super) folders: usize, +} + +/// Upload every [`UploadTask`] with at most `limit` in flight at once, each +/// streamed straight from disk and ticking its own transfer-registry guard. +/// Returns, per file, either `(parent_ino, new_uid)` for the caller to intern or +/// `(name, error)` to log — one failure never sinks the batch. +/// +/// `job` counts files finished (either way: a failure is still one file the batch +/// no longer waits on), so a front-end can show "12 of 40" over the per-file bars. +async fn run_uploads( + core: Core, + tasks: Vec, + limit: usize, + job: Arc, +) -> Vec> { + let sem = Arc::new(tokio::sync::Semaphore::new(limit)); + let mut set = tokio::task::JoinSet::new(); + for t in tasks { + let core = core.clone(); + let sem = sem.clone(); + set.spawn(async move { + let _permit = sem.acquire_owned().await.expect("semaphore not closed"); + let file = match std::fs::File::open(&t.path) { + Ok(f) => f, + Err(e) => return Err((t.name, format!("open {}: {e}", t.path.display()))), + }; + let mtime = std::fs::metadata(&t.path) + .and_then(|m| m.modified()) + .ok() + .and_then(|mt| mt.duration_since(std::time::UNIX_EPOCH).ok()) + .map(|d| d.as_secs() as i64); + let guard = core + .transfers + .begin(&t.name, "", TransferDirection::Upload, t.size); + let reader = OwnedCountingReader::new(file, guard); + match core + .client + .upload_file_from( + &t.parent_uid, + &t.name, + media_type_for(&t.name), + reader, + t.size as i64, + Vec::new(), + mtime, + false, + ) + .await + { + Ok(uid) => Ok((t.parent_ino, uid, t.size)), + Err(e) => Err((t.name, format!("upload: {e}"))), + } + }); + } + let mut out = Vec::new(); + while let Some(joined) = set.join_next().await { + job.step(); + match joined { + Ok(result) => out.push(result), + Err(e) => warn!(error = %e, "upload task panicked"), + } + } + out +} + +impl Core { + /// Bulk-upload local files and directory trees under `sources` into the + /// mountpoint-relative `parent_rel` folder. Directories are recreated (or + /// merged into an existing same-named folder) and walked; the resulting flat + /// set of files is uploaded with bounded concurrency, each ticking the + /// transfer registry so a front-end sees live progress. Runs on a background + /// thread — a large tree far outlasts the control socket's read timeout — so + /// it reports only a summary for the log. Individual failures are counted and + /// logged rather than aborting the whole batch. + pub(super) fn upload_paths( + &self, + parent_rel: &Path, + sources: &[PathBuf], + ) -> CoreResult { + let (pino, parent_uid) = self.resolve(parent_rel)?; + self.ensure_children(pino) + .map_err(|e| self.errno_error(e, "enumerate"))?; + + // Phase 1 (sequential): build the remote folder skeleton and collect the + // flat list of files to upload. Folders must exist before their children, + // so this can't be parallelised. On a deep tree this is a folder-creation + // round-trip per directory before a single byte moves — long enough that + // it needs a job of its own, or the daemon looks idle for minutes. + let mut tasks = Vec::new(); + let mut folders = 0usize; + { + let job = self.transfers.begin_job("Preparing upload"); + for src in sources { + if let Err(e) = + self.collect_uploads(pino, &parent_uid, src, &mut tasks, &mut folders, &job) + { + warn!(source = %src.display(), error = %e, "skipping source"); + } + } + } + + // Phase 2 (concurrent): upload the files, up to UPLOAD_CONCURRENCY at once. + // Each file reports its own bytes; this job is the batch's "N of M files", + // which is the number a user actually waits on. + let job = Arc::new(self.transfers.begin_job(match tasks.len() { + 1 => "Uploading 1 file".to_string(), + n => format!("Uploading {n} files"), + })); + job.set_total(tasks.len() as u64); + let outcomes = self.rt.block_on(run_uploads( + self.clone(), + tasks, + UPLOAD_CONCURRENCY, + job.clone(), + )); + drop(job); + + // Phase 3 (sequential): intern each uploaded node so it shows up in the + // listing without a re-enumeration. fetch_node uses `block_on`, so it must + // run here rather than inside the async batch — and it is a round-trip per + // file, so it too gets a job rather than a silent tail. + let mut stats = UploadStats { + folders, + ..UploadStats::default() + }; + let job = self.transfers.begin_job("Finishing upload"); + job.set_total(outcomes.len() as u64); + for outcome in outcomes { + job.step(); + match outcome { + Ok((parent_ino, uid, size)) => { + stats.uploaded += 1; + stats.bytes += size; + match self.fetch_node(&uid) { + Ok(node) => { + let mut st = self.state.lock(); + let ino = st.intern(parent_ino, node); + if let Some(kids) = st.children.get_mut(&parent_ino) + && !kids.contains(&ino) + { + kids.push(ino); + } + } + // Uploaded fine but the metadata refresh failed; it will + // appear on the next directory enumeration regardless. + Err(e) => warn!(%uid, error = ?e, "uploaded node metadata refresh failed"), + } + } + Err((name, msg)) => { + stats.failed += 1; + warn!(name, error = %msg, "file upload failed"); + } + } + } + info!( + uploaded = stats.uploaded, + failed = stats.failed, + "bulk upload finished" + ); + Ok(stats) + } + + /// Resolve a remote child folder named `name` under `pino`, creating it if it + /// doesn't exist, and return its `(inode, uid)`. Reusing an existing same-named + /// folder makes re-uploading a directory merge into it rather than fail on a + /// duplicate name. + fn ensure_remote_folder( + &self, + pino: u64, + parent_uid: &NodeUid, + name: &str, + ) -> CoreResult<(u64, NodeUid)> { + if name.is_empty() || name.contains('/') { + return Err(CoreError::invalid(format!("invalid folder name: {name:?}"))); + } + self.ensure_children(pino) + .map_err(|e| self.errno_error(e, "enumerate"))?; + { + let st = self.state.lock(); + if let Some(kids) = st.children.get(&pino) { + for &ino in kids { + if let Some(e) = st.entries.get(&ino) + && e.node.is_folder() + && e.node.name == name + { + return Ok((ino, e.uid.clone())); + } + } + } + } + let new_uid = self + .rt + .block_on( + self.client + .create_folder(parent_uid, name, Some(now_secs())), + ) + .map_err(|e| CoreError::from_api(&e, &format!("create folder {name}")))?; + let node = self + .fetch_node(&new_uid) + .map_err(|e| self.errno_error(e, "fetch node"))?; + let mut st = self.state.lock(); + let ino = st.intern(pino, node); + if let Some(kids) = st.children.get_mut(&pino) + && !kids.contains(&ino) + { + kids.push(ino); + } + Ok((ino, new_uid)) + } + + /// Walk one local source path, appending its files to `tasks`. A file becomes + /// one task; a directory is recreated remotely and recursed into (children + /// sorted for a stable order). Symlinks and other special files are skipped. + /// + /// `job` narrates the walk with the folder currently being mirrored. It stays + /// indeterminate: nothing knows the size of the tree until the walk has ended. + fn collect_uploads( + &self, + pino: u64, + parent_uid: &NodeUid, + src: &Path, + tasks: &mut Vec, + folders: &mut usize, + job: &JobGuard, + ) -> CoreResult<()> { + let meta = std::fs::symlink_metadata(src) + .map_err(|e| CoreError::internal(format!("stat {}: {e}", src.display())))?; + let name = src + .file_name() + .and_then(|n| n.to_str()) + .ok_or_else(|| CoreError::invalid(format!("unusable name: {}", src.display())))? + .to_string(); + + if meta.is_dir() { + job.detail(&name); + let (child_ino, child_uid) = self.ensure_remote_folder(pino, parent_uid, &name)?; + *folders += 1; + let mut entries: Vec = std::fs::read_dir(src) + .map_err(|e| CoreError::internal(format!("read dir {}: {e}", src.display())))? + .filter_map(|e| e.ok()) + .map(|e| e.path()) + .collect(); + entries.sort(); + for entry in entries { + if let Err(e) = + self.collect_uploads(child_ino, &child_uid, &entry, tasks, folders, job) + { + warn!(source = %entry.display(), error = %e, "skipping entry"); + } + } + // Deeper folders have retitled the job by now; put this one back so the + // line tracks the walk's position rather than its deepest leaf. + job.detail(&name); + } else if meta.is_file() { + if name.contains('/') { + return Err(CoreError::invalid(format!("invalid file name: {name:?}"))); + } + tasks.push(UploadTask { + parent_ino: pino, + parent_uid: parent_uid.clone(), + name, + path: src.to_path_buf(), + size: meta.len(), + }); + } + Ok(()) + } +} diff --git a/docs/BUGS.md b/docs/BUGS.md index dce2599..5ced3ef 100644 --- a/docs/BUGS.md +++ b/docs/BUGS.md @@ -19,7 +19,7 @@ Conventions: **Verified:** the exact repro below on the new daemon — `mv` returns 0, the file is present in the destination listing and reads back its content. Previously it was in neither directory. -**Where:** `crates/pdfs-fuse/src/lib.rs`, `Filesystem::rename` +**Where:** `crates/pdfs-fuse/src/filesystem.rs`, `Filesystem::rename` `rename` ended with: @@ -323,7 +323,7 @@ Each part of the fix is separately visible in the traces: ### Secondary finding (still open): `lookup` is O(n) per name -**Where:** `crates/pdfs-fuse/src/lib.rs`, `serve_lookup` +**Where:** `crates/pdfs-fuse/src/filesystem.rs`, `serve_lookup` `serve_lookup` linear-scans the parent's children comparing names, and with no `readdirplus` that is one `lookup` per child — O(n²) name comparisons under the @@ -358,7 +358,7 @@ judgement still holds now the real cause is known: `docs/ARCHITECTURE.md` §8. Not from a report — from checking a claim before asserting it in a doc. I had written "their default 0700 permissions protect them", went to verify, and found we set no modes at all. -**Where:** `crates/pdfs-fuse/src/lib.rs` (`UnixListener::bind(control_socket)`), +**Where:** `crates/pdfs-fuse/src/mount.rs` (`UnixListener::bind(control_socket)`), `crates/pdfs-core/src/config.rs` (`create_dir_all` on state/cache dirs) `grep -rn "set_permissions\|from_mode" crates/pdfs-core/src crates/pdfs-fuse/src` @@ -817,7 +817,7 @@ rsync: [receiver] rename ".../.Buunshin - heimwee (Original Mix).wav.b0akm3" -> ".../Buunshin - heimwee (Original Mix).wav": Input/output error (5) ``` -**Where:** `crates/pdfs-fuse/src/lib.rs`, `Filesystem::rename` (~4380) +**Where:** `crates/pdfs-fuse/src/filesystem.rs`, `Filesystem::rename` Daemon log, one per failed file: @@ -1187,7 +1187,7 @@ retracted. **Status:** Fixed (verified 2026-07-22) **Found:** 2026-07-21, multi-agent filesystem safety audit (`audit_bugs.md` CRIT-01) -**Where:** `crates/pdfs-fuse/src/lib.rs`, `ProtonFs::unlink`, `ProtonFs::rmdir` +**Where:** `crates/pdfs-fuse/src/filesystem.rs`, `ProtonFs::unlink`, `ProtonFs::rmdir` **Cause:** `unlink` on directories and `rmdir` on non-empty directories bypassed POSIX checks and called `trash_child` directly, silently deleting remote subtrees. @@ -1243,7 +1243,7 @@ switching back to `mirror` clears the baseline and restores the local copy. **Status:** Fixed (verified 2026-07-22) **Found:** 2026-07-21, multi-agent FUSE audit (`audit_bugs.md` HIGH-01) -**Where:** `crates/pdfs-fuse/src/lib.rs`, `ProtonFs::fallocate` +**Where:** `crates/pdfs-fuse/src/filesystem.rs`, `ProtonFs::fallocate` **Cause:** `FALLOC_FL_PUNCH_HOLE` zeroed scratch blocks but removed their `WriteHandle::written` authored status. Revision assembly therefore classified @@ -1286,7 +1286,7 @@ silently undoing the hole at commit. **Status:** Fixed (verified 2026-07-21) **Found:** 2026-07-21, multi-agent FUSE audit (`audit_bugs.md` HIGH-04) -**Where:** `crates/pdfs-fuse/src/lib.rs`, `ProtonFs::rename`, `crates/pdfs-fuse/src/state.rs:L237` +**Where:** `crates/pdfs-fuse/src/filesystem.rs`, `ProtonFs::rename`, `crates/pdfs-fuse/src/state.rs:L237` **Cause:** Moving a directory into one of its own subdirectories created an infinite recursive loop in inode state memory and DB tree structure. @@ -1462,7 +1462,7 @@ added later, but it must retain the row and blob. **Status:** Fixed in code (unverified on a live mount, 2026-07-22) **Found:** 2026-07-22, deep FUSE/POSIX audit -**Where:** `crates/pdfs-fuse/src/lib.rs`, `ProtonFs::trash_child` +**Where:** `crates/pdfs-fuse/src/filesystem.rs`, `ProtonFs::trash_child` **Cause:** `trash_child` called `forget_or_unlink` before the online `trash_nodes` request. If Drive rejected or failed the request, FUSE returned @@ -1558,7 +1558,7 @@ retain the baseline and increment `Outcome.errors`. **Status:** Open **Found:** 2026-07-22, FUSE/POSIX audit -**Where:** `crates/pdfs-fuse/src/lib.rs`, `ProtonFs::serve_readdir` +**Where:** `crates/pdfs-fuse/src/filesystem.rs`, `ProtonFs::serve_readdir` **Cause:** `readdir` rebuilds a live vector for every call and uses its array index as the continuation cookie. A namespace mutation between pages can shift @@ -1574,7 +1574,7 @@ small reply pages and mutation between calls. **Status:** Open **Found:** 2026-07-22, daemon lifecycle audit -**Where:** `crates/pdfs-fuse/src/lib.rs`, mount lifecycle; drain/index/sync loops +**Where:** `crates/pdfs-fuse/src/mount.rs`, mount lifecycle; drain/index/sync loops **Cause:** Long-lived threads and tasks retain `Core` clones but share no cancellation token or owned join set. Unmount removes the session/socket without @@ -1585,4 +1585,3 @@ continue DB/client activity after teardown. handles, and ordered shutdown. Repeated mount/unmount tests should return thread counts to baseline and observe no post-unmount DB work. -