From 242a28bb8f28be83442dfe02314eb57323a7a6dd Mon Sep 17 00:00:00 2001 From: Nils Pukropp Date: Wed, 22 Jul 2026 02:02:26 +0200 Subject: [PATCH] refactor from monolithic structure --- crates/pdfs-fuse/src/background.rs | 268 ++++ crates/pdfs-fuse/src/filesystem.rs | 1421 +++++++++++++++++++ crates/pdfs-fuse/src/lib.rs | 2088 +--------------------------- crates/pdfs-fuse/src/mount.rs | 392 ++++++ 4 files changed, 2094 insertions(+), 2075 deletions(-) create mode 100644 crates/pdfs-fuse/src/background.rs create mode 100644 crates/pdfs-fuse/src/filesystem.rs create mode 100644 crates/pdfs-fuse/src/mount.rs diff --git a/crates/pdfs-fuse/src/background.rs b/crates/pdfs-fuse/src/background.rs new file mode 100644 index 0000000..7bb12eb --- /dev/null +++ b/crates/pdfs-fuse/src/background.rs @@ -0,0 +1,268 @@ +//! Remote-event invalidation and local-index background workers. + +use super::*; + +/// Apply one remote event to the local cache and notify the kernel so it drops +/// any stale cached metadata/data for the affected inodes. +/// +/// The cache is authoritative-by-absence: dropping a directory's `children` +/// entry forces the next `lookup`/`readdir` to re-enumerate from the remote, so +/// most events only need to invalidate listings rather than re-fetch eagerly. +fn apply_event( + state: &Mutex, + content: &ContentCache, + pending: &Mutex>, + notifier: &Notifier, + event: &DriveEvent, +) { + match event { + DriveEvent::NodeUpdated { + node_uid, + parent_node_uid, + is_trashed, + .. + } => { + let mut st = state.lock(); + if *is_trashed { + // Trashing makes a node vanish from its parent listing. + let child = st.by_uid.get(node_uid).copied(); + if let Some((parent, name)) = st.forget(node_uid) { + content.evict(node_uid); + match child { + Some(child) => { + let _ = + notifier.delete(INodeNo(parent), INodeNo(child), OsStr::new(&name)); + } + None => { + let _ = notifier.inval_entry(INodeNo(parent), OsStr::new(&name)); + } + } + } + } else if pending.lock().contains_key(node_uid) { + // A node we owe an upload for is *ahead* of the remote, not + // behind it: this event is almost always the echo of our own + // empty-file create, and re-fetching would replace the size and + // mtime of the write we just accepted with the stale revision's + // — making a file that was copied in seconds ago read as empty + // until its upload lands (offline.md Phase 3). + debug!(uid = %node_uid, "ignoring remote event for a node with a queued write"); + } else if let Some(&ino) = st.by_uid.get(node_uid) { + // Known node changed: drop its cached attrs/data (and listing if + // it is a directory) so the next access re-fetches. Its content + // blob may now be stale, so evict it too. + st.invalidate_listing(ino); + content.evict(node_uid); + let _ = notifier.inval_inode(INodeNo(ino), 0, 0); + } + // A create (or move-in) shows up as a change to the parent listing; + // drop it so the new child is picked up on the next readdir. + if let Some(parent_uid) = parent_node_uid + && let Some(&parent) = st.by_uid.get(parent_uid) + { + st.invalidate_listing(parent); + let _ = notifier.inval_inode(INodeNo(parent), 0, 0); + } + } + DriveEvent::NodeDeleted { node_uid, .. } => { + let mut st = state.lock(); + // Capture the inode before `forget` clears the uid mapping. + let child = st.by_uid.get(node_uid).copied(); + content.evict(node_uid); + if let Some((parent, name)) = st.forget(node_uid) { + match child { + Some(child) => { + let _ = notifier.delete(INodeNo(parent), INodeNo(child), OsStr::new(&name)); + } + None => { + let _ = notifier.inval_entry(INodeNo(parent), OsStr::new(&name)); + } + } + } + } + // Continuity or scope was lost: our cached listings may be arbitrarily + // stale, so drop every listing and tell the kernel to forget all + // metadata. Inodes stay stable; dirs simply re-enumerate on next access. + DriveEvent::ContinuityLost { .. } | DriveEvent::ScopeAccessLost { .. } => { + warn!("event continuity lost; dropping all cached listings, resyncing lazily"); + let mut st = state.lock(); + let dirs: Vec = st.children.keys().copied().collect(); + for &ino in &dirs { + st.invalidate_listing(ino); + let _ = notifier.inval_inode(INodeNo(ino), 0, 0); + } + } + // No substantive local change; the cursor advance is handled by the + // caller persisting the event id. + DriveEvent::CursorAdvanced { .. } | DriveEvent::SharedWithMeUpdated { .. } => {} + } +} + +/// Poll the remote event cursor forever, applying each batch to the shared +/// state. Resumes from the cursor persisted in the DB so changes made while +/// unmounted are applied; only a first-ever mount seeds from the server head. +/// The cursor is persisted after every batch. Runs as a Tokio task; returns +/// only on fatal error. +pub(super) async fn run_event_sync( + client: ProtonDriveClient, + scope: DriveEventScopeId, + state: Arc>, + content: Arc, + db: Arc, + pending: Arc>>, + notifier: Notifier, +) { + let mut cursor: Option = match db.get_event_cursor() { + // Resume: pick up exactly where the last run left off. + Ok(Some(saved)) => Some(DriveEventId::from(saved)), + // First mount: a `None` cursor yields a single `CursorAdvanced` at the + // server head; persist it so the next restart resumes instead of + // reseeding (which would skip everything that changed offline). + // Seeding needs the network, and this task also runs on mounts that + // started offline (offline.md Phase 1) — so retry rather than giving up, + // which used to disable live sync for the life of the daemon. + Ok(None) => { + let mut delay = ONLINE_PROBE_MIN; + loop { + match client.enumerate_events(&scope, None).await { + Ok(events) => { + let head = events.last().map(|e| e.id().clone()); + if let Some(c) = &head + && let Err(e) = db.set_event_cursor(c.as_str()) + { + warn!(error = %e, "persist seed cursor failed"); + } + break head; + } + Err(e) => { + warn!(error = %e, ?delay, "seed event cursor failed; retrying"); + tokio::time::sleep(delay).await; + delay = (delay * 2).min(ONLINE_PROBE_MAX); + } + } + } + } + Err(e) => { + error!(error = %e, "read persisted cursor failed; live sync disabled"); + return; + } + }; + info!(?cursor, "event sync started"); + + loop { + tokio::time::sleep(POLL_INTERVAL).await; + let events = match client.enumerate_events(&scope, cursor.as_ref()).await { + Ok(events) => events, + Err(e) => { + warn!(error = %e, "event poll failed; retrying after interval"); + continue; + } + }; + if events.is_empty() { + continue; + } + debug!(count = events.len(), "applying remote events"); + for event in &events { + // Converge the SDK's own caches (folder keys, entity cache) on the + // server before applying the event to our tree. Without this, a node + // re-keyed/moved by another client keeps a stale key in the SDK for + // the life of the daemon (SDK plan #9). `apply_event` only touches + // our FUSE state, so nothing else does this. + if let Err(e) = client.invalidate_caches_for_event(event).await { + warn!(error = %e, "sdk cache invalidation for event failed"); + } + apply_event(&state, &content, &pending, ¬ifier, event); + } + cursor = events.last().map(|e| e.id().clone()); + if let Some(c) = &cursor + && let Err(e) = db.set_event_cursor(c.as_str()) + { + warn!(error = %e, "persist event cursor failed"); + } + } +} + +/// Keep the local-file index fresh for the launcher prompt's "This computer" +/// results. Rebuilds the index whenever it is older than [`LOCAL_INDEX_TTL`], +/// then sleeps; runs on its own thread for the life of the daemon. +/// +/// The walk is the one part of the daemon that touches the wider filesystem, so +/// it is deliberately kept off every hot path: it never runs on a FUSE or +/// control-socket thread, and it excludes the mountpoint (walking it would fault +/// every remote node in through FUSE, defeating on-demand hydration). +pub(super) fn run_local_index( + db: Arc, + indexing: Arc, + transfers: Arc, + mountpoint: PathBuf, +) { + loop { + let age = db.local_indexed_at().ok().flatten(); + let stale = + age.is_none_or(|at| now_secs().saturating_sub(at) >= LOCAL_INDEX_TTL.as_secs() as i64); + if stale { + scan_local_once(&db, &indexing, &transfers, &mountpoint); + } + std::thread::sleep(LOCAL_INDEX_CHECK); + } +} + +/// Walk `$HOME` once and replace the local-file index with what it finds. +/// Batches stream straight into SQLite, so peak memory is one batch — not the +/// whole home directory. +/// +/// Reports itself as a job: the first scan after a fresh install walks the whole +/// home directory, and `indexing` alone only tells the launcher prompt to say +/// "still indexing" — nothing else showed that the daemon was busy. +fn scan_local_once( + db: &Db, + indexing: &AtomicBool, + transfers: &Arc, + mountpoint: &Path, +) { + let dirs = match AppDirs::new() { + Ok(d) => d, + Err(e) => { + warn!(error = %e, "local index: cannot resolve app dirs"); + return; + } + }; + let Some(home) = dirs.home_dir() else { + warn!("local index: cannot resolve home directory"); + return; + }; + let generation = match db.local_begin_scan() { + Ok(g) => g, + Err(e) => { + warn!(error = %e, "local index: cannot open scan generation"); + return; + } + }; + + let excludes = localindex::default_excludes(mountpoint, &dirs.state_dir(), &dirs.cache_dir()); + indexing.store(true, Ordering::Relaxed); + let started = Instant::now(); + + // The walk has no idea how many files it will find, so the job counts what it + // has seen and stays indeterminate. + let job = transfers.begin_job("Indexing this computer"); + job.detail("Scanning your files"); + let walked = localindex::scan(&[home], &excludes, |batch| { + if let Err(e) = db.local_upsert_batch(generation, &batch) { + warn!(error = %e, "local index: batch write failed"); + } + }); + + // Prune what this scan did not see and rebuild the FTS index over the rest, + // even if some batches failed — a partial index still beats none. + job.detail("Building the search index"); + match db.local_finish_scan(generation, now_secs()) { + Ok(indexed) => info!( + walked, + indexed, + elapsed_ms = started.elapsed().as_millis() as u64, + "local index rebuilt" + ), + Err(e) => warn!(error = %e, "local index: finish failed"), + } + indexing.store(false, Ordering::Relaxed); +} diff --git a/crates/pdfs-fuse/src/filesystem.rs b/crates/pdfs-fuse/src/filesystem.rs new file mode 100644 index 0000000..a8fa056 --- /dev/null +++ b/crates/pdfs-fuse/src/filesystem.rs @@ -0,0 +1,1421 @@ +//! Kernel-facing FUSE callback adapter. + +use super::*; + +impl Filesystem for ProtonFs { + fn lookup(&self, _req: &Request, parent: INodeNo, name: &OsStr, reply: ReplyEntry) { + let parent = parent.0; + let name = name.to_string_lossy().into_owned(); + // A folder that has not been listed yet is enumerated from the remote, + // so serve it from a worker rather than stalling the dispatch loop. A + // listed folder — the common case — is a map hit, and answering it + // inline costs less than the handoff would. + if self.core.children_cached(parent) { + self.serve_lookup(parent, &name, reply, false); + return; + } + let fs = self.clone(); + self.core.workers.run(Lane::Meta, move || { + fs.serve_lookup(parent, &name, reply, true) + }); + } + + fn forget(&self, _req: &Request, ino: INodeNo, nlookup: u64) { + let mut st = self.core.state.lock(); + st.forget_lookup(ino.0, nlookup); + } + + fn getattr(&self, _req: &Request, ino: INodeNo, _fh: Option, reply: ReplyAttr) { + let (attr, provisional_in) = { + let st = self.core.state.lock(); + match st.entries.get(&ino.0) { + Some(e) => { + // A file still carrying the cheap enumeration's placeholder + // size (B12). + let provisional = matches!( + &e.node.kind, + NodeKind::File { + claimed_size: None, + .. + } + ) + .then(|| (e.parent, e.uid.clone())); + (self.attr(ino.0, &e.node), provisional) + } + None => { + reply.error(Errno::ENOENT); + return; + } + } + }; + // The common case: the size is real, answer from the lock we just held. + let Some((parent, uid)) = provisional_in else { + reply.attr(&TTL, &attr); + return; + }; + // A provisional size is the *ciphertext* size, which is larger than the + // file. Publishing it makes every reader that trusts `st_size` — rsync, + // mmap, sendfile, a sized read loop — run off the end of the file and + // fail. So resolve it before answering rather than after (bugs.md B14). + // + // The cost is one batched round trip for the whole folder, not one per + // file: `ls -l` is one `getattr` per entry and they collapse onto a + // single upgrade. This is why B12's split is still worth having — the + // plain listing never comes here at all. + // + // Off the dispatch loop, because it goes to the network: blocking there + // would stall every other operation on the mount (the B5 lesson). + let fs = self.clone(); + self.core.workers.run(Lane::Meta, move || { + fs.core.upgrade_sizes_for_parent(ino.0, &uid, parent); + // Re-read: the upgrade writes through `state`, and on timeout or + // failure this is simply the provisional attr we already had. + let attr = { + let st = fs.core.state.lock(); + match st.entries.get(&ino.0) { + Some(e) => fs.attr(ino.0, &e.node), + // Forgotten while we waited. + None => attr, + } + }; + reply.attr(&TTL, &attr); + }); + } + + fn readdir( + &self, + _req: &Request, + ino: INodeNo, + _fh: FileHandle, + offset: u64, + reply: ReplyDirectory, + ) { + let ino = ino.0; + // Same split as `lookup`: only the cold, remote-enumerating path pays + // for a worker. + if self.core.children_cached(ino) { + self.serve_readdir(ino, offset, reply); + return; + } + let fs = self.clone(); + self.core + .workers + .run(Lane::Meta, move || fs.serve_readdir(ino, offset, reply)); + } + + fn open(&self, _req: &Request, ino: INodeNo, flags: OpenFlags, reply: ReplyOpen) { + let (uid, base_mtime, base_size) = { + let mut st = self.core.state.lock(); + match st.entries.get_mut(&ino.0) { + Some(e) if e.node.is_file() => { + e.open_count = e.open_count.saturating_add(1); + (e.uid.clone(), e.node.modification_time, node_size(&e.node)) + } + Some(_) => { + reply.error(Errno::EISDIR); + return; + } + None => { + reply.error(Errno::ENOENT); + return; + } + } + }; + debug!(ino = ino.0, ?flags, base_size, base_mtime, "open"); + // Read-only opens stay stateless (fh 0). A write open allocates a + // disk-backed handle; bytes are authored into a scratch file and the + // untouched remainder is pulled from the base lazily (on read / commit). + if flags.acc_mode() == OpenAccMode::O_RDONLY { + reply.opened(FileHandle(0), FopenFlags::empty()); + return; + } + let (file, path) = match self.core.cache.create_scratch() { + Ok(x) => x, + Err(e) => { + if let Some(entry) = self.core.state.lock().entries.get_mut(&ino.0) { + entry.open_count = entry.open_count.saturating_sub(1); + } + error!(%uid, error = %e, "create scratch file failed"); + reply.error(Errno::EIO); + return; + } + }; + let mut st = self.core.state.lock(); + let fh = st.next_fh; + st.next_fh += 1; + let aw = st.active_writes.entry(ino.0).or_insert_with(|| { + WriteHandle { + ino: ino.0, + uid, + file: Arc::new(file), + path, + written: Intervals::default(), + // Starts at the current size; reads in [0, base_size) come from + // the base until overwritten. + len: base_size, + base_size, + base_mtime, + dirty: false, + open_count: 0, + } + }); + aw.open_count += 1; + st.handles.insert(fh, ino.0); + reply.opened(FileHandle(fh), FopenFlags::empty()); + } + + fn read( + &self, + _req: &Request, + ino: INodeNo, + _fh: FileHandle, + offset: u64, + size: u32, + _flags: OpenFlags, + _lock_owner: Option, + reply: ReplyData, + ) { + // A file open for writing is served from its handle so reads see the + // in-flight (possibly unsaved) content: authored bytes come from the + // scratch file, untouched bytes from the base. + // + // This path stays on the dispatch loop. `write` runs there too, so + // keeping its reads there as well preserves the ordering the write path + // has always had between a write and a read of the same handle; a mostly + // local read is not worth reasoning about that for. + let handle = { + let st = self.core.state.lock(); + st.active_writes.get(&ino.0).map(|h| { + ( + h.file.clone(), + h.len, + h.uid.clone(), + h.base_mtime, + h.base_size, + h.written.clone(), + ) + }) + }; + if let Some((file, len, uid, base_mtime, base_size, written)) = handle { + match self.core.serve_open_read( + &file, + len, + &uid, + base_mtime, + base_size, + &written, + offset, + size as u64, + ) { + Ok(bytes) => reply.data(&bytes), + Err(e) => reply.error(e), + } + return; + } + let (uid, mtime, fsize, is_video) = { + let st = self.core.state.lock(); + match st.entries.get(&ino.0) { + Some(e) if e.node.is_file() => { + let media_type = match &e.node.kind { + NodeKind::File { media_type, .. } => Some(media_type.as_str()), + NodeKind::Folder => None, + }; + let is_video = + PhotoKind::classify(Some(&e.node.name), media_type) == PhotoKind::Video; + let fsize = self + .core + .pending + .lock() + .get(&e.uid) + .map(|p| p.meta.len) + .unwrap_or_else(|| node_size(&e.node)); + (e.uid.clone(), e.node.modification_time, fsize, is_video) + } + Some(_) => { + reply.error(Errno::EISDIR); + return; + } + None => { + reply.error(Errno::ENOENT); + return; + } + } + }; + // A large unpinned video streams without polluting the block cache (see + // [`STREAM_BYPASS_MIN`]); everything else caches its blocks as usual. + let cache_blocks = + !(is_video && fsize >= STREAM_BYPASS_MIN && !self.core.cache.is_pinned(&uid)); + // Off the dispatch loop: this is the one handler that routinely goes to + // the network (block fetch + decrypt, hundreds of ms on a cold file), + // and until it returns the kernel's next request for this mount is not + // even read off the FUSE device. It only reads `state`, so moving it + // races with nothing. FUSE does not require replies in request order. + let core = self.core.clone(); + self.core.workers.run(Lane::Transfer, move || { + match core.read_range(&uid, mtime, fsize, offset, size as u64, cache_blocks) { + Ok(bytes) => reply.data(&bytes), + Err(e) => reply.error(e), + } + }); + } + + fn create( + &self, + _req: &Request, + parent: INodeNo, + name: &OsStr, + _mode: u32, + _umask: u32, + _flags: i32, + reply: ReplyCreate, + ) { + let parent = parent.0; + if let Err(e) = self.core.ensure_children(parent) { + reply.error(e); + return; + } + let name = name.to_string_lossy().into_owned(); + let parent_uid = { + let st = self.core.state.lock(); + match st.entries.get(&parent) { + Some(e) => e.uid.clone(), + None => { + reply.error(Errno::ENOENT); + return; + } + } + }; + // Offline the server cannot mint a uid, so invent one and queue the + // create. The file is real to the caller either way; only its identity is + // provisional until the drain (offline.md Phase 3b). + // + // A parent that is itself still queued forces the same path even when we + // are online: the API has no folder to put this in yet. + let node = if self.core.online.load(Ordering::Relaxed) && !is_local_uid(&parent_uid) { + // Create an empty file on the remote so it has a real uid immediately; + // written bytes are buffered and sealed as a new revision on close. + let new_uid = match self.core.rt.block_on(self.core.client.upload_file( + &parent_uid, + &name, + media_type_for(&name), + b"", + )) { + Ok(u) => u, + Err(e) => { + error!(%parent_uid, name, error = %e, "create file failed"); + reply.error(Errno::EIO); + return; + } + }; + match self.core.fetch_node(&new_uid) { + Ok(n) => n, + Err(e) => { + reply.error(e); + return; + } + } + } else { + match self.core.queue_local_node(&parent_uid, &name, false) { + Ok(n) => n, + Err(e) => { + reply.error(e); + return; + } + } + }; + let new_uid = node.uid.clone(); + // The base this handle writes over is the node as it actually exists — + // the empty file the server just minted — so its modification time comes + // from the node, never from the local clock. `queue_revision` turns this + // into `StagedWrite::based_on` and the drain compares that against the + // remote: a `now_secs()` here differs from the server's stamp by however + // long the create round-trip took, so the first write to a brand-new file + // conflicts with its own create whenever the second happens to tick in + // between. + let base_mtime = node.modification_time; + let (file, path) = match self.core.cache.create_scratch() { + Ok(x) => x, + Err(e) => { + error!(%new_uid, error = %e, "create scratch file failed"); + reply.error(Errno::EIO); + return; + } + }; + let mut st = self.core.state.lock(); + let ino = st.intern(parent, node); + if let Some(entry) = st.entries.get_mut(&ino) { + entry.open_count = entry.open_count.saturating_add(1); + } + if let Some(kids) = st.children.get_mut(&parent) + && !kids.contains(&ino) + { + kids.push(ino); + } + let fh = st.next_fh; + st.next_fh += 1; + let aw = st.active_writes.entry(ino).or_insert_with(|| { + // A brand-new file: empty base, everything written is authored. + WriteHandle { + ino, + uid: new_uid, + file: Arc::new(file), + path, + written: Intervals::default(), + len: 0, + base_size: 0, + base_mtime, + dirty: false, + open_count: 0, + } + }); + aw.open_count += 1; + st.handles.insert(fh, ino); + let attr = self.attr(ino, &st.entries.get(&ino).unwrap().node); + reply.created( + &TTL, + &attr, + Generation(0), + FileHandle(fh), + FopenFlags::empty(), + ); + } + + fn write( + &self, + _req: &Request, + ino: INodeNo, + fh: FileHandle, + offset: u64, + data: &[u8], + _write_flags: WriteFlags, + _flags: OpenFlags, + _lock_owner: Option, + reply: ReplyWrite, + ) { + let fh = fh.0; + // Stage the bytes straight into the scratch file (no base download): only + // the untouched remainder is pulled from the remote, and only at commit. + let file = { + let st = self.core.state.lock(); + match st.handles.get(&fh).and_then(|&i| st.active_writes.get(&i)) { + Some(aw) => aw.file.clone(), + None => { + reply.error(Errno::EBADF); + return; + } + } + }; + if let Err(e) = file.write_all_at(data, offset) { + error!(ino = ino.0, fh, error = %e, "scratch write failed"); + reply.error(Errno::EIO); + return; + } + let new_len = { + let mut st = self.core.state.lock(); + let Some(aw) = st + .handles + .get(&fh) + .copied() + .and_then(|i| st.active_writes.get_mut(&i)) + else { + reply.error(Errno::EBADF); + return; + }; + let end = offset + data.len() as u64; + aw.written.add(offset, end); + aw.len = aw.len.max(end); + aw.dirty = true; + let len = aw.len; + st.set_size(ino.0, len); + len + }; + debug!( + ino = ino.0, + fh, + offset, + len = data.len(), + new_len, + "staged write" + ); + reply.written(data.len() as u32); + } + + #[allow(clippy::too_many_arguments)] + fn setattr( + &self, + _req: &Request, + ino: INodeNo, + _mode: Option, + _uid: Option, + _gid: Option, + size: Option, + _atime: Option, + _mtime: Option, + _ctime: Option, + fh: Option, + _crtime: Option, + _chgtime: Option, + _bkuptime: Option, + _flags: Option, + reply: ReplyAttr, + ) { + // Only resizes change remote state; everything else (mode/owner/times) is + // accepted and ignored so tools that chmod/utimes after writing succeed. + if let Some(size) = size { + let fh = fh.map(|f| f.0).filter(|&f| f != 0); + let handled = { + let mut st = self.core.state.lock(); + match st.active_writes.get_mut(&ino.0) { + Some(aw) => { + if size < aw.len { + // Shrink: drop authored ranges past the new end. + aw.written.clip(size); + } else if size > aw.len { + // Grow: the new tail is defined as zeros, so claim + // it as authored rather than base content. + aw.written.add(aw.len, size); + } + let _ = aw.file.set_len(size); + aw.len = size; + aw.dirty = true; + true + } + None => false, + } + }; + debug!(ino = ino.0, size, fh = ?fh, handled, "setattr resize"); + if !handled { + // Path-based truncate with no open write handle — a shell's + // `> file`. This is the second write path into the API, and + // queueing it is what lets a redirect work offline at all: it + // never reaches `release`, so without this it failed before any + // byte was written (offline.md Phase 2/3b). + if let Err(e) = self.core.queue_truncate(ino.0, size) { + reply.error(e); + return; + } + } else { + self.core.state.lock().set_size(ino.0, size); + } + } + let st = self.core.state.lock(); + match st.entries.get(&ino.0) { + Some(e) => { + let attr = self.attr(ino.0, &e.node); + reply.attr(&TTL, &attr); + } + None => reply.error(Errno::ENOENT), + } + } + + fn fallocate( + &self, + _req: &Request, + ino: INodeNo, + _fh: FileHandle, + offset: u64, + length: u64, + mode: i32, + reply: ReplyEmpty, + ) { + let new_end = offset.saturating_add(length); + let keep_size = (mode & 1) != 0; // FALLOC_FL_KEEP_SIZE + + let is_punch_hole = (mode & 0x02) != 0; // FALLOC_FL_PUNCH_HOLE + + let res = { + let mut st = self.core.state.lock(); + match st + .handles + .get(&_fh.0) + .copied() + .and_then(|i| st.active_writes.get_mut(&i)) + { + Some(aw) => { + use std::os::unix::io::AsRawFd; + let fd = aw.file.as_raw_fd(); + let ret = unsafe { libc::fallocate(fd, mode, offset as i64, length as i64) }; + if ret == 0 { + if is_punch_hole { + // A punched range reads as zero. Mark it authored so + // queue_revision does not refill it from the remote + // baseline and silently undo the hole at commit. + aw.written.add(offset, new_end.min(aw.len)); + aw.dirty = true; + } else if !keep_size && new_end > aw.len { + aw.written.add(aw.len, new_end); + aw.len = new_end; + aw.dirty = true; + st.set_size(ino.0, new_end); + } + Ok(()) + } else { + Err(Errno::EIO) + } + } + None => { + if st.entries.contains_key(&ino.0) { + Err(Errno::EBADF) + } else { + Err(Errno::ENOENT) + } + } + } + }; + + match res { + Ok(_) => reply.ok(), + Err(e) => reply.error(e), + } + } + + /// `close(2)` calls flush before release. The upload is queued in `release` + /// and performed in the background (offline.md Phase 3), so there is nothing + /// to push here — the written bytes are already in the scratch file. + fn flush( + &self, + _req: &Request, + _ino: INodeNo, + _fh: FileHandle, + _lock_owner: LockOwner, + reply: ReplyEmpty, + ) { + reply.ok(); + } + + /// Durability here means the scratch file, not the remote: a caller asking + /// for fsync wants its bytes to survive a crash, and blocking it on an upload + /// (which offline would never finish) buys nothing the queue does not already + /// guarantee. + fn fsync( + &self, + _req: &Request, + _ino: INodeNo, + fh: FileHandle, + _datasync: bool, + reply: ReplyEmpty, + ) { + // Everything the durability sidecar needs, copied out so no lock is held + // across the sync. + let handle = { + let st = self.core.state.lock(); + let ino = st.handles.get(&fh.0).copied(); + ino.and_then(|i| st.active_writes.get(&i)).map(|aw| { + ( + aw.file.clone(), + aw.path.clone(), + aw.uid.clone(), + aw.dirty, + aw.written.clone(), + aw.len, + aw.base_size, + aw.base_mtime, + ) + }) + }; + let Some((f, path, uid, dirty, written, len, base_size, base_mtime)) = handle else { + reply.error(Errno::EBADF); + return; + }; + if let Err(e) = f.sync_all() { + error!(fh = fh.0, error = %e, "fsync of scratch file failed"); + reply.error(Errno::EIO); + return; + } + // A clean handle has nothing on it that the remote does not already + // hold, so there is nothing for a crash to lose. + if !dirty { + reply.ok(); + return; + } + // The bytes are on stable storage, but only the scratch file knows that, + // and the scratch directory is cleared at open. Record what the blob is + // so a restart can hand it to the upload queue instead of deleting it. + // + // Deliberately *not* the full release path: staging the blob here would + // move the file out from under a handle the application still has open, + // and queueing an upload per `fsync` would push a revision for every + // barrier in a write loop. This only makes the existing file findable. + let authored: Vec<(u64, u64)> = written + .segments(0, len) + .into_iter() + .filter(|&(_, _, authored)| authored) + .map(|(s, e, _)| (s, e)) + .collect(); + let meta = StagedWrite { + uid: uid.to_string(), + len, + base_size, + base_mtime, + complete: authored == [(0, len)], + authored, + based_on: self.core.remote_baseline(&uid, base_mtime, base_size), + }; + // A failed sidecar means the write is not durable, and `fsync` promising + // otherwise is the defect this exists to fix — so it is an error, not a + // warning. + if let Err(e) = self.core.cache.mark_scratch_durable(&path, &meta) { + error!(%uid, error = %e, "recording fsync durability failed"); + reply.error(Errno::EIO); + return; + } + reply.ok(); + } + + fn release( + &self, + _req: &Request, + _ino: INodeNo, + fh: FileHandle, + _flags: OpenFlags, + _lock_owner: Option, + _flush: bool, + reply: ReplyEmpty, + ) { + let (handle, unlinked_uid) = { + let mut st = self.core.state.lock(); + let unlinked_uid = if let Some(entry) = st.entries.get_mut(&_ino.0) { + entry.open_count = entry.open_count.saturating_sub(1); + if entry.open_count == 0 && entry.unlinked { + let uid = entry.uid.clone(); + st.forget(&uid); + Some(uid) + } else { + None + } + } else { + None + }; + let ino = st.handles.remove(&fh.0); + let h = ino.and_then(|i| { + let aw = st.active_writes.get_mut(&i)?; + aw.open_count = aw.open_count.saturating_sub(1); + if aw.open_count == 0 { + st.active_writes.remove(&i) + } else { + None + } + }); + (h, unlinked_uid) + }; + let was_unlinked = unlinked_uid.is_some(); + if let Some(uid) = unlinked_uid { + self.core.discard_queued_ops(&uid); + self.core.cache.evict(&uid); + self.core.evict_reader(&uid); + } + // Hand the bytes to the queue rather than uploading them here: the + // scratch file is the only copy of what was just written, and blocking + // the caller on the network is what made a copy into the mount run at + // upload speed (and fail outright offline). + match (handle, was_unlinked) { + (Some(h), true) => { + // POSIX keeps an unlinked file usable until the last close, but + // closing it must not resurrect those bytes as a new revision. + if let Err(e) = std::fs::remove_file(&h.path) + && e.kind() != std::io::ErrorKind::NotFound + { + warn!(path = %h.path.display(), error = %e, "discarding unlinked scratch file failed"); + } + reply.ok(); + } + (Some(h), false) => match self.core.queue_revision(&h) { + Ok(()) => reply.ok(), + Err(e) => reply.error(e), + }, + (None, _) => reply.ok(), + } + } + + fn unlink(&self, _req: &Request, parent: INodeNo, name: &OsStr, reply: ReplyEmpty) { + let name_str = name.to_string_lossy(); + if let Ok((ino, _)) = self.core.lookup_child(parent.0, &name_str) { + let st = self.core.state.lock(); + if let Some(entry) = st.entries.get(&ino) + && entry.node.is_folder() + { + reply.error(Errno::EISDIR); + return; + } + } + self.trash_child(parent.0, &name_str, reply); + } + + fn rmdir(&self, _req: &Request, parent: INodeNo, name: &OsStr, reply: ReplyEmpty) { + let name_str = name.to_string_lossy(); + if let Ok((ino, _)) = self.core.lookup_child(parent.0, &name_str) { + { + let st = self.core.state.lock(); + if st + .entries + .get(&ino) + .is_some_and(|entry| !entry.node.is_folder()) + { + reply.error(Errno::ENOTDIR); + return; + } + } + // A missing cached listing says "unknown", not "empty". Proton's + // trash call is recursive, so enumerate before applying rmdir's + // POSIX emptiness rule. + if let Err(e) = self.core.ensure_children(ino) { + reply.error(e); + return; + } + if self.core.state.lock().has_children(ino) { + reply.error(Errno::ENOTEMPTY); + return; + } + } + self.trash_child(parent.0, &name_str, reply); + } + + fn mkdir( + &self, + _req: &Request, + parent: INodeNo, + name: &OsStr, + _mode: u32, + _umask: u32, + reply: ReplyEntry, + ) { + let parent = parent.0; + if let Err(e) = self.core.ensure_children(parent) { + reply.error(e); + return; + } + let name = name.to_string_lossy().into_owned(); + let parent_uid = { + let st = self.core.state.lock(); + match st.entries.get(&parent) { + Some(e) => e.uid.clone(), + None => { + reply.error(Errno::ENOENT); + return; + } + } + }; + let now = SystemTime::now() + .duration_since(UNIX_EPOCH) + .map(|d| d.as_secs() as i64) + .ok(); + // As in `create`: offline — or under a parent that is itself still + // queued — the folder becomes a placeholder that the drain turns into a + // real one (offline.md Phase 3b). + let node = if self.core.online.load(Ordering::Relaxed) && !is_local_uid(&parent_uid) { + let new_uid = + match self + .core + .rt + .block_on(self.core.client.create_folder(&parent_uid, &name, now)) + { + Ok(u) => u, + Err(e) => { + error!(%parent_uid, name, error = %e, "create folder failed"); + reply.error(Errno::EIO); + return; + } + }; + match self.core.fetch_node(&new_uid) { + Ok(n) => n, + Err(e) => { + reply.error(e); + return; + } + } + } else { + match self.core.queue_local_node(&parent_uid, &name, true) { + Ok(n) => n, + Err(e) => { + reply.error(e); + return; + } + } + }; + let mut st = self.core.state.lock(); + let local = is_local_uid(&node.uid); + let uid = node.uid.clone(); + let ino = st.intern(parent, node); + if let Some(kids) = st.children.get_mut(&parent) + && !kids.contains(&ino) + { + kids.push(ino); + } + // A folder that exists only here has nothing to enumerate, and asking the + // API to enumerate a `local~` uid would 404. Record it as fully listed and + // empty, which it is, so reads stay offline-clean across a restart too. + if local { + st.children.insert(ino, Vec::new()); + if let Err(e) = self.core.db.set_listed(&uid, true) { + warn!(%uid, error = %e, "db set_listed(true) failed for local folder"); + } + } + let attr = self.attr(ino, &st.entries.get(&ino).unwrap().node); + reply.entry(&TTL, &attr, Generation(0)); + } + + fn rename( + &self, + _req: &Request, + parent: INodeNo, + name: &OsStr, + newparent: INodeNo, + newname: &OsStr, + flags: RenameFlags, + reply: ReplyEmpty, + ) { + let parent = parent.0; + let newparent = newparent.0; + let name = name.to_string_lossy().into_owned(); + let newname = newname.to_string_lossy().into_owned(); + let (ino, uid) = match self.core.lookup_child(parent, &name) { + Ok(x) => x, + Err(e) => { + reply.error(e); + return; + } + }; + // Ancestor cycle check: moving a directory into itself or one of its descendants is forbidden by POSIX (EINVAL). + { + let st = self.core.state.lock(); + if let Some(entry) = st.entries.get(&ino) + && entry.node.is_folder() + && st.is_ancestor_of(ino, newparent) + { + reply.error(Errno::EINVAL); + return; + } + } + // The destination has to be listed either way: the queued path pushes + // the node into that listing, and the online one drops it to force a + // re-enumeration. + if newparent != parent + && let Err(e) = self.core.ensure_children(newparent) + { + reply.error(e); + return; + } + let new_parent_uid = { + let st = self.core.state.lock(); + match st.entries.get(&newparent) { + Some(e) => e.uid.clone(), + None => { + reply.error(Errno::ENOENT); + return; + } + } + }; + // `rename(2)` replaces an existing destination; Proton has no replacing + // rename, so the victim has to be removed first and the operation stops + // being atomic. Without this the 422 surfaced as a blanket EIO and every + // write-to-temp-then-rename tool — rsync, atomic editor saves — failed + // at the very end of its transfer (bugs.md B13). + // + // `RENAME_EXCHANGE` has no Proton primitive and cannot be emulated + // without a window in which one of the two names does not exist. + if flags.contains(RenameFlags::RENAME_EXCHANGE) { + reply.error(Errno::EINVAL); + return; + } + let victim = match self.core.lookup_child(newparent, &newname) { + // Renaming a node onto its own name is a no-op, never a self-replace. + Ok((_, vuid)) if vuid == uid => None, + Ok(x) => Some(x), + // Nothing to replace: the overwhelmingly common case. + Err(e) if e.code() == libc::ENOENT => None, + Err(e) => { + reply.error(e); + return; + } + }; + if let Some((victim_ino, victim_uid)) = &victim { + if flags.contains(RenameFlags::RENAME_NOREPLACE) { + reply.error(Errno::EEXIST); + return; + } + // POSIX requires both ends to agree on being a directory, and the + // check has to happen before anything is trashed: getting it wrong + // turns a refusal into the destruction of the destination. + let (src_dir, dst_dir) = { + let st = self.core.state.lock(); + let is_dir = |i: &u64| { + st.entries + .get(i) + .map(|e| matches!(e.node.kind, NodeKind::Folder)) + }; + match (is_dir(&ino), is_dir(victim_ino)) { + (Some(a), Some(b)) => (a, b), + _ => { + reply.error(Errno::ENOENT); + return; + } + } + }; + // Replacing a directory means trashing it, and Proton trashes a + // folder with its whole subtree — so the destination's emptiness has + // to be known before the decision is made. + let dst_empty = if dst_dir { + if let Err(e) = self.core.ensure_children(*victim_ino) { + reply.error(e); + return; + } + self.core + .state + .lock() + .children + .get(victim_ino) + .is_none_or(|kids| kids.is_empty()) + } else { + true + }; + if let Err(e) = check_replaceable(src_dir, dst_dir, dst_empty) { + reply.error(e); + return; + } + if let Err(e) = self.core.remove_replaced(victim_uid, &newname) { + reply.error(e); + return; + } + } + // A node whose own creation is still queued has no server-side identity + // to rename: the queued op *is* the node, so rewriting its target is the + // whole rename. Nothing reaches the API, which is why this works offline + // (offline.md Phase 3b). + if is_local_uid(&uid) { + match self.core.db.rewrite_op_target( + &uid.to_string(), + &new_parent_uid.to_string(), + &newname, + ) { + Ok(true) => { + self.core.state.lock().rename_in_place( + ino, + newparent, + &new_parent_uid, + &newname, + ); + debug!(%uid, newname, "renamed a node whose create is still queued"); + reply.ok(); + } + // The create drained underneath us, so the node has a real uid + // now and this handle's is stale. A retry resolves it. + Ok(false) => { + warn!(%uid, name, newname, "queued create vanished under a rename"); + self.core.restore_replaced(victim.as_ref(), &newname); + reply.error(Errno::EBUSY); + } + Err(e) => { + error!(%uid, error = %e, "rewriting a queued create's target failed"); + self.core.restore_replaced(victim.as_ref(), &newname); + reply.error(Errno::EIO); + } + } + return; + } + // Offline, or into a folder that does not exist remotely yet: queue the + // rename rather than 404 or fail. Online moves into a real parent still + // take the synchronous path, so a genuine API refusal (permissions, a + // name clash) surfaces to the caller instead of becoming a queued op + // that can only ever conflict. + if !self.core.online.load(Ordering::Relaxed) || is_local_uid(&new_parent_uid) { + match self + .core + .queue_rename(ino, &uid, newparent, &new_parent_uid, &newname) + { + Ok(()) => reply.ok(), + Err(e) => { + self.core.restore_replaced(victim.as_ref(), &newname); + reply.error(e); + } + } + return; + } + // Move first if the parent changed, then rename if the name changed. + // A failure past this point has already trashed any node the rename was + // replacing, so put it back rather than leaving the caller with neither + // the source moved nor the destination intact. + if newparent != parent + && let Err(e) = self + .core + .rt + .block_on(self.core.client.move_node(&uid, &new_parent_uid)) + { + error!(%uid, error = %e, "move failed"); + self.core.restore_replaced(victim.as_ref(), &newname); + reply.error(Errno::EIO); + return; + } + if newname != name + && let Err(e) = self + .core + .rt + .block_on(self.core.client.rename_node(&uid, &newname, None)) + { + error!(%uid, error = %e, "rename failed"); + self.core.restore_replaced(victim.as_ref(), &newname); + reply.error(Errno::EIO); + return; + } + self.core + .state + .lock() + .relocate(ino, parent, newparent, &new_parent_uid, &newname); + reply.ok(); + } + + /// Expose a file's server-side thumbnail/preview as an extended attribute, so + /// a previewing client can fetch it without downloading the whole file. The + /// bytes are fetched on demand and cached; absence yields `ENODATA`. + fn getxattr(&self, _req: &Request, ino: INodeNo, name: &OsStr, size: u32, reply: ReplyXattr) { + // Always off the dispatch loop: a miss goes to the wire, and a lister + // that stats a directory issues one of these per file per advertised + // name — inline, that serialized the whole mount behind ~186 ms of + // network per call (B5). + let fs = self.clone(); + let name = name.to_os_string(); + self.core.workers.run(Lane::Meta, move || { + fs.serve_getxattr(ino, &name, size, reply) + }); + } + + fn listxattr(&self, _req: &Request, ino: INodeNo, size: u32, reply: ReplyXattr) { + self.serve_listxattr(ino, size, reply); + } + + fn ioctl( + &self, + _req: &Request, + _ino: INodeNo, + _fh: FileHandle, + _flags: IoctlFlags, + _cmd: u32, + _in_data: &[u8], + _out_size: u32, + reply: ReplyIoctl, + ) { + // Filesystems are not character devices or terminals; returning ENOTTY + // for unhandled ioctls is standard Linux POSIX behavior and prevents + // fuser warning logs when applications probe file descriptors. + reply.error(Errno::ENOTTY); + } +} + +/// The Proton Drive VFS. FUSE callbacks are synchronous, so the Tokio handle +/// bridges each one to the async SDK via `tokio::runtime::Handle::block_on`; the fuser +/// session thread is not a runtime worker, so blocking on it is sound. +/// Cloneable so a handler can move a copy onto a `Workers` thread and answer +/// from there; every field is a handle or a plain id. +#[derive(Clone)] +pub struct ProtonFs { + core: Core, + uid: u32, + gid: u32, +} + +impl ProtonFs { + /// Build the filesystem rooted at `root` (the user's My Files folder). + pub(super) fn new(core: Core, root: Node) -> Self { + { + let mut st = core.state.lock(); + if let Err(e) = st.db.upsert_node(&root) { + warn!(uid = %root.uid, error = %e, "db upsert root failed"); + } + st.by_uid.insert(root.uid.clone(), ROOT_INO); + st.entries.insert( + ROOT_INO, + Entry { + uid: root.uid.clone(), + parent: ROOT_INO, + node: root, + lookup_count: 1, + open_count: 0, + unlinked: false, + }, + ); + } + // SAFETY: geteuid/getegid are infallible and have no preconditions. + let (uid, gid) = unsafe { (libc::geteuid(), libc::getegid()) }; + Self { core, uid, gid } + } + + /// The body of [`Filesystem::lookup`], on whichever thread ends up serving it. + /// `off_loop` says whether the caller is already on a worker, and so whether + /// this may block. `lookup` answers a cached parent on the dispatch loop, + /// where it may not. + fn serve_lookup(&self, parent: u64, name: &str, reply: ReplyEntry, off_loop: bool) { + if let Err(e) = self.core.ensure_children(parent) { + reply.error(e); + return; + } + let hit = { + let st = self.core.state.lock(); + st.children.get(&parent).and_then(|kids| { + kids.iter().copied().find_map(|ino| { + st.entries + .get(&ino) + .filter(|e| e.node.name == name) + .map(|e| { + // A `lookup` reply carries attrs with the same TTL a + // `getattr` would, so `ls -l` — which is one `lookup` + // per entry and no `getattr` at all — takes its sizes + // from here. Resolving only in `getattr` left the whole + // listing provisional (bugs.md B14). + let provisional = matches!( + &e.node.kind, + NodeKind::File { + claimed_size: None, + .. + } + ) + .then(|| (e.parent, e.uid.clone())); + (ino, self.attr(ino, &e.node), provisional) + }) + }) + }) + }; + let Some((ino, attr, provisional)) = hit else { + reply.error(Errno::ENOENT); + return; + }; + let Some((grandparent, uid)) = provisional else { + reply.entry(&TTL, &attr, Generation(0)); + return; + }; + // Resolving goes to the network. On the dispatch loop that is not + // allowed (the B5 lesson), so hand off — but only from the loop: doing + // it while already on a worker would queue a job onto the lane this + // thread occupies. + if !off_loop { + let fs = self.clone(); + let name = name.to_string(); + self.core.workers.run(Lane::Meta, move || { + fs.serve_lookup(parent, &name, reply, true) + }); + return; + } + self.core.upgrade_sizes_for_parent(ino, &uid, grandparent); + // Re-read: on timeout or failure this is the provisional attr again, + // which is no worse than what this path used to always send. + let attr = { + let st = self.core.state.lock(); + st.entries + .get(&ino) + .map_or(attr, |e| self.attr(ino, &e.node)) + }; + reply.entry(&TTL, &attr, Generation(0)); + } + + /// The body of [`Filesystem::readdir`], on whichever thread ends up serving it. + fn serve_readdir(&self, ino: u64, offset: u64, mut reply: ReplyDirectory) { + if let Err(e) = self.core.ensure_children(ino) { + reply.error(e); + return; + } + let st = self.core.state.lock(); + let parent = st.entries.get(&ino).map_or(ROOT_INO, |e| e.parent); + + // "." and ".." occupy offsets 0 and 1; real children follow. + let mut listing: Vec<(u64, FileType, String)> = vec![ + (ino, FileType::Directory, ".".to_string()), + (parent, FileType::Directory, "..".to_string()), + ]; + if let Some(kids) = st.children.get(&ino) { + for &kid in kids { + if let Some(e) = st.entries.get(&kid) { + let ft = if e.node.is_folder() { + FileType::Directory + } else { + FileType::RegularFile + }; + listing.push((kid, ft, e.node.name.clone())); + } + } + } + drop(st); + + for (i, (ino, ft, name)) in listing.into_iter().enumerate().skip(offset as usize) { + // The stored offset is that of the *next* entry to resume at. + if reply.add(INodeNo(ino), (i + 1) as u64, ft, &name) { + break; + } + } + reply.ok(); + } + + fn attr(&self, ino: u64, node: &Node) -> FileAttr { + let (kind, perm) = match node.kind { + NodeKind::Folder => (FileType::Directory, 0o755), + NodeKind::File { .. } => (FileType::RegularFile, 0o644), + }; + let size = node_size(node); + let mtime = unix_secs(node.modification_time); + let crtime = unix_secs(node.creation_time); + FileAttr { + ino: INodeNo(ino), + size, + blocks: size.div_ceil(512), + atime: mtime, + mtime, + ctime: mtime, + crtime, + kind, + perm, + nlink: if kind == FileType::Directory { 2 } else { 1 }, + uid: self.uid, + gid: self.gid, + rdev: 0, + blksize: 512, + flags: 0, + } + } + + /// Trash the child `name` under `parent` on the remote, then drop it from the + /// local cache. Backs both `unlink` and `rmdir` (Proton trashes whole + /// subtrees, so an `rmdir` of a non-empty dir behaves the same). + fn trash_child(&self, parent: u64, name: &str, reply: ReplyEmpty) { + let (ino, uid) = match self.core.lookup_child(parent, name) { + Ok(x) => x, + Err(e) => { + reply.error(e); + return; + } + }; + let open_now = self + .core + .state + .lock() + .entries + .get(&ino) + .is_some_and(|e| e.open_count > 0); + // A node the server has never heard of cannot be trashed there; deleting + // it just means its queued creation is no longer wanted. This works + // offline, which the remote path below cannot (offline.md Phase 3b). + if is_local_uid(&uid) { + if !open_now { + self.core.discard_queued_ops(&uid); + } + self.core.state.lock().forget_or_unlink(&uid); + debug!(%uid, name, "deleted a node that had not been created remotely yet"); + reply.ok(); + return; + } + // Offline: queue it. Trashing is the one mutation a user expects to work + // regardless — the file is gone from their point of view the moment the + // command returns (offline.md Phase 3b). + if !self.core.online.load(Ordering::Relaxed) { + match self.core.queue_trash(&uid, name) { + Ok(()) => { + self.core.state.lock().forget_or_unlink(&uid); + reply.ok(); + } + Err(e) => reply.error(e), + } + return; + } + if let Err(e) = self + .core + .rt + .block_on(self.core.client.trash_nodes(std::slice::from_ref(&uid))) + { + error!(%uid, error = %e, "trash failed"); + self.core + .log_activity(ActivityKind::Trash, name, e.to_string(), false); + reply.error(Errno::EIO); + return; + } + // Only remove the namespace entry after Drive accepted the mutation. + // Otherwise an EIO response still made the path disappear locally. + self.core.state.lock().forget_or_unlink(&uid); + if !open_now { + self.core.discard_queued_ops(&uid); + self.core.cache.evict(&uid); + self.core.evict_reader(&uid); + } + self.core.invalidate_trash(); + // Every other trash site records itself; this one did not, which made a + // file found in the trash impossible to attribute after the fact — the + // activity log was the only record and it showed nothing (bugs.md B2). + self.core + .log_activity(ActivityKind::Trash, name, "trashed from the mount", true); + reply.ok(); + } +} + +impl ProtonFs { + /// The body of [`Filesystem::getxattr`], on a worker thread. + fn serve_getxattr(&self, ino: INodeNo, name: &OsStr, size: u32, reply: ReplyXattr) { + let ttype = match name.to_str() { + Some(XATTR_THUMBNAIL) => ThumbnailType::Thumbnail, + Some(XATTR_PREVIEW) => ThumbnailType::Preview, + // Any other attribute simply does not exist on this filesystem. + _ => { + reply.error(Errno::ENODATA); + return; + } + }; + let bytes = match self.core.thumbnail(ino.0, ttype) { + Ok(Some(b)) => b, + Ok(None) => { + reply.error(Errno::ENODATA); + return; + } + Err(e) => { + reply.error(e); + return; + } + }; + let len = bytes.len() as u32; + // The xattr protocol: a zero `size` is a probe for the length; otherwise + // the caller's buffer must be large enough or it gets `ERANGE`. + if size == 0 { + reply.size(len); + } else if size < len { + reply.error(Errno::ERANGE); + } else { + reply.data(&bytes); + } + } + + /// Advertise the thumbnail/preview attribute names, but only for files whose + /// media type can actually carry one. + /// + /// Listing them for every file is what made `ls -l` expensive: an xattr-aware + /// lister asks for each advertised name, and each ask that misses is a network + /// round trip (B5). Proton only ever generates thumbnails for images and + /// videos, so advertising them on a `.mkv` or a `.pdf` is an invitation to do + /// work that can only end in `ENODATA`. `getxattr` still serves an explicit + /// request for an unadvertised name, so nothing becomes unreachable. + fn serve_listxattr(&self, ino: INodeNo, size: u32, reply: ReplyXattr) { + let thumbnailable = { + let st = self.core.state.lock(); + match st.entries.get(&ino.0) { + Some(e) => match &e.node.kind { + NodeKind::File { media_type, .. } => { + media_type.starts_with("image/") || media_type.starts_with("video/") + } + NodeKind::Folder => false, + }, + None => { + reply.error(Errno::ENOENT); + return; + } + } + }; + // xattr names are returned as a NUL-terminated, concatenated list. + let mut buf = Vec::new(); + if thumbnailable { + for name in [XATTR_THUMBNAIL, XATTR_PREVIEW] { + buf.extend_from_slice(name.as_bytes()); + buf.push(0); + } + } + let len = buf.len() as u32; + if size == 0 { + reply.size(len); + } else if size < len { + reply.error(Errno::ERANGE); + } else { + reply.data(&buf); + } + } +} diff --git a/crates/pdfs-fuse/src/lib.rs b/crates/pdfs-fuse/src/lib.rs index ff1ad4d..1eea574 100644 --- a/crates/pdfs-fuse/src/lib.rs +++ b/crates/pdfs-fuse/src/lib.rs @@ -11,23 +11,23 @@ //! remote changes still show up promptly. //! //! Phase 3 makes the mount writable. Each file opened for writing gets a -//! [`WriteHandle`] whose buffer accumulates the full plaintext; on flush/release +//! `WriteHandle` whose buffer accumulates the full plaintext; on flush/release //! the buffer is sealed as a new revision via //! [`ProtonDriveClient::upload_new_revision`] (the SDK uploads whole revisions, //! not byte ranges). New files are created empty up front so they get a real //! uid; namespace ops map to `create_folder`, `trash_nodes`, `rename_node` and //! `move_node`. //! -//! Phase 4 adds Files-On-Demand pinning. A control socket (see [`control`]) +//! Phase 4 adds Files-On-Demand pinning. A control socket (see `control`) //! lets the CLI pin/unpin files; a pinned file's plaintext is downloaded once //! into the on-disk [`ContentCache`] and every later `read` is served from disk //! instead of the network. Writes and remote events evict the cache so it never //! goes stale. //! -//! Reads of unpinned files no longer hit the network per call: [`Core::read_range`] +//! Reads of unpinned files no longer hit the network per call: `Core::read_range` //! fetches and caches [`BLOCK_SIZE`]-aligned blocks, so sequential/sparse reads -//! reuse the on-disk block cache. Writes are disk-backed: each [`WriteHandle`] -//! stages authored bytes in a scratch file and tracks them with an [`Intervals`] +//! reuse the on-disk block cache. Writes are disk-backed: each `WriteHandle` +//! stages authored bytes in a scratch file and tracks them with an `Intervals` //! set, so a multi-GiB write never buffers in RAM and only the untouched //! remainder of the file is pulled from the remote — lazily, at commit. @@ -70,10 +70,14 @@ use proton_drive_rs::{ ProtonPhotosClient, RevisionReader, ThumbnailType, }; +mod background; mod control; use control::run_control_socket; mod devices; mod drain; +mod filesystem; +pub use filesystem::ProtonFs; +mod mount; mod photos; mod profile; mod sharing; @@ -81,6 +85,10 @@ mod state; mod sync; mod transfers; 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 state::{Entry, Intervals, PendingRevision, State, WriteHandle}; use tracing::{debug, error, info, warn}; use transfers::{CountingWriter, JobGuard, OwnedCountingReader, TransferRegistry}; @@ -3305,248 +3313,6 @@ fn parse_uid(s: &str) -> Option { let (vol, link) = s.split_once('~')?; Some(NodeUid::new(VolumeId::from(vol), LinkId::from(link))) } - -/// The Proton Drive VFS. FUSE callbacks are synchronous, so the Tokio handle -/// bridges each one to the async SDK via [`Handle::block_on`]; the fuser -/// session thread is not a runtime worker, so blocking on it is sound. -/// Cloneable so a handler can move a copy onto a [`Workers`] thread and answer -/// from there; every field is a handle or a plain id. -#[derive(Clone)] -pub struct ProtonFs { - core: Core, - uid: u32, - gid: u32, -} - -impl ProtonFs { - /// Build the filesystem rooted at `root` (the user's My Files folder). - fn new(core: Core, root: Node) -> Self { - { - let mut st = core.state.lock(); - if let Err(e) = st.db.upsert_node(&root) { - warn!(uid = %root.uid, error = %e, "db upsert root failed"); - } - st.by_uid.insert(root.uid.clone(), ROOT_INO); - st.entries.insert( - ROOT_INO, - Entry { - uid: root.uid.clone(), - parent: ROOT_INO, - node: root, - lookup_count: 1, - open_count: 0, - unlinked: false, - }, - ); - } - // SAFETY: geteuid/getegid are infallible and have no preconditions. - let (uid, gid) = unsafe { (libc::geteuid(), libc::getegid()) }; - Self { core, uid, gid } - } - - /// The body of [`Filesystem::lookup`], on whichever thread ends up serving it. - /// `off_loop` says whether the caller is already on a worker, and so whether - /// this may block. `lookup` answers a cached parent on the dispatch loop, - /// where it may not. - fn serve_lookup(&self, parent: u64, name: &str, reply: ReplyEntry, off_loop: bool) { - if let Err(e) = self.core.ensure_children(parent) { - reply.error(e); - return; - } - let hit = { - let st = self.core.state.lock(); - st.children.get(&parent).and_then(|kids| { - kids.iter().copied().find_map(|ino| { - st.entries - .get(&ino) - .filter(|e| e.node.name == name) - .map(|e| { - // A `lookup` reply carries attrs with the same TTL a - // `getattr` would, so `ls -l` — which is one `lookup` - // per entry and no `getattr` at all — takes its sizes - // from here. Resolving only in `getattr` left the whole - // listing provisional (bugs.md B14). - let provisional = matches!( - &e.node.kind, - NodeKind::File { - claimed_size: None, - .. - } - ) - .then(|| (e.parent, e.uid.clone())); - (ino, self.attr(ino, &e.node), provisional) - }) - }) - }) - }; - let Some((ino, attr, provisional)) = hit else { - reply.error(Errno::ENOENT); - return; - }; - let Some((grandparent, uid)) = provisional else { - reply.entry(&TTL, &attr, Generation(0)); - return; - }; - // Resolving goes to the network. On the dispatch loop that is not - // allowed (the B5 lesson), so hand off — but only from the loop: doing - // it while already on a worker would queue a job onto the lane this - // thread occupies. - if !off_loop { - let fs = self.clone(); - let name = name.to_string(); - self.core.workers.run(Lane::Meta, move || { - fs.serve_lookup(parent, &name, reply, true) - }); - return; - } - self.core.upgrade_sizes_for_parent(ino, &uid, grandparent); - // Re-read: on timeout or failure this is the provisional attr again, - // which is no worse than what this path used to always send. - let attr = { - let st = self.core.state.lock(); - st.entries - .get(&ino) - .map_or(attr, |e| self.attr(ino, &e.node)) - }; - reply.entry(&TTL, &attr, Generation(0)); - } - - /// The body of [`Filesystem::readdir`], on whichever thread ends up serving it. - fn serve_readdir(&self, ino: u64, offset: u64, mut reply: ReplyDirectory) { - if let Err(e) = self.core.ensure_children(ino) { - reply.error(e); - return; - } - let st = self.core.state.lock(); - let parent = st.entries.get(&ino).map_or(ROOT_INO, |e| e.parent); - - // "." and ".." occupy offsets 0 and 1; real children follow. - let mut listing: Vec<(u64, FileType, String)> = vec![ - (ino, FileType::Directory, ".".to_string()), - (parent, FileType::Directory, "..".to_string()), - ]; - if let Some(kids) = st.children.get(&ino) { - for &kid in kids { - if let Some(e) = st.entries.get(&kid) { - let ft = if e.node.is_folder() { - FileType::Directory - } else { - FileType::RegularFile - }; - listing.push((kid, ft, e.node.name.clone())); - } - } - } - drop(st); - - for (i, (ino, ft, name)) in listing.into_iter().enumerate().skip(offset as usize) { - // The stored offset is that of the *next* entry to resume at. - if reply.add(INodeNo(ino), (i + 1) as u64, ft, &name) { - break; - } - } - reply.ok(); - } - - fn attr(&self, ino: u64, node: &Node) -> FileAttr { - let (kind, perm) = match node.kind { - NodeKind::Folder => (FileType::Directory, 0o755), - NodeKind::File { .. } => (FileType::RegularFile, 0o644), - }; - let size = node_size(node); - let mtime = unix_secs(node.modification_time); - let crtime = unix_secs(node.creation_time); - FileAttr { - ino: INodeNo(ino), - size, - blocks: size.div_ceil(512), - atime: mtime, - mtime, - ctime: mtime, - crtime, - kind, - perm, - nlink: if kind == FileType::Directory { 2 } else { 1 }, - uid: self.uid, - gid: self.gid, - rdev: 0, - blksize: 512, - flags: 0, - } - } - - /// Trash the child `name` under `parent` on the remote, then drop it from the - /// local cache. Backs both `unlink` and `rmdir` (Proton trashes whole - /// subtrees, so an `rmdir` of a non-empty dir behaves the same). - fn trash_child(&self, parent: u64, name: &str, reply: ReplyEmpty) { - let (ino, uid) = match self.core.lookup_child(parent, name) { - Ok(x) => x, - Err(e) => { - reply.error(e); - return; - } - }; - let open_now = self - .core - .state - .lock() - .entries - .get(&ino) - .is_some_and(|e| e.open_count > 0); - // A node the server has never heard of cannot be trashed there; deleting - // it just means its queued creation is no longer wanted. This works - // offline, which the remote path below cannot (offline.md Phase 3b). - if is_local_uid(&uid) { - if !open_now { - self.core.discard_queued_ops(&uid); - } - self.core.state.lock().forget_or_unlink(&uid); - debug!(%uid, name, "deleted a node that had not been created remotely yet"); - reply.ok(); - return; - } - // Offline: queue it. Trashing is the one mutation a user expects to work - // regardless — the file is gone from their point of view the moment the - // command returns (offline.md Phase 3b). - if !self.core.online.load(Ordering::Relaxed) { - match self.core.queue_trash(&uid, name) { - Ok(()) => { - self.core.state.lock().forget_or_unlink(&uid); - reply.ok(); - } - Err(e) => reply.error(e), - } - return; - } - if let Err(e) = self - .core - .rt - .block_on(self.core.client.trash_nodes(std::slice::from_ref(&uid))) - { - error!(%uid, error = %e, "trash failed"); - self.core - .log_activity(ActivityKind::Trash, name, e.to_string(), false); - reply.error(Errno::EIO); - return; - } - // Only remove the namespace entry after Drive accepted the mutation. - // Otherwise an EIO response still made the path disappear locally. - self.core.state.lock().forget_or_unlink(&uid); - if !open_now { - self.core.discard_queued_ops(&uid); - self.core.cache.evict(&uid); - self.core.evict_reader(&uid); - } - self.core.invalidate_trash(); - // Every other trash site records itself; this one did not, which made a - // file found in the trash impossible to attribute after the fact — the - // activity log was the only record and it showed nothing (bugs.md B2). - self.core - .log_activity(ActivityKind::Trash, name, "trashed from the mount", true); - reply.ok(); - } -} - /// Current wall-clock time as epoch seconds (0 if the clock is before the epoch). fn now_secs() -> i64 { SystemTime::now() @@ -3978,1834 +3744,6 @@ fn unix_secs(secs: i64) -> SystemTime { } } -impl Filesystem for ProtonFs { - fn lookup(&self, _req: &Request, parent: INodeNo, name: &OsStr, reply: ReplyEntry) { - let parent = parent.0; - let name = name.to_string_lossy().into_owned(); - // A folder that has not been listed yet is enumerated from the remote, - // so serve it from a worker rather than stalling the dispatch loop. A - // listed folder — the common case — is a map hit, and answering it - // inline costs less than the handoff would. - if self.core.children_cached(parent) { - self.serve_lookup(parent, &name, reply, false); - return; - } - let fs = self.clone(); - self.core.workers.run(Lane::Meta, move || { - fs.serve_lookup(parent, &name, reply, true) - }); - } - - fn forget(&self, _req: &Request, ino: INodeNo, nlookup: u64) { - let mut st = self.core.state.lock(); - st.forget_lookup(ino.0, nlookup); - } - - fn getattr(&self, _req: &Request, ino: INodeNo, _fh: Option, reply: ReplyAttr) { - let (attr, provisional_in) = { - let st = self.core.state.lock(); - match st.entries.get(&ino.0) { - Some(e) => { - // A file still carrying the cheap enumeration's placeholder - // size (B12). - let provisional = matches!( - &e.node.kind, - NodeKind::File { - claimed_size: None, - .. - } - ) - .then(|| (e.parent, e.uid.clone())); - (self.attr(ino.0, &e.node), provisional) - } - None => { - reply.error(Errno::ENOENT); - return; - } - } - }; - // The common case: the size is real, answer from the lock we just held. - let Some((parent, uid)) = provisional_in else { - reply.attr(&TTL, &attr); - return; - }; - // A provisional size is the *ciphertext* size, which is larger than the - // file. Publishing it makes every reader that trusts `st_size` — rsync, - // mmap, sendfile, a sized read loop — run off the end of the file and - // fail. So resolve it before answering rather than after (bugs.md B14). - // - // The cost is one batched round trip for the whole folder, not one per - // file: `ls -l` is one `getattr` per entry and they collapse onto a - // single upgrade. This is why B12's split is still worth having — the - // plain listing never comes here at all. - // - // Off the dispatch loop, because it goes to the network: blocking there - // would stall every other operation on the mount (the B5 lesson). - let fs = self.clone(); - self.core.workers.run(Lane::Meta, move || { - fs.core.upgrade_sizes_for_parent(ino.0, &uid, parent); - // Re-read: the upgrade writes through `state`, and on timeout or - // failure this is simply the provisional attr we already had. - let attr = { - let st = fs.core.state.lock(); - match st.entries.get(&ino.0) { - Some(e) => fs.attr(ino.0, &e.node), - // Forgotten while we waited. - None => attr, - } - }; - reply.attr(&TTL, &attr); - }); - } - - fn readdir( - &self, - _req: &Request, - ino: INodeNo, - _fh: FileHandle, - offset: u64, - reply: ReplyDirectory, - ) { - let ino = ino.0; - // Same split as `lookup`: only the cold, remote-enumerating path pays - // for a worker. - if self.core.children_cached(ino) { - self.serve_readdir(ino, offset, reply); - return; - } - let fs = self.clone(); - self.core - .workers - .run(Lane::Meta, move || fs.serve_readdir(ino, offset, reply)); - } - - fn open(&self, _req: &Request, ino: INodeNo, flags: OpenFlags, reply: ReplyOpen) { - let (uid, base_mtime, base_size) = { - let mut st = self.core.state.lock(); - match st.entries.get_mut(&ino.0) { - Some(e) if e.node.is_file() => { - e.open_count = e.open_count.saturating_add(1); - (e.uid.clone(), e.node.modification_time, node_size(&e.node)) - } - Some(_) => { - reply.error(Errno::EISDIR); - return; - } - None => { - reply.error(Errno::ENOENT); - return; - } - } - }; - debug!(ino = ino.0, ?flags, base_size, base_mtime, "open"); - // Read-only opens stay stateless (fh 0). A write open allocates a - // disk-backed handle; bytes are authored into a scratch file and the - // untouched remainder is pulled from the base lazily (on read / commit). - if flags.acc_mode() == OpenAccMode::O_RDONLY { - reply.opened(FileHandle(0), FopenFlags::empty()); - return; - } - let (file, path) = match self.core.cache.create_scratch() { - Ok(x) => x, - Err(e) => { - if let Some(entry) = self.core.state.lock().entries.get_mut(&ino.0) { - entry.open_count = entry.open_count.saturating_sub(1); - } - error!(%uid, error = %e, "create scratch file failed"); - reply.error(Errno::EIO); - return; - } - }; - let mut st = self.core.state.lock(); - let fh = st.next_fh; - st.next_fh += 1; - let aw = st.active_writes.entry(ino.0).or_insert_with(|| { - WriteHandle { - ino: ino.0, - uid, - file: Arc::new(file), - path, - written: Intervals::default(), - // Starts at the current size; reads in [0, base_size) come from - // the base until overwritten. - len: base_size, - base_size, - base_mtime, - dirty: false, - open_count: 0, - } - }); - aw.open_count += 1; - st.handles.insert(fh, ino.0); - reply.opened(FileHandle(fh), FopenFlags::empty()); - } - - fn read( - &self, - _req: &Request, - ino: INodeNo, - _fh: FileHandle, - offset: u64, - size: u32, - _flags: OpenFlags, - _lock_owner: Option, - reply: ReplyData, - ) { - // A file open for writing is served from its handle so reads see the - // in-flight (possibly unsaved) content: authored bytes come from the - // scratch file, untouched bytes from the base. - // - // This path stays on the dispatch loop. `write` runs there too, so - // keeping its reads there as well preserves the ordering the write path - // has always had between a write and a read of the same handle; a mostly - // local read is not worth reasoning about that for. - let handle = { - let st = self.core.state.lock(); - st.active_writes.get(&ino.0).map(|h| { - ( - h.file.clone(), - h.len, - h.uid.clone(), - h.base_mtime, - h.base_size, - h.written.clone(), - ) - }) - }; - if let Some((file, len, uid, base_mtime, base_size, written)) = handle { - match self.core.serve_open_read( - &file, - len, - &uid, - base_mtime, - base_size, - &written, - offset, - size as u64, - ) { - Ok(bytes) => reply.data(&bytes), - Err(e) => reply.error(e), - } - return; - } - let (uid, mtime, fsize, is_video) = { - let st = self.core.state.lock(); - match st.entries.get(&ino.0) { - Some(e) if e.node.is_file() => { - let media_type = match &e.node.kind { - NodeKind::File { media_type, .. } => Some(media_type.as_str()), - NodeKind::Folder => None, - }; - let is_video = - PhotoKind::classify(Some(&e.node.name), media_type) == PhotoKind::Video; - let fsize = self - .core - .pending - .lock() - .get(&e.uid) - .map(|p| p.meta.len) - .unwrap_or_else(|| node_size(&e.node)); - (e.uid.clone(), e.node.modification_time, fsize, is_video) - } - Some(_) => { - reply.error(Errno::EISDIR); - return; - } - None => { - reply.error(Errno::ENOENT); - return; - } - } - }; - // A large unpinned video streams without polluting the block cache (see - // [`STREAM_BYPASS_MIN`]); everything else caches its blocks as usual. - let cache_blocks = - !(is_video && fsize >= STREAM_BYPASS_MIN && !self.core.cache.is_pinned(&uid)); - // Off the dispatch loop: this is the one handler that routinely goes to - // the network (block fetch + decrypt, hundreds of ms on a cold file), - // and until it returns the kernel's next request for this mount is not - // even read off the FUSE device. It only reads `state`, so moving it - // races with nothing. FUSE does not require replies in request order. - let core = self.core.clone(); - self.core.workers.run(Lane::Transfer, move || { - match core.read_range(&uid, mtime, fsize, offset, size as u64, cache_blocks) { - Ok(bytes) => reply.data(&bytes), - Err(e) => reply.error(e), - } - }); - } - - fn create( - &self, - _req: &Request, - parent: INodeNo, - name: &OsStr, - _mode: u32, - _umask: u32, - _flags: i32, - reply: ReplyCreate, - ) { - let parent = parent.0; - if let Err(e) = self.core.ensure_children(parent) { - reply.error(e); - return; - } - let name = name.to_string_lossy().into_owned(); - let parent_uid = { - let st = self.core.state.lock(); - match st.entries.get(&parent) { - Some(e) => e.uid.clone(), - None => { - reply.error(Errno::ENOENT); - return; - } - } - }; - // Offline the server cannot mint a uid, so invent one and queue the - // create. The file is real to the caller either way; only its identity is - // provisional until the drain (offline.md Phase 3b). - // - // A parent that is itself still queued forces the same path even when we - // are online: the API has no folder to put this in yet. - let node = if self.core.online.load(Ordering::Relaxed) && !is_local_uid(&parent_uid) { - // Create an empty file on the remote so it has a real uid immediately; - // written bytes are buffered and sealed as a new revision on close. - let new_uid = match self.core.rt.block_on(self.core.client.upload_file( - &parent_uid, - &name, - media_type_for(&name), - b"", - )) { - Ok(u) => u, - Err(e) => { - error!(%parent_uid, name, error = %e, "create file failed"); - reply.error(Errno::EIO); - return; - } - }; - match self.core.fetch_node(&new_uid) { - Ok(n) => n, - Err(e) => { - reply.error(e); - return; - } - } - } else { - match self.core.queue_local_node(&parent_uid, &name, false) { - Ok(n) => n, - Err(e) => { - reply.error(e); - return; - } - } - }; - let new_uid = node.uid.clone(); - // The base this handle writes over is the node as it actually exists — - // the empty file the server just minted — so its modification time comes - // from the node, never from the local clock. `queue_revision` turns this - // into `StagedWrite::based_on` and the drain compares that against the - // remote: a `now_secs()` here differs from the server's stamp by however - // long the create round-trip took, so the first write to a brand-new file - // conflicts with its own create whenever the second happens to tick in - // between. - let base_mtime = node.modification_time; - let (file, path) = match self.core.cache.create_scratch() { - Ok(x) => x, - Err(e) => { - error!(%new_uid, error = %e, "create scratch file failed"); - reply.error(Errno::EIO); - return; - } - }; - let mut st = self.core.state.lock(); - let ino = st.intern(parent, node); - if let Some(entry) = st.entries.get_mut(&ino) { - entry.open_count = entry.open_count.saturating_add(1); - } - if let Some(kids) = st.children.get_mut(&parent) - && !kids.contains(&ino) - { - kids.push(ino); - } - let fh = st.next_fh; - st.next_fh += 1; - let aw = st.active_writes.entry(ino).or_insert_with(|| { - // A brand-new file: empty base, everything written is authored. - WriteHandle { - ino, - uid: new_uid, - file: Arc::new(file), - path, - written: Intervals::default(), - len: 0, - base_size: 0, - base_mtime, - dirty: false, - open_count: 0, - } - }); - aw.open_count += 1; - st.handles.insert(fh, ino); - let attr = self.attr(ino, &st.entries.get(&ino).unwrap().node); - reply.created( - &TTL, - &attr, - Generation(0), - FileHandle(fh), - FopenFlags::empty(), - ); - } - - fn write( - &self, - _req: &Request, - ino: INodeNo, - fh: FileHandle, - offset: u64, - data: &[u8], - _write_flags: WriteFlags, - _flags: OpenFlags, - _lock_owner: Option, - reply: ReplyWrite, - ) { - let fh = fh.0; - // Stage the bytes straight into the scratch file (no base download): only - // the untouched remainder is pulled from the remote, and only at commit. - let file = { - let st = self.core.state.lock(); - match st.handles.get(&fh).and_then(|&i| st.active_writes.get(&i)) { - Some(aw) => aw.file.clone(), - None => { - reply.error(Errno::EBADF); - return; - } - } - }; - if let Err(e) = file.write_all_at(data, offset) { - error!(ino = ino.0, fh, error = %e, "scratch write failed"); - reply.error(Errno::EIO); - return; - } - let new_len = { - let mut st = self.core.state.lock(); - let Some(aw) = st - .handles - .get(&fh) - .copied() - .and_then(|i| st.active_writes.get_mut(&i)) - else { - reply.error(Errno::EBADF); - return; - }; - let end = offset + data.len() as u64; - aw.written.add(offset, end); - aw.len = aw.len.max(end); - aw.dirty = true; - let len = aw.len; - st.set_size(ino.0, len); - len - }; - debug!( - ino = ino.0, - fh, - offset, - len = data.len(), - new_len, - "staged write" - ); - reply.written(data.len() as u32); - } - - #[allow(clippy::too_many_arguments)] - fn setattr( - &self, - _req: &Request, - ino: INodeNo, - _mode: Option, - _uid: Option, - _gid: Option, - size: Option, - _atime: Option, - _mtime: Option, - _ctime: Option, - fh: Option, - _crtime: Option, - _chgtime: Option, - _bkuptime: Option, - _flags: Option, - reply: ReplyAttr, - ) { - // Only resizes change remote state; everything else (mode/owner/times) is - // accepted and ignored so tools that chmod/utimes after writing succeed. - if let Some(size) = size { - let fh = fh.map(|f| f.0).filter(|&f| f != 0); - let handled = { - let mut st = self.core.state.lock(); - match st.active_writes.get_mut(&ino.0) { - Some(aw) => { - if size < aw.len { - // Shrink: drop authored ranges past the new end. - aw.written.clip(size); - } else if size > aw.len { - // Grow: the new tail is defined as zeros, so claim - // it as authored rather than base content. - aw.written.add(aw.len, size); - } - let _ = aw.file.set_len(size); - aw.len = size; - aw.dirty = true; - true - } - None => false, - } - }; - debug!(ino = ino.0, size, fh = ?fh, handled, "setattr resize"); - if !handled { - // Path-based truncate with no open write handle — a shell's - // `> file`. This is the second write path into the API, and - // queueing it is what lets a redirect work offline at all: it - // never reaches `release`, so without this it failed before any - // byte was written (offline.md Phase 2/3b). - if let Err(e) = self.core.queue_truncate(ino.0, size) { - reply.error(e); - return; - } - } else { - self.core.state.lock().set_size(ino.0, size); - } - } - let st = self.core.state.lock(); - match st.entries.get(&ino.0) { - Some(e) => { - let attr = self.attr(ino.0, &e.node); - reply.attr(&TTL, &attr); - } - None => reply.error(Errno::ENOENT), - } - } - - fn fallocate( - &self, - _req: &Request, - ino: INodeNo, - _fh: FileHandle, - offset: u64, - length: u64, - mode: i32, - reply: ReplyEmpty, - ) { - let new_end = offset.saturating_add(length); - let keep_size = (mode & 1) != 0; // FALLOC_FL_KEEP_SIZE - - let is_punch_hole = (mode & 0x02) != 0; // FALLOC_FL_PUNCH_HOLE - - let res = { - let mut st = self.core.state.lock(); - match st - .handles - .get(&_fh.0) - .copied() - .and_then(|i| st.active_writes.get_mut(&i)) - { - Some(aw) => { - use std::os::unix::io::AsRawFd; - let fd = aw.file.as_raw_fd(); - let ret = unsafe { libc::fallocate(fd, mode, offset as i64, length as i64) }; - if ret == 0 { - if is_punch_hole { - // A punched range reads as zero. Mark it authored so - // queue_revision does not refill it from the remote - // baseline and silently undo the hole at commit. - aw.written.add(offset, new_end.min(aw.len)); - aw.dirty = true; - } else if !keep_size && new_end > aw.len { - aw.written.add(aw.len, new_end); - aw.len = new_end; - aw.dirty = true; - st.set_size(ino.0, new_end); - } - Ok(()) - } else { - Err(Errno::EIO) - } - } - None => { - if st.entries.contains_key(&ino.0) { - Err(Errno::EBADF) - } else { - Err(Errno::ENOENT) - } - } - } - }; - - match res { - Ok(_) => reply.ok(), - Err(e) => reply.error(e), - } - } - - /// `close(2)` calls flush before release. The upload is queued in `release` - /// and performed in the background (offline.md Phase 3), so there is nothing - /// to push here — the written bytes are already in the scratch file. - fn flush( - &self, - _req: &Request, - _ino: INodeNo, - _fh: FileHandle, - _lock_owner: LockOwner, - reply: ReplyEmpty, - ) { - reply.ok(); - } - - /// Durability here means the scratch file, not the remote: a caller asking - /// for fsync wants its bytes to survive a crash, and blocking it on an upload - /// (which offline would never finish) buys nothing the queue does not already - /// guarantee. - fn fsync( - &self, - _req: &Request, - _ino: INodeNo, - fh: FileHandle, - _datasync: bool, - reply: ReplyEmpty, - ) { - // Everything the durability sidecar needs, copied out so no lock is held - // across the sync. - let handle = { - let st = self.core.state.lock(); - let ino = st.handles.get(&fh.0).copied(); - ino.and_then(|i| st.active_writes.get(&i)).map(|aw| { - ( - aw.file.clone(), - aw.path.clone(), - aw.uid.clone(), - aw.dirty, - aw.written.clone(), - aw.len, - aw.base_size, - aw.base_mtime, - ) - }) - }; - let Some((f, path, uid, dirty, written, len, base_size, base_mtime)) = handle else { - reply.error(Errno::EBADF); - return; - }; - if let Err(e) = f.sync_all() { - error!(fh = fh.0, error = %e, "fsync of scratch file failed"); - reply.error(Errno::EIO); - return; - } - // A clean handle has nothing on it that the remote does not already - // hold, so there is nothing for a crash to lose. - if !dirty { - reply.ok(); - return; - } - // The bytes are on stable storage, but only the scratch file knows that, - // and the scratch directory is cleared at open. Record what the blob is - // so a restart can hand it to the upload queue instead of deleting it. - // - // Deliberately *not* the full release path: staging the blob here would - // move the file out from under a handle the application still has open, - // and queueing an upload per `fsync` would push a revision for every - // barrier in a write loop. This only makes the existing file findable. - let authored: Vec<(u64, u64)> = written - .segments(0, len) - .into_iter() - .filter(|&(_, _, authored)| authored) - .map(|(s, e, _)| (s, e)) - .collect(); - let meta = StagedWrite { - uid: uid.to_string(), - len, - base_size, - base_mtime, - complete: authored == [(0, len)], - authored, - based_on: self.core.remote_baseline(&uid, base_mtime, base_size), - }; - // A failed sidecar means the write is not durable, and `fsync` promising - // otherwise is the defect this exists to fix — so it is an error, not a - // warning. - if let Err(e) = self.core.cache.mark_scratch_durable(&path, &meta) { - error!(%uid, error = %e, "recording fsync durability failed"); - reply.error(Errno::EIO); - return; - } - reply.ok(); - } - - fn release( - &self, - _req: &Request, - _ino: INodeNo, - fh: FileHandle, - _flags: OpenFlags, - _lock_owner: Option, - _flush: bool, - reply: ReplyEmpty, - ) { - let (handle, unlinked_uid) = { - let mut st = self.core.state.lock(); - let unlinked_uid = if let Some(entry) = st.entries.get_mut(&_ino.0) { - entry.open_count = entry.open_count.saturating_sub(1); - if entry.open_count == 0 && entry.unlinked { - let uid = entry.uid.clone(); - st.forget(&uid); - Some(uid) - } else { - None - } - } else { - None - }; - let ino = st.handles.remove(&fh.0); - let h = ino.and_then(|i| { - let aw = st.active_writes.get_mut(&i)?; - aw.open_count = aw.open_count.saturating_sub(1); - if aw.open_count == 0 { - st.active_writes.remove(&i) - } else { - None - } - }); - (h, unlinked_uid) - }; - let was_unlinked = unlinked_uid.is_some(); - if let Some(uid) = unlinked_uid { - self.core.discard_queued_ops(&uid); - self.core.cache.evict(&uid); - self.core.evict_reader(&uid); - } - // Hand the bytes to the queue rather than uploading them here: the - // scratch file is the only copy of what was just written, and blocking - // the caller on the network is what made a copy into the mount run at - // upload speed (and fail outright offline). - match (handle, was_unlinked) { - (Some(h), true) => { - // POSIX keeps an unlinked file usable until the last close, but - // closing it must not resurrect those bytes as a new revision. - if let Err(e) = std::fs::remove_file(&h.path) - && e.kind() != std::io::ErrorKind::NotFound - { - warn!(path = %h.path.display(), error = %e, "discarding unlinked scratch file failed"); - } - reply.ok(); - } - (Some(h), false) => match self.core.queue_revision(&h) { - Ok(()) => reply.ok(), - Err(e) => reply.error(e), - }, - (None, _) => reply.ok(), - } - } - - fn unlink(&self, _req: &Request, parent: INodeNo, name: &OsStr, reply: ReplyEmpty) { - let name_str = name.to_string_lossy(); - if let Ok((ino, _)) = self.core.lookup_child(parent.0, &name_str) { - let st = self.core.state.lock(); - if let Some(entry) = st.entries.get(&ino) - && entry.node.is_folder() - { - reply.error(Errno::EISDIR); - return; - } - } - self.trash_child(parent.0, &name_str, reply); - } - - fn rmdir(&self, _req: &Request, parent: INodeNo, name: &OsStr, reply: ReplyEmpty) { - let name_str = name.to_string_lossy(); - if let Ok((ino, _)) = self.core.lookup_child(parent.0, &name_str) { - { - let st = self.core.state.lock(); - if st - .entries - .get(&ino) - .is_some_and(|entry| !entry.node.is_folder()) - { - reply.error(Errno::ENOTDIR); - return; - } - } - // A missing cached listing says "unknown", not "empty". Proton's - // trash call is recursive, so enumerate before applying rmdir's - // POSIX emptiness rule. - if let Err(e) = self.core.ensure_children(ino) { - reply.error(e); - return; - } - if self.core.state.lock().has_children(ino) { - reply.error(Errno::ENOTEMPTY); - return; - } - } - self.trash_child(parent.0, &name_str, reply); - } - - fn mkdir( - &self, - _req: &Request, - parent: INodeNo, - name: &OsStr, - _mode: u32, - _umask: u32, - reply: ReplyEntry, - ) { - let parent = parent.0; - if let Err(e) = self.core.ensure_children(parent) { - reply.error(e); - return; - } - let name = name.to_string_lossy().into_owned(); - let parent_uid = { - let st = self.core.state.lock(); - match st.entries.get(&parent) { - Some(e) => e.uid.clone(), - None => { - reply.error(Errno::ENOENT); - return; - } - } - }; - let now = SystemTime::now() - .duration_since(UNIX_EPOCH) - .map(|d| d.as_secs() as i64) - .ok(); - // As in `create`: offline — or under a parent that is itself still - // queued — the folder becomes a placeholder that the drain turns into a - // real one (offline.md Phase 3b). - let node = if self.core.online.load(Ordering::Relaxed) && !is_local_uid(&parent_uid) { - let new_uid = - match self - .core - .rt - .block_on(self.core.client.create_folder(&parent_uid, &name, now)) - { - Ok(u) => u, - Err(e) => { - error!(%parent_uid, name, error = %e, "create folder failed"); - reply.error(Errno::EIO); - return; - } - }; - match self.core.fetch_node(&new_uid) { - Ok(n) => n, - Err(e) => { - reply.error(e); - return; - } - } - } else { - match self.core.queue_local_node(&parent_uid, &name, true) { - Ok(n) => n, - Err(e) => { - reply.error(e); - return; - } - } - }; - let mut st = self.core.state.lock(); - let local = is_local_uid(&node.uid); - let uid = node.uid.clone(); - let ino = st.intern(parent, node); - if let Some(kids) = st.children.get_mut(&parent) - && !kids.contains(&ino) - { - kids.push(ino); - } - // A folder that exists only here has nothing to enumerate, and asking the - // API to enumerate a `local~` uid would 404. Record it as fully listed and - // empty, which it is, so reads stay offline-clean across a restart too. - if local { - st.children.insert(ino, Vec::new()); - if let Err(e) = self.core.db.set_listed(&uid, true) { - warn!(%uid, error = %e, "db set_listed(true) failed for local folder"); - } - } - let attr = self.attr(ino, &st.entries.get(&ino).unwrap().node); - reply.entry(&TTL, &attr, Generation(0)); - } - - fn rename( - &self, - _req: &Request, - parent: INodeNo, - name: &OsStr, - newparent: INodeNo, - newname: &OsStr, - flags: RenameFlags, - reply: ReplyEmpty, - ) { - let parent = parent.0; - let newparent = newparent.0; - let name = name.to_string_lossy().into_owned(); - let newname = newname.to_string_lossy().into_owned(); - let (ino, uid) = match self.core.lookup_child(parent, &name) { - Ok(x) => x, - Err(e) => { - reply.error(e); - return; - } - }; - // Ancestor cycle check: moving a directory into itself or one of its descendants is forbidden by POSIX (EINVAL). - { - let st = self.core.state.lock(); - if let Some(entry) = st.entries.get(&ino) - && entry.node.is_folder() - && st.is_ancestor_of(ino, newparent) - { - reply.error(Errno::EINVAL); - return; - } - } - // The destination has to be listed either way: the queued path pushes - // the node into that listing, and the online one drops it to force a - // re-enumeration. - if newparent != parent - && let Err(e) = self.core.ensure_children(newparent) - { - reply.error(e); - return; - } - let new_parent_uid = { - let st = self.core.state.lock(); - match st.entries.get(&newparent) { - Some(e) => e.uid.clone(), - None => { - reply.error(Errno::ENOENT); - return; - } - } - }; - // `rename(2)` replaces an existing destination; Proton has no replacing - // rename, so the victim has to be removed first and the operation stops - // being atomic. Without this the 422 surfaced as a blanket EIO and every - // write-to-temp-then-rename tool — rsync, atomic editor saves — failed - // at the very end of its transfer (bugs.md B13). - // - // `RENAME_EXCHANGE` has no Proton primitive and cannot be emulated - // without a window in which one of the two names does not exist. - if flags.contains(RenameFlags::RENAME_EXCHANGE) { - reply.error(Errno::EINVAL); - return; - } - let victim = match self.core.lookup_child(newparent, &newname) { - // Renaming a node onto its own name is a no-op, never a self-replace. - Ok((_, vuid)) if vuid == uid => None, - Ok(x) => Some(x), - // Nothing to replace: the overwhelmingly common case. - Err(e) if e.code() == libc::ENOENT => None, - Err(e) => { - reply.error(e); - return; - } - }; - if let Some((victim_ino, victim_uid)) = &victim { - if flags.contains(RenameFlags::RENAME_NOREPLACE) { - reply.error(Errno::EEXIST); - return; - } - // POSIX requires both ends to agree on being a directory, and the - // check has to happen before anything is trashed: getting it wrong - // turns a refusal into the destruction of the destination. - let (src_dir, dst_dir) = { - let st = self.core.state.lock(); - let is_dir = |i: &u64| { - st.entries - .get(i) - .map(|e| matches!(e.node.kind, NodeKind::Folder)) - }; - match (is_dir(&ino), is_dir(victim_ino)) { - (Some(a), Some(b)) => (a, b), - _ => { - reply.error(Errno::ENOENT); - return; - } - } - }; - // Replacing a directory means trashing it, and Proton trashes a - // folder with its whole subtree — so the destination's emptiness has - // to be known before the decision is made. - let dst_empty = if dst_dir { - if let Err(e) = self.core.ensure_children(*victim_ino) { - reply.error(e); - return; - } - self.core - .state - .lock() - .children - .get(victim_ino) - .is_none_or(|kids| kids.is_empty()) - } else { - true - }; - if let Err(e) = check_replaceable(src_dir, dst_dir, dst_empty) { - reply.error(e); - return; - } - if let Err(e) = self.core.remove_replaced(victim_uid, &newname) { - reply.error(e); - return; - } - } - // A node whose own creation is still queued has no server-side identity - // to rename: the queued op *is* the node, so rewriting its target is the - // whole rename. Nothing reaches the API, which is why this works offline - // (offline.md Phase 3b). - if is_local_uid(&uid) { - match self.core.db.rewrite_op_target( - &uid.to_string(), - &new_parent_uid.to_string(), - &newname, - ) { - Ok(true) => { - self.core.state.lock().rename_in_place( - ino, - newparent, - &new_parent_uid, - &newname, - ); - debug!(%uid, newname, "renamed a node whose create is still queued"); - reply.ok(); - } - // The create drained underneath us, so the node has a real uid - // now and this handle's is stale. A retry resolves it. - Ok(false) => { - warn!(%uid, name, newname, "queued create vanished under a rename"); - self.core.restore_replaced(victim.as_ref(), &newname); - reply.error(Errno::EBUSY); - } - Err(e) => { - error!(%uid, error = %e, "rewriting a queued create's target failed"); - self.core.restore_replaced(victim.as_ref(), &newname); - reply.error(Errno::EIO); - } - } - return; - } - // Offline, or into a folder that does not exist remotely yet: queue the - // rename rather than 404 or fail. Online moves into a real parent still - // take the synchronous path, so a genuine API refusal (permissions, a - // name clash) surfaces to the caller instead of becoming a queued op - // that can only ever conflict. - if !self.core.online.load(Ordering::Relaxed) || is_local_uid(&new_parent_uid) { - match self - .core - .queue_rename(ino, &uid, newparent, &new_parent_uid, &newname) - { - Ok(()) => reply.ok(), - Err(e) => { - self.core.restore_replaced(victim.as_ref(), &newname); - reply.error(e); - } - } - return; - } - // Move first if the parent changed, then rename if the name changed. - // A failure past this point has already trashed any node the rename was - // replacing, so put it back rather than leaving the caller with neither - // the source moved nor the destination intact. - if newparent != parent - && let Err(e) = self - .core - .rt - .block_on(self.core.client.move_node(&uid, &new_parent_uid)) - { - error!(%uid, error = %e, "move failed"); - self.core.restore_replaced(victim.as_ref(), &newname); - reply.error(Errno::EIO); - return; - } - if newname != name - && let Err(e) = self - .core - .rt - .block_on(self.core.client.rename_node(&uid, &newname, None)) - { - error!(%uid, error = %e, "rename failed"); - self.core.restore_replaced(victim.as_ref(), &newname); - reply.error(Errno::EIO); - return; - } - self.core - .state - .lock() - .relocate(ino, parent, newparent, &new_parent_uid, &newname); - reply.ok(); - } - - /// Expose a file's server-side thumbnail/preview as an extended attribute, so - /// a previewing client can fetch it without downloading the whole file. The - /// bytes are fetched on demand and cached; absence yields `ENODATA`. - fn getxattr(&self, _req: &Request, ino: INodeNo, name: &OsStr, size: u32, reply: ReplyXattr) { - // Always off the dispatch loop: a miss goes to the wire, and a lister - // that stats a directory issues one of these per file per advertised - // name — inline, that serialized the whole mount behind ~186 ms of - // network per call (B5). - let fs = self.clone(); - let name = name.to_os_string(); - self.core.workers.run(Lane::Meta, move || { - fs.serve_getxattr(ino, &name, size, reply) - }); - } - - fn listxattr(&self, _req: &Request, ino: INodeNo, size: u32, reply: ReplyXattr) { - self.serve_listxattr(ino, size, reply); - } - - fn ioctl( - &self, - _req: &Request, - _ino: INodeNo, - _fh: FileHandle, - _flags: IoctlFlags, - _cmd: u32, - _in_data: &[u8], - _out_size: u32, - reply: ReplyIoctl, - ) { - // Filesystems are not character devices or terminals; returning ENOTTY - // for unhandled ioctls is standard Linux POSIX behavior and prevents - // fuser warning logs when applications probe file descriptors. - reply.error(Errno::ENOTTY); - } -} - -impl ProtonFs { - /// The body of [`Filesystem::getxattr`], on a worker thread. - fn serve_getxattr(&self, ino: INodeNo, name: &OsStr, size: u32, reply: ReplyXattr) { - let ttype = match name.to_str() { - Some(XATTR_THUMBNAIL) => ThumbnailType::Thumbnail, - Some(XATTR_PREVIEW) => ThumbnailType::Preview, - // Any other attribute simply does not exist on this filesystem. - _ => { - reply.error(Errno::ENODATA); - return; - } - }; - let bytes = match self.core.thumbnail(ino.0, ttype) { - Ok(Some(b)) => b, - Ok(None) => { - reply.error(Errno::ENODATA); - return; - } - Err(e) => { - reply.error(e); - return; - } - }; - let len = bytes.len() as u32; - // The xattr protocol: a zero `size` is a probe for the length; otherwise - // the caller's buffer must be large enough or it gets `ERANGE`. - if size == 0 { - reply.size(len); - } else if size < len { - reply.error(Errno::ERANGE); - } else { - reply.data(&bytes); - } - } - - /// Advertise the thumbnail/preview attribute names, but only for files whose - /// media type can actually carry one. - /// - /// Listing them for every file is what made `ls -l` expensive: an xattr-aware - /// lister asks for each advertised name, and each ask that misses is a network - /// round trip (B5). Proton only ever generates thumbnails for images and - /// videos, so advertising them on a `.mkv` or a `.pdf` is an invitation to do - /// work that can only end in `ENODATA`. `getxattr` still serves an explicit - /// request for an unadvertised name, so nothing becomes unreachable. - fn serve_listxattr(&self, ino: INodeNo, size: u32, reply: ReplyXattr) { - let thumbnailable = { - let st = self.core.state.lock(); - match st.entries.get(&ino.0) { - Some(e) => match &e.node.kind { - NodeKind::File { media_type, .. } => { - media_type.starts_with("image/") || media_type.starts_with("video/") - } - NodeKind::Folder => false, - }, - None => { - reply.error(Errno::ENOENT); - return; - } - } - }; - // xattr names are returned as a NUL-terminated, concatenated list. - let mut buf = Vec::new(); - if thumbnailable { - for name in [XATTR_THUMBNAIL, XATTR_PREVIEW] { - buf.extend_from_slice(name.as_bytes()); - buf.push(0); - } - } - let len = buf.len() as u32; - if size == 0 { - reply.size(len); - } else if size < len { - reply.error(Errno::ERANGE); - } else { - reply.data(&buf); - } - } -} - -/// Apply one remote event to the local cache and notify the kernel so it drops -/// any stale cached metadata/data for the affected inodes. -/// -/// The cache is authoritative-by-absence: dropping a directory's `children` -/// entry forces the next `lookup`/`readdir` to re-enumerate from the remote, so -/// most events only need to invalidate listings rather than re-fetch eagerly. -fn apply_event( - state: &Mutex, - content: &ContentCache, - pending: &Mutex>, - notifier: &Notifier, - event: &DriveEvent, -) { - match event { - DriveEvent::NodeUpdated { - node_uid, - parent_node_uid, - is_trashed, - .. - } => { - let mut st = state.lock(); - if *is_trashed { - // Trashing makes a node vanish from its parent listing. - let child = st.by_uid.get(node_uid).copied(); - if let Some((parent, name)) = st.forget(node_uid) { - content.evict(node_uid); - match child { - Some(child) => { - let _ = - notifier.delete(INodeNo(parent), INodeNo(child), OsStr::new(&name)); - } - None => { - let _ = notifier.inval_entry(INodeNo(parent), OsStr::new(&name)); - } - } - } - } else if pending.lock().contains_key(node_uid) { - // A node we owe an upload for is *ahead* of the remote, not - // behind it: this event is almost always the echo of our own - // empty-file create, and re-fetching would replace the size and - // mtime of the write we just accepted with the stale revision's - // — making a file that was copied in seconds ago read as empty - // until its upload lands (offline.md Phase 3). - debug!(uid = %node_uid, "ignoring remote event for a node with a queued write"); - } else if let Some(&ino) = st.by_uid.get(node_uid) { - // Known node changed: drop its cached attrs/data (and listing if - // it is a directory) so the next access re-fetches. Its content - // blob may now be stale, so evict it too. - st.invalidate_listing(ino); - content.evict(node_uid); - let _ = notifier.inval_inode(INodeNo(ino), 0, 0); - } - // A create (or move-in) shows up as a change to the parent listing; - // drop it so the new child is picked up on the next readdir. - if let Some(parent_uid) = parent_node_uid - && let Some(&parent) = st.by_uid.get(parent_uid) - { - st.invalidate_listing(parent); - let _ = notifier.inval_inode(INodeNo(parent), 0, 0); - } - } - DriveEvent::NodeDeleted { node_uid, .. } => { - let mut st = state.lock(); - // Capture the inode before `forget` clears the uid mapping. - let child = st.by_uid.get(node_uid).copied(); - content.evict(node_uid); - if let Some((parent, name)) = st.forget(node_uid) { - match child { - Some(child) => { - let _ = notifier.delete(INodeNo(parent), INodeNo(child), OsStr::new(&name)); - } - None => { - let _ = notifier.inval_entry(INodeNo(parent), OsStr::new(&name)); - } - } - } - } - // Continuity or scope was lost: our cached listings may be arbitrarily - // stale, so drop every listing and tell the kernel to forget all - // metadata. Inodes stay stable; dirs simply re-enumerate on next access. - DriveEvent::ContinuityLost { .. } | DriveEvent::ScopeAccessLost { .. } => { - warn!("event continuity lost; dropping all cached listings, resyncing lazily"); - let mut st = state.lock(); - let dirs: Vec = st.children.keys().copied().collect(); - for &ino in &dirs { - st.invalidate_listing(ino); - let _ = notifier.inval_inode(INodeNo(ino), 0, 0); - } - } - // No substantive local change; the cursor advance is handled by the - // caller persisting the event id. - DriveEvent::CursorAdvanced { .. } | DriveEvent::SharedWithMeUpdated { .. } => {} - } -} - -/// Poll the remote event cursor forever, applying each batch to the shared -/// state. Resumes from the cursor persisted in the DB so changes made while -/// unmounted are applied; only a first-ever mount seeds from the server head. -/// The cursor is persisted after every batch. Runs as a Tokio task; returns -/// only on fatal error. -async fn run_event_sync( - client: ProtonDriveClient, - scope: DriveEventScopeId, - state: Arc>, - content: Arc, - db: Arc, - pending: Arc>>, - notifier: Notifier, -) { - let mut cursor: Option = match db.get_event_cursor() { - // Resume: pick up exactly where the last run left off. - Ok(Some(saved)) => Some(DriveEventId::from(saved)), - // First mount: a `None` cursor yields a single `CursorAdvanced` at the - // server head; persist it so the next restart resumes instead of - // reseeding (which would skip everything that changed offline). - // Seeding needs the network, and this task also runs on mounts that - // started offline (offline.md Phase 1) — so retry rather than giving up, - // which used to disable live sync for the life of the daemon. - Ok(None) => { - let mut delay = ONLINE_PROBE_MIN; - loop { - match client.enumerate_events(&scope, None).await { - Ok(events) => { - let head = events.last().map(|e| e.id().clone()); - if let Some(c) = &head - && let Err(e) = db.set_event_cursor(c.as_str()) - { - warn!(error = %e, "persist seed cursor failed"); - } - break head; - } - Err(e) => { - warn!(error = %e, ?delay, "seed event cursor failed; retrying"); - tokio::time::sleep(delay).await; - delay = (delay * 2).min(ONLINE_PROBE_MAX); - } - } - } - } - Err(e) => { - error!(error = %e, "read persisted cursor failed; live sync disabled"); - return; - } - }; - info!(?cursor, "event sync started"); - - loop { - tokio::time::sleep(POLL_INTERVAL).await; - let events = match client.enumerate_events(&scope, cursor.as_ref()).await { - Ok(events) => events, - Err(e) => { - warn!(error = %e, "event poll failed; retrying after interval"); - continue; - } - }; - if events.is_empty() { - continue; - } - debug!(count = events.len(), "applying remote events"); - for event in &events { - // Converge the SDK's own caches (folder keys, entity cache) on the - // server before applying the event to our tree. Without this, a node - // re-keyed/moved by another client keeps a stale key in the SDK for - // the life of the daemon (SDK plan #9). `apply_event` only touches - // our FUSE state, so nothing else does this. - if let Err(e) = client.invalidate_caches_for_event(event).await { - warn!(error = %e, "sdk cache invalidation for event failed"); - } - apply_event(&state, &content, &pending, ¬ifier, event); - } - cursor = events.last().map(|e| e.id().clone()); - if let Some(c) = &cursor - && let Err(e) = db.set_event_cursor(c.as_str()) - { - warn!(error = %e, "persist event cursor failed"); - } - } -} - -/// Keep the local-file index fresh for the launcher prompt's "This computer" -/// results. Rebuilds the index whenever it is older than [`LOCAL_INDEX_TTL`], -/// then sleeps; runs on its own thread for the life of the daemon. -/// -/// The walk is the one part of the daemon that touches the wider filesystem, so -/// it is deliberately kept off every hot path: it never runs on a FUSE or -/// control-socket thread, and it excludes the mountpoint (walking it would fault -/// every remote node in through FUSE, defeating on-demand hydration). -fn run_local_index( - db: Arc, - indexing: Arc, - transfers: Arc, - mountpoint: PathBuf, -) { - loop { - let age = db.local_indexed_at().ok().flatten(); - let stale = - age.is_none_or(|at| now_secs().saturating_sub(at) >= LOCAL_INDEX_TTL.as_secs() as i64); - if stale { - scan_local_once(&db, &indexing, &transfers, &mountpoint); - } - std::thread::sleep(LOCAL_INDEX_CHECK); - } -} - -/// Walk `$HOME` once and replace the local-file index with what it finds. -/// Batches stream straight into SQLite, so peak memory is one batch — not the -/// whole home directory. -/// -/// Reports itself as a job: the first scan after a fresh install walks the whole -/// home directory, and `indexing` alone only tells the launcher prompt to say -/// "still indexing" — nothing else showed that the daemon was busy. -fn scan_local_once( - db: &Db, - indexing: &AtomicBool, - transfers: &Arc, - mountpoint: &Path, -) { - let dirs = match AppDirs::new() { - Ok(d) => d, - Err(e) => { - warn!(error = %e, "local index: cannot resolve app dirs"); - return; - } - }; - let Some(home) = dirs.home_dir() else { - warn!("local index: cannot resolve home directory"); - return; - }; - let generation = match db.local_begin_scan() { - Ok(g) => g, - Err(e) => { - warn!(error = %e, "local index: cannot open scan generation"); - return; - } - }; - - let excludes = localindex::default_excludes(mountpoint, &dirs.state_dir(), &dirs.cache_dir()); - indexing.store(true, Ordering::Relaxed); - let started = Instant::now(); - - // The walk has no idea how many files it will find, so the job counts what it - // has seen and stays indeterminate. - let job = transfers.begin_job("Indexing this computer"); - job.detail("Scanning your files"); - let walked = localindex::scan(&[home], &excludes, |batch| { - if let Err(e) = db.local_upsert_batch(generation, &batch) { - warn!(error = %e, "local index: batch write failed"); - } - }); - - // Prune what this scan did not see and rebuild the FTS index over the rest, - // even if some batches failed — a partial index still beats none. - job.detail("Building the search index"); - match db.local_finish_scan(generation, now_secs()) { - Ok(indexed) => info!( - walked, - indexed, - elapsed_ms = started.elapsed().as_millis() as u64, - "local index rebuilt" - ), - Err(e) => warn!(error = %e, "local index: finish failed"), - } - indexing.store(false, Ordering::Relaxed); -} - -/// Why a [`mount`] call returned. Lets the daemon decide whether to exit (clean -/// shutdown) or remount (the mount went away under it). -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -pub enum MountOutcome { - /// The daemon was asked to stop (SIGTERM/SIGINT) and we lazily unmounted - /// ourselves. The caller should exit cleanly. - Shutdown, - /// The kernel mount ended on its own (e.g. an external `fusermount -u`). - /// The caller may want to remount. - Unmounted, -} - -/// Whether `path` is a mountpoint whose FUSE connection is dead — the state a -/// daemon killed before it could unmount leaves behind. -/// -/// The kernel answers every operation on such a path with `ENOTCONN`, which the -/// ordinary existence checks (`is_dir`, `exists`) report as plain `false`: the -/// path looks *absent* rather than broken. Callers that would otherwise treat -/// that as "nothing to mount here" use this to tell the two apart. -pub(crate) fn is_stale_mount(path: &Path) -> bool { - matches!( - std::fs::metadata(path).map_err(|e| e.raw_os_error()), - Err(Some(libc::ENOTCONN)) - ) -} - -/// A secondary (on-demand sync folder) FUSE session, paired with its FUSE -/// connection id so teardown can abort a mid-transfer connection rather than -/// block on it. See [`Core::mounts`] and [`umount_session_unblocked`]. -type SecondaryMount = (BackgroundSession, Option); - -/// The kernel's id for the FUSE connection backing the mount at `mountpoint` — -/// the directory name under `/sys/fs/fuse/connections`, which is the minor -/// number of the mountpoint's device. Must be read *while still mounted*; after -/// unmount the path resolves to the underlying directory on another device. -/// `None` when the path can't be stat'd. -fn fuse_connection_id(mountpoint: &Path) -> Option { - use std::os::unix::fs::MetadataExt; - let dev = std::fs::metadata(mountpoint).ok()?.dev(); - Some(libc::minor(dev)) -} - -/// Force the kernel to abort FUSE connection `id`, erroring every in-flight -/// request. -/// -/// On a stop signal we unmount lazily (`MNT_DETACH`) so the call succeeds even -/// mid-transfer — but detach only removes the mountpoint; it does *not* end the -/// connection while a request is still in flight. fuser's session loop then -/// blocks on `/dev/fuse` waiting for a next request that never comes, so `join` -/// hangs — long enough during a transfer that systemd's stop timeout SIGKILLs -/// the daemon mid-unmount, stranding the on-demand mounts as dead endpoints. -/// Writing the connection's `abort` file makes the pending reads fail with -/// `ENODEV`, so the loop returns and `join` completes at once. Best-effort: -/// there is nothing more to do on the shutdown path if it fails. -fn abort_fuse_connection(id: u32) { - let path = format!("/sys/fs/fuse/connections/{id}/abort"); - if std::fs::write(&path, b"1").is_ok() { - info!(id, "aborted FUSE connection to unblock unmount"); - } -} - -/// Unmount a background session that may be mid-transfer without wedging: abort -/// its connection first (so the session loop exits promptly), then lazily -/// unmount and join. `conn` is the id captured at mount time; `None` skips the -/// abort and just unmounts (a healthy idle mount joins on its own `Destroy`). -fn umount_session_unblocked(session: BackgroundSession, conn: Option) -> std::io::Result<()> { - if let Some(id) = conn { - abort_fuse_connection(id); - } - session.umount_and_join() -} - -/// Best-effort teardown of a stale mount left behind by a crashed daemon. A -/// previous run that died without unmounting leaves the kernel mount in place, -/// so the fresh `Session::new` below would fail with EBUSY ("Device or resource -/// busy"). `fusermount3 -u -z` is the lazy (detach) unmount, which succeeds even -/// when the old mount is still busy. Swallow all output/errors: if there is no -/// stale mount this is simply a no-op. -fn clear_stale_mount(mountpoint: &Path) { - for bin in ["fusermount3", "fusermount"] { - let ok = std::process::Command::new(bin) - .arg("-u") - .arg("-z") - .arg(mountpoint) - .stdout(std::process::Stdio::null()) - .stderr(std::process::Stdio::null()) - .status() - .map(|s| s.success()) - .unwrap_or(false); - if ok { - info!(mountpoint = %mountpoint.display(), "cleared stale mount before remount"); - return; - } - } -} - -/// `sync_state` key holding the uid of the My Files root, so a later run can -/// recover it from `nodes` without the API (offline.md Phase 1). -const ROOT_UID_KEY: &str = "root_uid"; - -/// The My Files root, and whether we got it from the API (`true`) or from the -/// cache because the API was unreachable (`false`). -/// -/// A successful fetch also records the root's uid, which is what makes the -/// fallback possible on a later run: `nodes` is keyed by uid, so without this we -/// would have the root's row on disk and no way to tell which one it is. -/// -/// Failing to fetch is only fatal on a genuinely cold start — no cached root -/// means an empty tree, and mounting that would show the user an empty Drive -/// rather than an honest error. -fn fetch_or_recall_root( - client: &ProtonDriveClient, - rt: &tokio::runtime::Handle, - db: &Db, -) -> std::io::Result<(Node, bool)> { - let err = match rt.block_on(client.get_my_files_folder()) { - Ok(root) => { - if let Err(e) = db.set_state_str(ROOT_UID_KEY, &root.uid.to_string()) { - warn!(error = %e, "persist root uid failed"); - } - return Ok((root, true)); - } - Err(e) => e, - }; - let cached = db - .state_str(ROOT_UID_KEY) - .ok() - .flatten() - .and_then(|uid| db.node_by_uid(&uid).ok().flatten()); - match cached { - Some(root) => { - warn!(error = %err, "fetch My Files root failed; mounting from cache (offline)"); - Ok((root, false)) - } - None => Err(std::io::Error::other(format!("fetch My Files root: {err}"))), - } -} - -/// Mount the filesystem at `mountpoint` and block until it is unmounted or the -/// daemon is asked to stop. -/// -/// Resolves the My Files root up front — from the API, or from the cached tree -/// when the API is unreachable ([`fetch_or_recall_root`]) — then spawns the -/// Phase 2 event-sync task, the -/// Phase 4 control socket, and runs the FUSE session on its own thread while -/// this thread waits for either a stop signal (SIGTERM/SIGINT) or the kernel -/// mount ending on its own. On a stop signal we lazily unmount ourselves -/// (`umount_and_join`, the MNT_DETACH path that succeeds even while downloads -/// are in flight), so `systemctl --user stop` is always a clean teardown. -/// `rt` must be a handle to a *running* multi-threaded runtime. -pub fn mount( - client: ProtonDriveClient, - rt: tokio::runtime::Handle, - mountpoint: &Path, - cache: ContentCache, - control_socket: &Path, - db: Arc, - username: String, -) -> std::io::Result { - let (root, online) = fetch_or_recall_root(&client, &rt, &db)?; - let scope = root.tree_event_scope_id(); - - // The folder-sync engine (devices.md Phase 2) runs on its own thread and is - // nudged over this channel; the sender lives in Core so control-socket - // handlers can trigger reconciles. - let (sync_tx, sync_rx) = std::sync::mpsc::channel::(); - - let core = Core { - client: client.clone(), - rt: rt.clone(), - state: Arc::new(Mutex::new(State { - entries: HashMap::new(), - by_uid: HashMap::new(), - children: HashMap::new(), - next_ino: 2, - active_writes: HashMap::new(), - handles: HashMap::new(), - next_fh: 1, - db: db.clone(), - })), - cache: Arc::new(cache), - readers: Arc::new(Mutex::new(HashMap::new())), - stream_ring: Arc::new(Mutex::new(StreamRing::default())), - workers: Arc::new(Workers::new(FUSE_WORKERS)?), - db, - online: Arc::new(AtomicBool::new(online)), - pending: Arc::new(Mutex::new(HashMap::new())), - drain_wake: Arc::new((Mutex::new(false), Condvar::new())), - timeline_refreshing: Arc::new(AtomicBool::new(false)), - trash_refreshing: Arc::new(AtomicBool::new(false)), - thumb_gen: Arc::new(Mutex::new(HashSet::new())), - no_thumbnail: Arc::new(Mutex::new(HashMap::new())), - size_upgrades: Arc::new(Mutex::new(HashMap::new())), - notifier: Arc::new(OnceLock::new()), - transfers: TransferRegistry::new(), - indexing: Arc::new(AtomicBool::new(false)), - sync_progress: Arc::new(Mutex::new(HashMap::new())), - sync_tx, - mounts: Arc::new(Mutex::new(HashMap::new())), - sync_locks: Arc::new(Mutex::new(HashMap::new())), - }; - - // Writes queued by a previous run (or left behind by a crash) are still owed - // an upload, and reads must be served from their staged blobs until they land. - core.hydrate_pending(); - // Then the writes that were fsynced but never closed, which the cache moved - // aside at open. After `hydrate_pending`, so a recovered partial write can - // see an already-queued write to the same node. - core.recover_fsynced_writes(); - { - let core = core.clone(); - std::thread::Builder::new() - .name("pdfs-drain".into()) - .spawn(move || core.run_pending_drain())?; - } - - // Start the folder-sync engine. It watches every mirror folder, polls the - // remotes, and reconciles on its own thread — never in front of a FUSE call. - sync::spawn(core.clone(), sync_rx); - - // Mounted from the cache: watch for the network coming back so the mount can - // stop being read-only-ish without the user restarting the daemon. - if !online { - let core = core.clone(); - std::thread::Builder::new() - .name("pdfs-online-probe".into()) - .spawn(move || core.run_online_probe())?; - } - - // Re-establish on-demand mounts left over from a previous run (devices.md - // Phase 4). On its own thread: each remount fetches a remote node, and we - // must not block the main mount below on the network. - { - let core = core.clone(); - std::thread::Builder::new() - .name("pdfs-restore-ondemand".into()) - .spawn(move || core.restore_ondemand_mounts())?; - } - - // Keep the launcher prompt's "This computer" index fresh. Its own thread: - // the walk is I/O-heavy and must never sit in front of a FUSE callback. - { - let db = core.db.clone(); - let indexing = core.indexing.clone(); - let transfers = core.transfers.clone(); - let mountpoint = mountpoint.to_path_buf(); - std::thread::Builder::new() - .name("pdfs-localindex".into()) - .spawn(move || run_local_index(db, indexing, transfers, mountpoint))?; - } - - let mut config = Config::default(); - config.mount_options = vec![ - MountOption::FSName("protondrive".to_string()), - MountOption::Subtype("protondrive".to_string()), - MountOption::DefaultPermissions, - ]; - - // A crashed previous run can leave the kernel mount in place, which makes - // the fresh mount below fail with EBUSY. Lazily detach any leftover first. - clear_stale_mount(mountpoint); - info!(mountpoint = %mountpoint.display(), "mounting Proton Drive"); - - // Bind the control socket before the FUSE session takes over the thread. A - // stale socket file from a previous run would block the bind, so clear it. - let _ = std::fs::remove_file(control_socket); - let old_umask = unsafe { libc::umask(0o77) }; - let listener_res = UnixListener::bind(control_socket); - unsafe { libc::umask(old_umask) }; - let listener = listener_res?; - // Owner-only before anything can connect: a peer on this socket commands the - // daemon's authenticated session without a credential of its own (B6). - if let Err(e) = pdfs_core::config::restrict_socket(control_socket) { - error!(error = %e, "cannot restrict control socket permissions; refusing to serve"); - let _ = std::fs::remove_file(control_socket); - return Err(std::io::Error::other(format!( - "control socket permissions: {e}" - ))); - } - { - let core = core.clone(); - let username = username.clone(); - let mountpoint = mountpoint.to_path_buf(); - std::thread::Builder::new() - .name("pdfs-control".into()) - .spawn(move || run_control_socket(core, username, mountpoint, listener))?; - } - - let fs = ProtonFs::new(core.clone(), root); - // Warm the in-memory maps from the DB so a cold start serves previously - // seen metadata without re-hitting the API (plan.md P1). - core.hydrate(); - - // Build the session explicitly (not `mount2`) so we can grab a `Notifier` - // for the event task. `spawn` runs the session loop on its own background - // thread; we then wait here for either a stop signal or the mount ending. - let bg = Session::new(fs, mountpoint, &config)?.spawn()?; - // The connection id, captured now while the mount is live, so a stop signal - // mid-transfer can abort it rather than block `join` (see `abort_fuse_connection`). - let main_conn = fuse_connection_id(mountpoint); - let notifier = bg.notifier(); - // Same channel, kept on the `Core` so background work (a size upgrade, say) - // can invalidate kernel-cached metadata without threading a handle through. - let _ = core.notifier.set(notifier.clone()); - rt.spawn(run_event_sync( - client, - scope, - core.state, - core.cache, - core.db, - core.pending, - notifier, - )); - - // Stop signals (SIGTERM from `systemctl --user stop`, SIGINT from Ctrl-C) - // are delivered onto the async runtime; bridge them onto a sync channel so - // the loop below can react without blocking a worker thread. A bounded - // channel of 1 is enough — we only need to know that *a* stop arrived. - let (sig_tx, sig_rx) = std::sync::mpsc::sync_channel::<()>(1); - rt.spawn(async move { - let mut sigterm = - match tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate()) { - Ok(s) => s, - Err(e) => { - warn!(error = %e, "install SIGTERM handler failed"); - return; - } - }; - let mut sigint = - match tokio::signal::unix::signal(tokio::signal::unix::SignalKind::interrupt()) { - Ok(s) => s, - Err(e) => { - warn!(error = %e, "install SIGINT handler failed"); - return; - } - }; - tokio::select! { - _ = sigterm.recv() => info!("received SIGTERM"), - _ = sigint.recv() => info!("received SIGINT"), - } - let _ = sig_tx.try_send(()); - }); - - // Wait for whichever happens first: a stop signal (→ we unmount ourselves - // via the lazy MNT_DETACH path, clean even mid-download), or the kernel - // mount ending on its own (→ the session thread finishes). Poll instead of - // blocking on `join` so we can also notice the signal. - let outcome = loop { - match sig_rx.recv_timeout(Duration::from_millis(500)) { - Ok(()) => { - info!("stop requested; unmounting"); - if let Err(e) = umount_session_unblocked(bg, main_conn) { - warn!(error = %e, "umount_and_join failed"); - } - break MountOutcome::Shutdown; - } - Err(std::sync::mpsc::RecvTimeoutError::Timeout) => { - if bg.guard.is_finished() { - info!("mount ended externally"); - if let Err(e) = bg.join() { - warn!(error = %e, "session join failed"); - } - break MountOutcome::Unmounted; - } - } - Err(std::sync::mpsc::RecvTimeoutError::Disconnected) => { - // Signal task gone (failed to install); fall back to join. - let _ = bg.join(); - break MountOutcome::Unmounted; - } - } - }; - - // Unmount every on-demand sync folder too, or the kernel mounts linger as - // stale and the next start fails with EBUSY (devices.md Phase 3). - let secondaries: Vec<_> = core.mounts.lock().drain().collect(); - for (id, (session, conn)) in secondaries { - if let Err(e) = umount_session_unblocked(session, conn) { - warn!(id, error = %e, "unmount on-demand folder failed"); - } - } - - let _ = std::fs::remove_file(control_socket); - Ok(outcome) -} - #[cfg(test)] mod size_upgrade_tests { use super::SizeUpgrade; diff --git a/crates/pdfs-fuse/src/mount.rs b/crates/pdfs-fuse/src/mount.rs new file mode 100644 index 0000000..7494a44 --- /dev/null +++ b/crates/pdfs-fuse/src/mount.rs @@ -0,0 +1,392 @@ +//! Mount lifecycle and background-service orchestration. + +use super::*; + +/// Why a [`mount`] call returned. Lets the daemon decide whether to exit (clean +/// shutdown) or remount (the mount went away under it). +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum MountOutcome { + /// The daemon was asked to stop (SIGTERM/SIGINT) and we lazily unmounted + /// ourselves. The caller should exit cleanly. + Shutdown, + /// The kernel mount ended on its own (e.g. an external `fusermount -u`). + /// The caller may want to remount. + Unmounted, +} + +/// Whether `path` is a mountpoint whose FUSE connection is dead — the state a +/// daemon killed before it could unmount leaves behind. +/// +/// The kernel answers every operation on such a path with `ENOTCONN`, which the +/// ordinary existence checks (`is_dir`, `exists`) report as plain `false`: the +/// path looks *absent* rather than broken. Callers that would otherwise treat +/// that as "nothing to mount here" use this to tell the two apart. +pub(crate) fn is_stale_mount(path: &Path) -> bool { + matches!( + std::fs::metadata(path).map_err(|e| e.raw_os_error()), + Err(Some(libc::ENOTCONN)) + ) +} + +/// A secondary (on-demand sync folder) FUSE session, paired with its FUSE +/// connection id so teardown can abort a mid-transfer connection rather than +/// block on it. See [`Core::mounts`] and [`umount_session_unblocked`]. +pub(super) type SecondaryMount = (BackgroundSession, Option); + +/// The kernel's id for the FUSE connection backing the mount at `mountpoint` — +/// the directory name under `/sys/fs/fuse/connections`, which is the minor +/// number of the mountpoint's device. Must be read *while still mounted*; after +/// unmount the path resolves to the underlying directory on another device. +/// `None` when the path can't be stat'd. +pub(super) fn fuse_connection_id(mountpoint: &Path) -> Option { + use std::os::unix::fs::MetadataExt; + let dev = std::fs::metadata(mountpoint).ok()?.dev(); + Some(libc::minor(dev)) +} + +/// Force the kernel to abort FUSE connection `id`, erroring every in-flight +/// request. +/// +/// On a stop signal we unmount lazily (`MNT_DETACH`) so the call succeeds even +/// mid-transfer — but detach only removes the mountpoint; it does *not* end the +/// connection while a request is still in flight. fuser's session loop then +/// blocks on `/dev/fuse` waiting for a next request that never comes, so `join` +/// hangs — long enough during a transfer that systemd's stop timeout SIGKILLs +/// the daemon mid-unmount, stranding the on-demand mounts as dead endpoints. +/// Writing the connection's `abort` file makes the pending reads fail with +/// `ENODEV`, so the loop returns and `join` completes at once. Best-effort: +/// there is nothing more to do on the shutdown path if it fails. +pub(super) fn abort_fuse_connection(id: u32) { + let path = format!("/sys/fs/fuse/connections/{id}/abort"); + if std::fs::write(&path, b"1").is_ok() { + info!(id, "aborted FUSE connection to unblock unmount"); + } +} + +/// Unmount a background session that may be mid-transfer without wedging: abort +/// its connection first (so the session loop exits promptly), then lazily +/// unmount and join. `conn` is the id captured at mount time; `None` skips the +/// abort and just unmounts (a healthy idle mount joins on its own `Destroy`). +pub(super) fn umount_session_unblocked( + session: BackgroundSession, + conn: Option, +) -> std::io::Result<()> { + if let Some(id) = conn { + abort_fuse_connection(id); + } + session.umount_and_join() +} + +/// Best-effort teardown of a stale mount left behind by a crashed daemon. A +/// previous run that died without unmounting leaves the kernel mount in place, +/// so the fresh `Session::new` below would fail with EBUSY ("Device or resource +/// busy"). `fusermount3 -u -z` is the lazy (detach) unmount, which succeeds even +/// when the old mount is still busy. Swallow all output/errors: if there is no +/// stale mount this is simply a no-op. +pub(super) fn clear_stale_mount(mountpoint: &Path) { + for bin in ["fusermount3", "fusermount"] { + let ok = std::process::Command::new(bin) + .arg("-u") + .arg("-z") + .arg(mountpoint) + .stdout(std::process::Stdio::null()) + .stderr(std::process::Stdio::null()) + .status() + .map(|s| s.success()) + .unwrap_or(false); + if ok { + info!(mountpoint = %mountpoint.display(), "cleared stale mount before remount"); + return; + } + } +} + +/// `sync_state` key holding the uid of the My Files root, so a later run can +/// recover it from `nodes` without the API (offline.md Phase 1). +const ROOT_UID_KEY: &str = "root_uid"; + +/// The My Files root, and whether we got it from the API (`true`) or from the +/// cache because the API was unreachable (`false`). +/// +/// A successful fetch also records the root's uid, which is what makes the +/// fallback possible on a later run: `nodes` is keyed by uid, so without this we +/// would have the root's row on disk and no way to tell which one it is. +/// +/// Failing to fetch is only fatal on a genuinely cold start — no cached root +/// means an empty tree, and mounting that would show the user an empty Drive +/// rather than an honest error. +fn fetch_or_recall_root( + client: &ProtonDriveClient, + rt: &tokio::runtime::Handle, + db: &Db, +) -> std::io::Result<(Node, bool)> { + let err = match rt.block_on(client.get_my_files_folder()) { + Ok(root) => { + if let Err(e) = db.set_state_str(ROOT_UID_KEY, &root.uid.to_string()) { + warn!(error = %e, "persist root uid failed"); + } + return Ok((root, true)); + } + Err(e) => e, + }; + let cached = db + .state_str(ROOT_UID_KEY) + .ok() + .flatten() + .and_then(|uid| db.node_by_uid(&uid).ok().flatten()); + match cached { + Some(root) => { + warn!(error = %err, "fetch My Files root failed; mounting from cache (offline)"); + Ok((root, false)) + } + None => Err(std::io::Error::other(format!("fetch My Files root: {err}"))), + } +} + +/// Mount the filesystem at `mountpoint` and block until it is unmounted or the +/// daemon is asked to stop. +/// +/// Resolves the My Files root up front — from the API, or from the cached tree +/// when the API is unreachable (`fetch_or_recall_root`) — then spawns the +/// Phase 2 event-sync task, the +/// Phase 4 control socket, and runs the FUSE session on its own thread while +/// this thread waits for either a stop signal (SIGTERM/SIGINT) or the kernel +/// mount ending on its own. On a stop signal we lazily unmount ourselves +/// (`umount_and_join`, the MNT_DETACH path that succeeds even while downloads +/// are in flight), so `systemctl --user stop` is always a clean teardown. +/// `rt` must be a handle to a *running* multi-threaded runtime. +pub fn mount( + client: ProtonDriveClient, + rt: tokio::runtime::Handle, + mountpoint: &Path, + cache: ContentCache, + control_socket: &Path, + db: Arc, + username: String, +) -> std::io::Result { + let (root, online) = fetch_or_recall_root(&client, &rt, &db)?; + let scope = root.tree_event_scope_id(); + + // The folder-sync engine (devices.md Phase 2) runs on its own thread and is + // nudged over this channel; the sender lives in Core so control-socket + // handlers can trigger reconciles. + let (sync_tx, sync_rx) = std::sync::mpsc::channel::(); + + let core = Core { + client: client.clone(), + rt: rt.clone(), + state: Arc::new(Mutex::new(State { + entries: HashMap::new(), + by_uid: HashMap::new(), + children: HashMap::new(), + next_ino: 2, + active_writes: HashMap::new(), + handles: HashMap::new(), + next_fh: 1, + db: db.clone(), + })), + cache: Arc::new(cache), + readers: Arc::new(Mutex::new(HashMap::new())), + stream_ring: Arc::new(Mutex::new(StreamRing::default())), + workers: Arc::new(Workers::new(FUSE_WORKERS)?), + db, + online: Arc::new(AtomicBool::new(online)), + pending: Arc::new(Mutex::new(HashMap::new())), + drain_wake: Arc::new((Mutex::new(false), Condvar::new())), + timeline_refreshing: Arc::new(AtomicBool::new(false)), + trash_refreshing: Arc::new(AtomicBool::new(false)), + thumb_gen: Arc::new(Mutex::new(HashSet::new())), + no_thumbnail: Arc::new(Mutex::new(HashMap::new())), + size_upgrades: Arc::new(Mutex::new(HashMap::new())), + notifier: Arc::new(OnceLock::new()), + transfers: TransferRegistry::new(), + indexing: Arc::new(AtomicBool::new(false)), + sync_progress: Arc::new(Mutex::new(HashMap::new())), + sync_tx, + mounts: Arc::new(Mutex::new(HashMap::new())), + sync_locks: Arc::new(Mutex::new(HashMap::new())), + }; + + // Writes queued by a previous run (or left behind by a crash) are still owed + // an upload, and reads must be served from their staged blobs until they land. + core.hydrate_pending(); + // Then the writes that were fsynced but never closed, which the cache moved + // aside at open. After `hydrate_pending`, so a recovered partial write can + // see an already-queued write to the same node. + core.recover_fsynced_writes(); + { + let core = core.clone(); + std::thread::Builder::new() + .name("pdfs-drain".into()) + .spawn(move || core.run_pending_drain())?; + } + + // Start the folder-sync engine. It watches every mirror folder, polls the + // remotes, and reconciles on its own thread — never in front of a FUSE call. + sync::spawn(core.clone(), sync_rx); + + // Mounted from the cache: watch for the network coming back so the mount can + // stop being read-only-ish without the user restarting the daemon. + if !online { + let core = core.clone(); + std::thread::Builder::new() + .name("pdfs-online-probe".into()) + .spawn(move || core.run_online_probe())?; + } + + // Re-establish on-demand mounts left over from a previous run (devices.md + // Phase 4). On its own thread: each remount fetches a remote node, and we + // must not block the main mount below on the network. + { + let core = core.clone(); + std::thread::Builder::new() + .name("pdfs-restore-ondemand".into()) + .spawn(move || core.restore_ondemand_mounts())?; + } + + // Keep the launcher prompt's "This computer" index fresh. Its own thread: + // the walk is I/O-heavy and must never sit in front of a FUSE callback. + { + let db = core.db.clone(); + let indexing = core.indexing.clone(); + let transfers = core.transfers.clone(); + let mountpoint = mountpoint.to_path_buf(); + std::thread::Builder::new() + .name("pdfs-localindex".into()) + .spawn(move || run_local_index(db, indexing, transfers, mountpoint))?; + } + + let mut config = Config::default(); + config.mount_options = vec![ + MountOption::FSName("protondrive".to_string()), + MountOption::Subtype("protondrive".to_string()), + MountOption::DefaultPermissions, + ]; + + // A crashed previous run can leave the kernel mount in place, which makes + // the fresh mount below fail with EBUSY. Lazily detach any leftover first. + clear_stale_mount(mountpoint); + info!(mountpoint = %mountpoint.display(), "mounting Proton Drive"); + + // Bind the control socket before the FUSE session takes over the thread. A + // stale socket file from a previous run would block the bind, so clear it. + let _ = std::fs::remove_file(control_socket); + let old_umask = unsafe { libc::umask(0o77) }; + let listener_res = UnixListener::bind(control_socket); + unsafe { libc::umask(old_umask) }; + let listener = listener_res?; + // Owner-only before anything can connect: a peer on this socket commands the + // daemon's authenticated session without a credential of its own (B6). + if let Err(e) = pdfs_core::config::restrict_socket(control_socket) { + error!(error = %e, "cannot restrict control socket permissions; refusing to serve"); + let _ = std::fs::remove_file(control_socket); + return Err(std::io::Error::other(format!( + "control socket permissions: {e}" + ))); + } + { + let core = core.clone(); + let username = username.clone(); + let mountpoint = mountpoint.to_path_buf(); + std::thread::Builder::new() + .name("pdfs-control".into()) + .spawn(move || run_control_socket(core, username, mountpoint, listener))?; + } + + let fs = ProtonFs::new(core.clone(), root); + // Warm the in-memory maps from the DB so a cold start serves previously + // seen metadata without re-hitting the API (plan.md P1). + core.hydrate(); + + // Build the session explicitly (not `mount2`) so we can grab a `Notifier` + // for the event task. `spawn` runs the session loop on its own background + // thread; we then wait here for either a stop signal or the mount ending. + let bg = Session::new(fs, mountpoint, &config)?.spawn()?; + // The connection id, captured now while the mount is live, so a stop signal + // mid-transfer can abort it rather than block `join` (see `abort_fuse_connection`). + let main_conn = fuse_connection_id(mountpoint); + let notifier = bg.notifier(); + // Same channel, kept on the `Core` so background work (a size upgrade, say) + // can invalidate kernel-cached metadata without threading a handle through. + let _ = core.notifier.set(notifier.clone()); + rt.spawn(run_event_sync( + client, + scope, + core.state, + core.cache, + core.db, + core.pending, + notifier, + )); + + // Stop signals (SIGTERM from `systemctl --user stop`, SIGINT from Ctrl-C) + // are delivered onto the async runtime; bridge them onto a sync channel so + // the loop below can react without blocking a worker thread. A bounded + // channel of 1 is enough — we only need to know that *a* stop arrived. + let (sig_tx, sig_rx) = std::sync::mpsc::sync_channel::<()>(1); + rt.spawn(async move { + let mut sigterm = + match tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate()) { + Ok(s) => s, + Err(e) => { + warn!(error = %e, "install SIGTERM handler failed"); + return; + } + }; + let mut sigint = + match tokio::signal::unix::signal(tokio::signal::unix::SignalKind::interrupt()) { + Ok(s) => s, + Err(e) => { + warn!(error = %e, "install SIGINT handler failed"); + return; + } + }; + tokio::select! { + _ = sigterm.recv() => info!("received SIGTERM"), + _ = sigint.recv() => info!("received SIGINT"), + } + let _ = sig_tx.try_send(()); + }); + + // Wait for whichever happens first: a stop signal (→ we unmount ourselves + // via the lazy MNT_DETACH path, clean even mid-download), or the kernel + // mount ending on its own (→ the session thread finishes). Poll instead of + // blocking on `join` so we can also notice the signal. + let outcome = loop { + match sig_rx.recv_timeout(Duration::from_millis(500)) { + Ok(()) => { + info!("stop requested; unmounting"); + if let Err(e) = umount_session_unblocked(bg, main_conn) { + warn!(error = %e, "umount_and_join failed"); + } + break MountOutcome::Shutdown; + } + Err(std::sync::mpsc::RecvTimeoutError::Timeout) => { + if bg.guard.is_finished() { + info!("mount ended externally"); + if let Err(e) = bg.join() { + warn!(error = %e, "session join failed"); + } + break MountOutcome::Unmounted; + } + } + Err(std::sync::mpsc::RecvTimeoutError::Disconnected) => { + // Signal task gone (failed to install); fall back to join. + let _ = bg.join(); + break MountOutcome::Unmounted; + } + } + }; + + // Unmount every on-demand sync folder too, or the kernel mounts linger as + // stale and the next start fails with EBUSY (devices.md Phase 3). + let secondaries: Vec<_> = core.mounts.lock().drain().collect(); + for (id, (session, conn)) in secondaries { + if let Err(e) = umount_session_unblocked(session, conn) { + warn!(id, error = %e, "unmount on-demand folder failed"); + } + } + + let _ = std::fs::remove_file(control_socket); + Ok(outcome) +}