diff --git a/crates/pdfs-fuse/src/background.rs b/crates/pdfs-fuse/src/background.rs index 2414691..7c1835a 100644 --- a/crates/pdfs-fuse/src/background.rs +++ b/crates/pdfs-fuse/src/background.rs @@ -2,19 +2,48 @@ use super::*; +/// Drop `uid` from one mount's tree and tell that mount's kernel session the +/// entry is gone. Shared by the trash and delete paths, which differ only in +/// which event carried them. +/// +/// Returns whether this mount actually held the node, so the caller can evict +/// its content blob once if *any* mount did. +fn forget_and_notify(st: &mut State, notifier: Option<&Notifier>, uid: &NodeUid) -> bool { + // Capture the inode before `forget` clears the uid mapping. + let child = st.by_uid.get(uid).copied(); + let Some((parent, name)) = st.forget(uid) else { + return false; + }; + if let Some(notifier) = notifier { + match child { + Some(child) => { + let _ = notifier.delete(INodeNo(parent), INodeNo(child), OsStr::new(&name)); + } + None => { + let _ = notifier.inval_entry(INodeNo(parent), OsStr::new(&name)); + } + } + } + true +} + /// 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, -) { +/// +/// Applied to **every** mounted inode space, not just the primary one. This task +/// is per-daemon while node state is per-mount, and a `DriveEvent` names a uid, +/// not a mount — so a file trashed from another device has to be withdrawn from +/// whichever of our sessions is showing it, which for anything under a sync +/// folder is a [`Core::fork_state`] fork rather than `core.state`. Reaching for +/// the primary state alone left forks serving a deleted file indefinitely, the +/// read-side twin of `docs/BUGS.md` B74. Inode numbers are per-mount, so each +/// mount must be notified through its **own** channel; that pairing is what +/// [`Core::for_each_mount`] exists to preserve. +fn apply_event(core: &Core, event: &DriveEvent) { match event { DriveEvent::NodeUpdated { node_uid, @@ -22,74 +51,69 @@ fn apply_event( 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)); - } + // 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). The parent listing is still refreshed + // below; it is only the node itself that is ours to define. + let ours = !*is_trashed && core.pending.lock().contains_key(node_uid); + if ours { + debug!(uid = %node_uid, "ignoring remote event for a node with a queued write"); + } + let mut had_node = false; + core.for_each_mount(|st, notifier| { + if *is_trashed { + // Trashing makes a node vanish from its parent listing. + had_node |= forget_and_notify(st, notifier, node_uid); + } else if !ours && let Some(&ino) = st.by_uid.get(node_uid) { + // Known node changed: drop its cached attrs/data (and + // listing if it is a directory) so the next access + // re-fetches. Its content blob may now be stale too. + had_node = true; + st.invalidate_listing(ino); + if let Some(notifier) = notifier { + let _ = notifier.inval_inode(INodeNo(ino), 0, 0); } } - } 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); + // 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); + if let Some(notifier) = notifier { + let _ = notifier.inval_inode(INodeNo(parent), 0, 0); + } + } + }); + // Once, not once per mount: the content cache is per-daemon. + if had_node { + core.cache.evict(node_uid); } } 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)); - } - } - } + core.cache.evict(node_uid); + core.for_each_mount(|st, notifier| { + forget_and_notify(st, notifier, node_uid); + }); } // 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); - } + core.for_each_mount(|st, notifier| { + let dirs: Vec = st.children.keys().copied().collect(); + for &ino in &dirs { + st.invalidate_listing(ino); + if let Some(notifier) = notifier { + let _ = notifier.inval_inode(INodeNo(ino), 0, 0); + } + } + }); } // No substantive local change; the cursor advance is handled by the // caller persisting the event id. @@ -102,15 +126,16 @@ fn apply_event( /// 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. +/// +/// Takes the whole `Core` rather than the primary mount's pieces: an event has +/// to be applied to every mounted inode space, and the registry that enumerates +/// them lives on the `Core` (see [`apply_event`]). pub(super) async fn run_event_sync( client: ProtonDriveClient, scope: DriveEventScopeId, - state: Arc>, - content: Arc, - db: Arc, - pending: Arc>>, - notifier: Notifier, + core: Core, ) { + let db = core.db.clone(); 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)), @@ -175,7 +200,7 @@ pub(super) async fn run_event_sync( warn!(error = %e, "sdk cache invalidation for event failed; retaining cursor for retry"); break; } - apply_event(&state, &content, &pending, ¬ifier, event); + apply_event(&core, event); let applied = event.id().clone(); if let Err(e) = db.set_event_cursor(applied.as_str()) { warn!(error = %e, "persist event cursor failed; retaining prior cursor for retry"); diff --git a/crates/pdfs-fuse/src/devices.rs b/crates/pdfs-fuse/src/devices.rs index 1195027..5754e2f 100644 --- a/crates/pdfs-fuse/src/devices.rs +++ b/crates/pdfs-fuse/src/devices.rs @@ -505,6 +505,11 @@ impl Core { fork.notifier = Arc::new(OnceLock::new()); // Size upgrades are keyed by inode, which is per-fork too. fork.size_upgrades = Arc::new(Mutex::new(HashMap::new())); + // The daemon's single drain thread belongs to the primary mount, but it + // drains ops for nodes that live *here*. Publishing this inode space is + // what lets it find them; without it a queued create's node keeps its + // `local~` uid after the upload lands and reads it as empty (B74). + fork.register_state(); fork } diff --git a/crates/pdfs-fuse/src/drain.rs b/crates/pdfs-fuse/src/drain.rs index dfd94dd..069b6b4 100644 --- a/crates/pdfs-fuse/src/drain.rs +++ b/crates/pdfs-fuse/src/drain.rs @@ -263,18 +263,19 @@ impl Core { /// it away from the one the user asked for. Best effort: the event sync /// would correct the tree anyway, but not before the user has looked at it. pub(crate) fn adopt_drained_name(&self, uid: &NodeUid, name: &str) { - let mut st = self.state.lock(); - let Some(&ino) = st.by_uid.get(uid) else { - return; - }; - let Some(entry) = st.entries.get_mut(&ino) else { - return; - }; - entry.node.name = name.to_string(); - let node = entry.node.clone(); - if let Err(e) = st.db.upsert_node(&node) { - warn!(%uid, error = %e, "db upsert_node failed after a conflict rename"); - } + self.for_each_state(|st| { + let Some(&ino) = st.by_uid.get(uid) else { + return; + }; + let Some(entry) = st.entries.get_mut(&ino) else { + return; + }; + entry.node.name = name.to_string(); + let node = entry.node.clone(); + if let Err(e) = st.db.upsert_node(&node) { + warn!(%uid, error = %e, "db upsert_node failed after a conflict rename"); + } + }); } /// Make a node that so far exists only on this machine real, and adopt the @@ -419,6 +420,14 @@ impl Core { /// /// The inode is deliberately kept, so anything already holding the file open /// keeps working across the drain. + /// + /// Applied to **every** live inode space, not just this Core's. A node + /// created inside an on-demand sync folder lives in that fork's `state`, + /// while the drain that lands its create belongs to the primary mount — so + /// reaching for `self.state` here found nothing and left the fork's entry on + /// its `local~` uid, which reads as an empty file for as long as the daemon + /// runs (`docs/BUGS.md` B74). A uid is unique across mounts, so at most one + /// of them matches and the others are no-ops. pub(crate) fn adopt_real_uid( &self, local: &NodeUid, @@ -431,8 +440,10 @@ impl Core { self.db .remap_local_uid(&local.to_string(), &real.to_string())?; - let mut st = self.state.lock(); - if let Some(ino) = st.by_uid.remove(local) { + self.for_each_state(|st| { + let Some(ino) = st.by_uid.remove(local) else { + return; + }; st.by_uid.insert(real.clone(), ino); if let Some(e) = st.entries.get_mut(&ino) { e.uid = real.clone(); @@ -460,8 +471,7 @@ impl Core { kids.push(ino); } } - } - drop(st); + }); // Write the real node through, now that nothing points at the old uid. if let Err(e) = self.db.upsert_node(&node) { warn!(%real, error = %e, "db upsert_node failed after remap"); @@ -604,11 +614,11 @@ impl Core { self.cache.evict(uid); self.evict_reader(uid); // The conflict copy is a node the tree has never seen. - let mut st = self.state.lock(); - if let Some(&ino) = st.by_uid.get(&parent) { - st.invalidate_listing(ino); - } - drop(st); + self.for_each_state(|st| { + if let Some(&ino) = st.by_uid.get(&parent) { + st.invalidate_listing(ino); + } + }); self.log_activity( ActivityKind::Upload, &name, @@ -662,13 +672,25 @@ impl Core { /// A node's parent uid and name, as the tree currently has them. pub(crate) fn node_place(&self, uid: &NodeUid) -> Option<(NodeUid, String)> { - { - let st = self.state.lock(); + // Across every mount: the drain runs on the primary Core, but a node + // written inside a sync folder is interned in that fork's inode space, + // so asking `self.state` alone always missed it and fell through to the + // DB. That still gave the right answer — the fallback below exists for + // exactly this shape of miss — but it went to disk for something already + // in memory, and only *because* of the fallback was it not a bug. + let mut found = None; + self.for_each_state(|st| { + if found.is_some() { + return; + } if let Some(entry) = st.by_uid.get(uid).and_then(|ino| st.entries.get(ino)) && let Some(parent) = entry.node.parent_uid.clone() { - return Some((parent, entry.node.name.clone())); + found = Some((parent, entry.node.name.clone())); } + }); + if found.is_some() { + return found; } // The in-memory tree only holds what has been walked to since the daemon // started, and the drain routinely runs before anything has walked to @@ -843,16 +865,17 @@ impl Core { if let Some(rev) = sealed_rev.clone() { self.own_sealed_revs.lock().insert(uid.clone(), rev); } - let mut st = self.state.lock(); - for aw in st.active_writes.values_mut() { - if aw.uid == *uid { - aw.base_mtime = sealed_mtime; - aw.base_size = sealed_size; - aw.base_revision_id = sealed_rev.clone(); - debug!(%uid, mtime = sealed_mtime, size = sealed_size, - "rebased open write handle onto the revision just uploaded"); + self.for_each_state(|st| { + for aw in st.active_writes.values_mut() { + if aw.uid == *uid { + aw.base_mtime = sealed_mtime; + aw.base_size = sealed_size; + aw.base_revision_id = sealed_rev.clone(); + debug!(%uid, mtime = sealed_mtime, size = sealed_size, + "rebased open write handle onto the revision just uploaded"); + } } - } + }); } // Ordered so that a write queued *during* the fetch above is still // caught: it took its baseline from the node's optimistic stamp, and @@ -860,16 +883,17 @@ impl Core { if self.rebaseline_pending(uid, &node) { return; } - let mut st = self.state.lock(); - let Some(parent) = st - .by_uid - .get(uid) - .and_then(|ino| st.entries.get(ino)) - .map(|e| e.parent) - else { - return; - }; - st.intern(parent, node); + self.for_each_state(|st| { + let Some(parent) = st + .by_uid + .get(uid) + .and_then(|ino| st.entries.get(ino)) + .map(|e| e.parent) + else { + return; + }; + st.intern(parent, node.clone()); + }); } /// Point a still-queued write at the revision this upload just sealed, and diff --git a/crates/pdfs-fuse/src/filesystem.rs b/crates/pdfs-fuse/src/filesystem.rs index d351635..fc1cd70 100644 --- a/crates/pdfs-fuse/src/filesystem.rs +++ b/crates/pdfs-fuse/src/filesystem.rs @@ -1264,6 +1264,124 @@ impl Filesystem for ProtonFs { // fuser warning logs when applications probe file descriptors. reply.error(Errno::ENOTTY); } + + // The callbacks below are deliberately unsupported. Each answers with the + // same errno `fuser`'s default would, minus the `[Not Implemented]` WARN it + // logs on every call — a probe is a normal thing for an application to do, + // not an incident. git alone emits one `link` per loose object, which + // buried real diagnostics in the journal (`docs/BUGS.md` B73). The errno + // contract these keep is asserted by the acceptance suite's clean-refusal + // case, so it must not drift. + + fn mknod( + &self, + _req: &Request, + _parent: INodeNo, + _name: &OsStr, + _mode: u32, + _umask: u32, + _rdev: u32, + reply: ReplyEntry, + ) { + // Drive stores files and folders; devices, FIFOs and sockets have no + // representation. Regular files arrive through `create`, never here. + reply.error(Errno::ENOSYS); + } + + fn symlink( + &self, + _req: &Request, + _parent: INodeNo, + _link_name: &OsStr, + _target: &Path, + reply: ReplyEntry, + ) { + // EPERM, not ENOSYS: the operation is meaningful, Drive simply cannot + // represent it, and callers treat EPERM as a definite "no" rather than + // retrying against an older interface. + reply.error(Errno::EPERM); + } + + fn link( + &self, + _req: &Request, + _ino: INodeNo, + _newparent: INodeNo, + _newname: &OsStr, + reply: ReplyEntry, + ) { + // A node has exactly one parent on Drive, so there are no hard links. + // git falls back to rename() on EPERM, which this filesystem supports. + reply.error(Errno::EPERM); + } + + fn fsyncdir( + &self, + _req: &Request, + _ino: INodeNo, + _fh: FileHandle, + _datasync: bool, + reply: ReplyEmpty, + ) { + // ENOSYS, which the kernel absorbs: it clears its fsyncdir probe and + // reports success to the caller. There is no dirty directory buffer to + // flush — directory changes are durable in SQLite before the callback + // that made them returns. + reply.error(Errno::ENOSYS); + } + + fn setxattr( + &self, + _req: &Request, + _ino: INodeNo, + _name: &OsStr, + _value: &[u8], + _flags: i32, + _position: u32, + reply: ReplyEmpty, + ) { + // Drive has no user-extended-attribute store. ENOSYS reaches userspace + // as EOPNOTSUPP, the errno callers expect from a filesystem without + // xattrs. Reads (`getxattr`/`listxattr`) are implemented, since the + // thumbnail interface is exposed that way. + reply.error(Errno::ENOSYS); + } + + fn lseek( + &self, + _req: &Request, + _ino: INodeNo, + _fh: FileHandle, + _offset: i64, + _whence: i32, + reply: ReplyLseek, + ) { + // Must stay ENOSYS. A file's holes are not knowable without fetching + // its blocks, and ENOSYS is what makes the kernel stop asking and + // emulate SEEK_DATA/SEEK_HOLE itself. Any other errno would surface to + // the caller as a failed seek. + reply.error(Errno::ENOSYS); + } + + #[allow(clippy::too_many_arguments)] + fn copy_file_range( + &self, + _req: &Request, + _ino_in: INodeNo, + _fh_in: FileHandle, + _offset_in: u64, + _ino_out: INodeNo, + _fh_out: FileHandle, + _offset_out: u64, + _len: u64, + _flags: CopyFileRangeFlags, + reply: ReplyWrite, + ) { + // Also must stay ENOSYS: there is no server-side copy, and this is the + // errno that sends coreutils (and the kernel) down the read/write path + // instead of failing the copy outright. + reply.error(Errno::ENOSYS); + } } /// The Proton Drive VFS. FUSE callbacks are synchronous, so the Tokio handle diff --git a/crates/pdfs-fuse/src/lib.rs b/crates/pdfs-fuse/src/lib.rs index e5c3879..1d38004 100644 --- a/crates/pdfs-fuse/src/lib.rs +++ b/crates/pdfs-fuse/src/lib.rs @@ -47,10 +47,11 @@ use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; use fuser::ReplyXattr; use fuser::{ - BackgroundSession, BsdFileFlags, Config, Errno, FileAttr, FileHandle, FileType, Filesystem, - FopenFlags, Generation, INodeNo, IoctlFlags, LockOwner, MountOption, Notifier, OpenAccMode, - OpenFlags, RenameFlags, ReplyAttr, ReplyCreate, ReplyData, ReplyDirectory, ReplyEmpty, - ReplyEntry, ReplyIoctl, ReplyOpen, ReplyWrite, Request, Session, TimeOrNow, WriteFlags, + BackgroundSession, BsdFileFlags, Config, CopyFileRangeFlags, Errno, FileAttr, FileHandle, + FileType, Filesystem, FopenFlags, Generation, INodeNo, IoctlFlags, LockOwner, MountOption, + Notifier, OpenAccMode, OpenFlags, RenameFlags, ReplyAttr, ReplyCreate, ReplyData, + ReplyDirectory, ReplyEmpty, ReplyEntry, ReplyIoctl, ReplyLseek, ReplyOpen, ReplyWrite, Request, + Session, TimeOrNow, WriteFlags, }; use pdfs_core::cache::{BLOCK_SIZE, Baseline, ContentCache, StagedWrite}; use pdfs_core::config::{AppDirs, SweepMode}; @@ -319,6 +320,75 @@ struct Core { /// uploading that same tree — the engine would upload files as they vanish /// and then walk the FUSE mount as if it were local. sync_locks: Arc>>>>, + /// Every live inode space: this mount's own `state` plus one per on-demand + /// fork ([`Core::fork_state`]). Shared by every fork, unlike `state` itself. + /// + /// There is exactly **one** drain thread per daemon, owned by the primary + /// mount, and it serves the whole `pending_op` table — including ops queued + /// by a forked mount, which keeps its nodes in its own `state`. So a drain + /// that reaches for `self.state` looks in the wrong inode space and silently + /// finds nothing: the fork's entry keeps its `local~` placeholder uid + /// forever, and every read of it short-circuits to empty because + /// [`Core::read_range`] refuses to ask the API about a `local~` uid. That + /// was `docs/BUGS.md` B74. Background work that rewrites node identity must + /// walk this instead ([`Core::for_each_state`]). + /// + /// `Weak`, so an unmounted fork's state is dropped rather than pinned here. + states: Arc, +} + +/// One live inode space, as published to the shared registry. +/// +/// The notifier travels with the state because the two are only meaningful +/// together: an inode number names a different node in every mount, so telling +/// the kernel to drop inode 42 is only correct on the session that minted it. +/// Background work that invalidates as well as mutates needs the matching pair, +/// which is what [`Core::for_each_mount`] hands it. +type LiveMount = (Arc>, Arc>); + +struct MountedState { + /// `Weak`, so an unmounted fork's state is dropped rather than pinned here; + /// a dead entry is reaped on the next registry walk. + state: std::sync::Weak>, + /// This mount's kernel notification channel — the same cell as + /// [`Core::notifier`], still empty until its session is spawned. + notifier: Arc>, +} + +/// Every mounted inode space in the daemon, shared by the primary `Core` and +/// every [`Core::fork_state`] clone of it. +/// +/// Split out from `Core` so the reap-and-walk rule can be tested on its own: a +/// fork that fails to appear here, or one that lingers after unmount, is exactly +/// the failure `docs/BUGS.md` B74 was. +#[derive(Default)] +struct StateRegistry(Mutex>); + +impl StateRegistry { + /// Publish an inode space and the channel to the session that owns it. + /// + /// Registering the same `state` twice would make every walk visit it twice; + /// callers register exactly once, at mount time. + fn register(&self, state: &Arc>, notifier: Arc>) { + let mut states = self.0.lock(); + states.retain(|m| m.state.strong_count() > 0); + states.push(MountedState { + state: Arc::downgrade(state), + notifier, + }); + } + + /// Every live mount, reaping the entries whose session has gone. Returns + /// owned handles so callers take the per-state locks one at a time rather + /// than holding the registry lock across the work. + fn live(&self) -> Vec { + let mut states = self.0.lock(); + states.retain(|m| m.state.strong_count() > 0); + states + .iter() + .filter_map(|m| Some((m.state.upgrade()?, m.notifier.clone()))) + .collect() + } } /// Why [`Core::apply_sync_folder_mode`] did not switch a folder. The two cases are @@ -333,6 +403,39 @@ enum SwitchBlocked { } impl Core { + /// Register an inode space so background work can reach it. Called once for + /// the primary mount and once per on-demand fork. + fn register_state(&self) { + self.states.register(&self.state, self.notifier.clone()); + } + + /// Run `apply` against every live inode space — this daemon's primary mount + /// and each on-demand fork — one lock at a time. + /// + /// The drain and the sync engine are per-daemon but node state is per-mount, + /// so anything they change about a node has to be offered to whichever mount + /// actually holds it. Which one that is cannot be known from the op: a + /// `pending_op` row records a uid, not the session that queued it. Applying + /// to all of them is correct because a uid is unique across mounts — at most + /// one state has an entry for it, and the rest are no-ops. + fn for_each_state(&self, mut apply: impl FnMut(&mut State)) { + for (state, _) in self.states.live() { + apply(&mut state.lock()); + } + } + + /// Like [`Core::for_each_state`], but also hands over the mount's kernel + /// notification channel, so work that invalidates cached metadata reaches + /// the session that actually minted the inodes it is naming. + /// + /// `None` for a mount whose session has not been spawned yet — there is + /// nothing to invalidate in a kernel that has never been told about it. + fn for_each_mount(&self, mut apply: impl FnMut(&mut State, Option<&Notifier>)) { + for (state, notifier) in self.states.live() { + apply(&mut state.lock(), notifier.get()); + } + } + /// Rehydrate the in-memory `State` maps from the DB on mount, so a cold /// start serves previously-seen metadata (stable inodes, instant listings) /// without re-hitting the API. The root inode is already installed by @@ -1537,7 +1640,14 @@ impl Core { Errno::EIO })?; self.hidden.lock().insert(uid.clone()); - self.state.lock().forget(uid); + // Withdrawn from every mount, not just the one the unlink came through: + // a sync folder maps a remote folder that also exists under My Files, so + // the same uid can be interned in two inode spaces at once and the other + // one would go on serving a file the user just trashed + // (`docs/BUGS.md` B74). + self.for_each_state(|st| { + st.forget(uid); + }); self.cache.evict(uid); self.evict_reader(uid); self.wake_drain(); @@ -2154,7 +2264,11 @@ impl Core { self.rt .block_on(self.client.rename_node(&uid, new_name, None)) .map_err(|e| CoreError::from_api(&e, "rename"))?; - self.state.lock().forget(&uid); + // Every mount, so a fork showing the same node re-interns it under the + // new name instead of keeping the old one (`docs/BUGS.md` B74). + self.for_each_state(|st| { + st.forget(&uid); + }); self.invalidate_parent_listing(rel); Ok(new_name.to_string()) } @@ -2215,7 +2329,9 @@ impl Core { fn remove_replaced(&self, uid: &NodeUid, name: &str) -> Result<(), Errno> { if is_local_uid(uid) { self.discard_queued_ops(uid); - self.state.lock().forget(uid); + self.for_each_state(|st| { + st.forget(uid); + }); debug!(%uid, name, "replaced a node whose create was still queued"); return Ok(()); } @@ -2228,7 +2344,11 @@ impl Core { return Err(Errno::EIO); } self.discard_queued_ops(uid); - self.state.lock().forget(uid); + // The node was trashed on the server; withdraw it from every inode + // space that had it, not only the mount the rename came through. + self.for_each_state(|st| { + st.forget(uid); + }); self.cache.evict(uid); self.evict_reader(uid); self.invalidate_trash(); @@ -2413,14 +2533,15 @@ impl Core { hidden.remove(uid); } } - { - let mut st = self.state.lock(); - for parent in parents { - if let Some(&ino) = st.by_uid.get(&parent) { + // Keyed by uid, so it applies to every mount: a restored node reappears + // in whichever inode spaces show its parent, not only the primary one. + self.for_each_state(|st| { + for parent in &parents { + if let Some(&ino) = st.by_uid.get(parent) { st.invalidate_listing(ino); } } - } + }); self.invalidate_trash(); Ok(parsed.len()) } @@ -2456,11 +2577,15 @@ impl Core { /// Forget every trace of nodes that no longer exist anywhere: their inode and /// DB row, and their cached content. fn drop_local(&self, uids: &[NodeUid]) { - let mut st = self.state.lock(); - for uid in uids { - st.forget(uid); - } - drop(st); + // Every mount: these nodes are gone from the server, and a sync-folder + // fork showing the same uid would otherwise keep serving them + // (`docs/BUGS.md` B74). A uid is unique across inode spaces, so this is + // a no-op in every mount but the one that holds it. + self.for_each_state(|st| { + for uid in uids { + st.forget(uid); + } + }); for uid in uids { self.cache.evict(uid); self.evict_reader(uid); @@ -3423,11 +3548,58 @@ mod pending_size_tests { #[cfg(test)] mod tests { use super::{ - Intervals, PendingRevision, conflict_name, copy_pending_for_truncate, fuse_name, - is_stale_mount, node_visible, rename_needs_queue, + Intervals, PendingRevision, StateRegistry, conflict_name, copy_pending_for_truncate, + fuse_name, is_stale_mount, node_visible, rename_needs_queue, }; use pdfs_core::cache::{Baseline, StagedWrite}; + /// The registry every mount publishes itself into is what lets the daemon's + /// single drain thread reach a node living in an on-demand fork's inode + /// space. A fork missing from it reads as an empty file for the life of the + /// daemon (`docs/BUGS.md` B74), so the walk has to see *every* registered + /// mount — and only the ones still mounted. + #[test] + fn state_registry_walks_every_live_mount_and_reaps_the_rest() { + let registry = StateRegistry::default(); + assert!(registry.live().is_empty(), "a fresh registry has no mounts"); + + // The primary mount plus two on-demand forks. + let (primary, _p) = state_test_helper(); + let (fork_a, _a) = state_test_helper(); + let (fork_b, _b) = state_test_helper(); + let primary = std::sync::Arc::new(parking_lot::Mutex::new(primary)); + let fork_a = std::sync::Arc::new(parking_lot::Mutex::new(fork_a)); + let fork_b = std::sync::Arc::new(parking_lot::Mutex::new(fork_b)); + for state in [&primary, &fork_a, &fork_b] { + registry.register(state, std::sync::Arc::new(std::sync::OnceLock::new())); + } + assert_eq!(registry.live().len(), 3, "every registered mount is walked"); + + // `live()` must not itself pin a mount alive, or an unmounted fork could + // never be reaped. + assert_eq!( + std::sync::Arc::strong_count(&fork_a), + 1, + "the registry holds forks weakly" + ); + + // Unmounting a fork drops its state; the dead entry goes on the next walk. + drop(fork_a); + assert_eq!(registry.live().len(), 2, "an unmounted fork is dropped"); + assert_eq!( + registry.0.lock().len(), + 2, + "and its slot is reaped, not merely skipped" + ); + + drop(primary); + drop(fork_b); + assert!( + registry.live().is_empty(), + "the last mount going away empties the registry" + ); + } + /// The predicate must answer *only* for a dead FUSE connection. A healthy /// directory and an absent path are both "not stale" — widening it to any /// `metadata` error would make the on-demand restore lazily unmount paths diff --git a/crates/pdfs-fuse/src/mount.rs b/crates/pdfs-fuse/src/mount.rs index 7f55288..052173c 100644 --- a/crates/pdfs-fuse/src/mount.rs +++ b/crates/pdfs-fuse/src/mount.rs @@ -222,7 +222,11 @@ pub fn mount( sync_tx, mounts: Arc::new(Mutex::new(HashMap::new())), sync_locks: Arc::new(Mutex::new(HashMap::new())), + states: Arc::new(StateRegistry::default()), }; + // Before anything can queue work against it: the drain thread below reaches + // every mount's inode space through this registry, not through `core.state`. + core.register_state(); // 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. @@ -341,15 +345,7 @@ pub fn mount( // 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, - )); + rt.spawn(run_event_sync(client, scope, core.clone())); // Stop signals (SIGTERM from `systemctl --user stop`, SIGINT from Ctrl-C) // are delivered onto the async runtime; bridge them onto a sync channel so diff --git a/crates/pdfs-fuse/src/sharing.rs b/crates/pdfs-fuse/src/sharing.rs index 8d9f36f..8a20db2 100644 --- a/crates/pdfs-fuse/src/sharing.rs +++ b/crates/pdfs-fuse/src/sharing.rs @@ -466,15 +466,33 @@ impl Core { /// been seen through the mount (e.g. shared but not browsed to this session) — /// the caller then leaves the path empty. pub(crate) fn rel_path_for_uid(&self, uid: &NodeUid) -> Option { - let st = self.state.lock(); - let mut ino = *st.by_uid.get(uid)?; - let mut parts = Vec::new(); - while ino != ROOT_INO { - let entry = st.entries.get(&ino)?; - parts.push(entry.node.name.clone()); - ino = entry.parent; - } - parts.reverse(); - Some(parts.join("/")) + // Searched across every mount: a node browsed through a sync folder is + // interned in that fork's inode space, so the primary state alone had no + // entry for it and the caller silently showed an empty path + // (`docs/BUGS.md` B74). The path is relative to whichever mount holds + // it, which is the one the user reached it through. + let mut path = None; + self.for_each_state(|st| { + if path.is_some() { + return; + } + path = walk_to_root(st, uid); + }); + path } } + +/// Walk one inode space from `uid` up to its root, building the path. `None` if +/// that mount has never interned the node, or if the chain breaks partway — +/// a broken chain is a missing path, never a partial one. +fn walk_to_root(st: &super::State, uid: &NodeUid) -> Option { + let mut ino = *st.by_uid.get(uid)?; + let mut parts = Vec::new(); + while ino != ROOT_INO { + let entry = st.entries.get(&ino)?; + parts.push(entry.node.name.clone()); + ino = entry.parent; + } + parts.reverse(); + Some(parts.join("/")) +} diff --git a/crates/pdfs-fuse/src/sweep.rs b/crates/pdfs-fuse/src/sweep.rs index 71eab9d..835c982 100644 --- a/crates/pdfs-fuse/src/sweep.rs +++ b/crates/pdfs-fuse/src/sweep.rs @@ -153,7 +153,13 @@ impl Core { return; } } - self.state.lock().forget_or_unlink(uid); + // Every mount, not just the primary: a conflict copy inside a sync + // folder lives in that fork's inode space, and leaving it hooked up + // there keeps a node we have just trashed remotely visible and + // readable there (`docs/BUGS.md` B74). + self.for_each_state(|st| { + st.forget_or_unlink(uid); + }); self.discard_queued_ops(uid); self.cache.evict(uid); self.evict_reader(uid); @@ -232,12 +238,21 @@ impl Core { /// Whether a node has a write in flight or a file handle open against it. /// Read-only opens use fh 0 and are not tracked, so this is specifically the /// writer/holder check. + /// Asked across **every** mount. The sweep runs on the primary Core but + /// trashes nodes anywhere in the tree, and a file open inside a sync folder + /// is tracked in that fork's `State` — so consulting `self.state` alone + /// answered "idle" for every such file and let the sweep delete one with a + /// writer attached (`docs/BUGS.md` B74). Busy anywhere means busy. fn is_busy(&self, uid: &NodeUid) -> bool { - let state = self.state.lock(); - let Some(&ino) = state.by_uid.get(uid) else { - return false; - }; - state.active_writes.contains_key(&ino) || state.handles.values().any(|&held| held == ino) + let mut busy = false; + self.for_each_state(|st| { + let Some(&ino) = st.by_uid.get(uid) else { + return; + }; + busy |= + st.active_writes.contains_key(&ino) || st.handles.values().any(|&held| held == ino); + }); + busy } /// Surface a conflict copy that needs the user's attention, at most once per diff --git a/docs/BUGS.md b/docs/BUGS.md index bf27317..09db95e 100644 --- a/docs/BUGS.md +++ b/docs/BUGS.md @@ -2148,7 +2148,10 @@ default (B71) precisely so that stays true until there is field evidence. ## B70 — Browser in-flight temp files uploaded and conflict-forked on the on-demand mount -**Status:** Layers A + B fixed in-tree 2026-07-25 — offline-tested; **live validation pending**. +**Status:** Layers A + B fixed in-tree 2026-07-25 — **live-validated 2026-07-26**. +Validation was blocked until B74 was fixed (its scenario ends in a rename, which +B74 made read back as empty); with that out of the way the acceptance case +`regression B70: transient download name is not sealed` passes live. **Found:** triaging the divergent conflict copies B69 left behind. In `~/Downloads` (an on-demand mount) the complete files had been forked into `(sync-conflict)` copies while their truncated partials kept the canonical name — e.g. @@ -2362,3 +2365,171 @@ or the CLI's response read in `crates/pdfs-cli/src/main.rs`. **Impact:** the trash listing is the user's recovery path after any destructive operation, including the B71 sweep. It needs to work before the sweep is allowed to trash anything. + +## B73 — Unimplemented FUSE handlers log a WARN per probe + +**Status:** Fixed in-tree 2026-07-26 — live-validated. Found 2026-07-26 by the +expanded acceptance suite. +**Found:** `scripts/fuse-acceptance.sh --live` step "unsupported operations refuse +cleanly". Each probe produced a multi-line `fuser` warning in the daemon journal: + +``` +WARN fuser: [Not Implemented] symlink(parent: INodeNo(0x2d), link_name: "symlink", target: "link-target") +WARN fuser: [Not Implemented] link(ino: INodeNo(0x94), newparent: INodeNo(0x2d), newname: "hardlink") +WARN fuser: [Not Implemented] mknod(parent: INodeNo(0x2d), name: "fifo", mode: 4516, umask: 0x12, rdev: 0) +WARN fuser: [Not Implemented] setxattr(ino: INodeNo(0x91), name: "user.pdfs.probe", flags: 0x0, position: 0) +WARN fuser: [Not Implemented] lseek(ino: INodeNo(0x93), fh: 0, offset: 0, whence: 3) +``` + +**Cause:** `ProtonFs` does not implement these callbacks, so `fuser`'s default +trait method answers `ENOSYS` *and logs at WARN*. The refusal is correct — the +acceptance suite accepts every one of them as a clean refusal — but the logging +is not conditional on anything the user did wrong. +**Impact:** cosmetic, but these are not exotic calls: `git` probes symlinks, +`cp -a` probes hard links, `tar` and `cp --sparse` probe `SEEK_HOLE`/`SEEK_DATA` +(`whence: 3` above is `SEEK_DATA`), and any xattr-aware tool probes `setxattr`. +Warnings that look like faults bury real ones and make both `--journal-check` +and manual triage noisier. +**Measured** across two full acceptance runs against one daemon lifetime (PID +887441, mount never remounted): + +| op | warnings | behaviour | +|---|---|---| +| `link` | 11 | repeats on every call | +| `symlink` | 4 | repeats on every call | +| `mknod` | 2 | repeats on every call | +| `setxattr` | 1 | kernel cached the `ENOSYS`; never asked again | +| `lseek` | 1 | cached | +| `copy_file_range` | 1 | cached | +| `fsyncdir` | 1 | cached | + +So the FUSE kernel module remembers `ENOSYS` for the file-level operations and +stops issuing them for the lifetime of the mount, but not for the namespace +operations, which are re-issued forever. + +`link` leads for a sharper reason than "tools probe it": git uses `link()` to +place **every loose object**, falling back to `rename()` when it fails. In one +workloads run, 9 of 10 `link` warnings carried a 38-hex-character name — the +`.git/objects/ab/<38 hex>` layout — from just 3 commits over 3 files. The volume +is therefore proportional to work done, not a fixed startup cost: a `git clone` +of a real repository emits one multi-line warning per object. The lone `symlink` +that is not the acceptance probe is git's `core.symlinks` detection at +`git init` (a random name pointing at `testing`). + +That moves this from cosmetic to a genuine operational problem: ordinary use of +git inside a mount can bury every other log line. + +**Fix (applied):** all seven callbacks are now implemented explicitly in +`crates/pdfs-fuse/src/filesystem.rs`, each answering with **the same errno +`fuser`'s default would** and nothing else — no WARN. Following `ioctl` +(which already used this pattern and says why in its comment), the errno +contract is unchanged, so the acceptance suite's clean-refusal assertions hold +byte for byte: + +| op | errno | why that one | +|---|---|---| +| `symlink`, `link` | `EPERM` | meaningful operation, unrepresentable on Drive; git falls back to `rename()` | +| `mknod` | `ENOSYS` | no devices/FIFOs/sockets; regular files arrive via `create` | +| `setxattr` | `ENOSYS` | surfaces as `EOPNOTSUPP`; reads stay implemented for the thumbnail interface | +| `fsyncdir` | `ENOSYS` | kernel absorbs it and reports success; dir changes are already durable in SQLite | +| `lseek`, `copy_file_range` | `ENOSYS` | **must stay** `ENOSYS` — it is what makes the kernel emulate `SEEK_HOLE`/`SEEK_DATA` and coreutils fall back to read/write instead of failing | + +Only the three namespace ops were strictly required (the other four silence +themselves once the kernel caches `ENOSYS`), but implementing all seven keeps the +refusal explicit and documented at the call site rather than inherited. + +**Verified:** full live acceptance run after the fix — 19 pass / 1 skip, the +capability diff identical to the pre-fix run, and **0** `Not Implemented` lines +in the journal across the whole run (was 17, 10 of them `link`). +**Regression cover:** the acceptance suite's clean-refusal probes assert the +errno contract; they pass unchanged. + +## B74 — A drained create never reaches a sync-folder mount's inode space, so the file reads as empty + +**Status:** Fixed in-tree 2026-07-26 — live-validated. Found 2026-07-26 by the +expanded acceptance suite (`regression B70` case), then isolated to a broader +trigger. +**Impact:** on a *sync folder* mount (not `~/ProtonDrive`), a file whose create +was still queued reads back as **0 bytes for as long as the daemon runs**, even +though the bytes are on Drive intact. This is the write-to-temp-then-rename +pattern used by every browser download, most editor saves, `rsync`, and `git`. + +**Not data loss.** An earlier revision of this entry called it permanent data +loss; that was wrong and is corrected here. The uploaded revision is complete and +correct: for every "lost" file the node row carried the right `size` and +`active_revision_state: "Active"`, and a daemon restart made all of them read +back with the expected sha256. Nothing had to be recovered. The defect was +entirely local and entirely in the read path. + +**Reproduction** (on an on-demand mount, per file: write, close, rename, wait for +`pdfs status` to report the queue drained, then read back): + +| scenario | result | +|---|---| +| plain name, never renamed | content intact | +| plain name, renamed after close | **0 bytes until restart** | +| `.crdownload`, renamed after close | **0 bytes until restart** | +| `.crdownload`, never renamed | content intact | +| already-settled file, renamed | content intact | +| any of the above, after a daemon restart | content intact | + +The gap between `close()` and `rename()` does not matter — 0 s, 2 s and 20 s all +behave the same. The boundary is not timing but state: the affected node is one +whose **create op was still queued**, on a mount that is not the primary one. + +**Root cause — one drain thread, many inode spaces.** The daemon mounts +`~/ProtonDrive` plus one FUSE session per `ondemand` sync folder. Each of those +is a `Core::fork_state` clone with its **own** `State` — its own `entries`, +`by_uid`, `children`, `next_ino` — while sharing `db`, `cache`, `pending` and +`client`. There is exactly **one** `pdfs-drain` thread in the process, owned by +the primary mount, and it serves the whole shared `pending_op` table. + +So when a create queued by a sync-folder mount lands, the drain calls +`adopt_real_uid` — which looked only at `self.state`, the *primary* mount's inode +space. The fork's entry keeps its `local~…` placeholder uid forever. And +`Core::read_range` (`crates/pdfs-fuse/src/reads.rs:314`) short-circuits a local +uid to an empty vec: + +```rust +if is_local_uid(uid) { return Ok(Vec::new()); } +``` + +which is why the read is silent, instant, and zero-length. A restart rebuilds +every state from the DB, where `remap_local_uid` had already written the real +uid, so the file reads correctly from then on. + +Confirmed live: `read handler uid=local~1785067300752-0 fsize=140000` three +seconds after `pending create landed`, and at the same moment +`adopt_real_uid … hit=None strays=[]` — the primary state had no entry for the +uid at all. `/proc//task/*/comm` showed 1 `pdfs-drain` against 8 fuser +sessions. + +**Fix:** `Core` gained a `states: Arc>>>>` registry +that every mount publishes itself into (`Core::register_state`, called from +`mount.rs` for the primary and from `fork_state` for each fork), plus +`Core::for_each_state`, which walks every live inode space one lock at a time. +A uid is unique across mounts, so at most one state matches and the rest are +no-ops. Five drain sites were converted: `adopt_real_uid`, `adopt_drained_name`, +the conflict-copy `invalidate_listing`, the `active_writes` baseline rebase in +`refresh_after_upload`, and that function's closing `intern`. + +**It was completely silent.** No `ERROR`, no `WARN`, no failed op, no conflict +copy, no retry — `pdfs status` reported the queue drained, because it had. Any +diagnosis that relies on the journal or on queue state will not see this. + +**Regression cover:** `scripts/fuse-acceptance.py`, case +`regression B74: rename after close keeps its content` — the narrowest form +(plain names, no overwrite, no transient suffix). It failed in two consecutive +live runs before the fix and passes after it. The pre-existing B70 case failed on +the same cause and now passes too. + +**Remaining work.** `drain.rs` still has fork-blind *reads* (`node_place`, +`root_uid`, `parent_is_gone`) and `lib.rs` has ~31 direct `state.lock()` sites; +any of them reached from a background thread has the same shape of problem. +Worth an audit, and worth a unit test over `for_each_state` so a future fork +cannot silently drop out of the registry. + +**Note on B70:** B70's own fix was never implicated. The `.crdownload` park/un-park +behaves correctly in isolation; the empty read arrived with the rename, on the +same path a plain temp file takes. B70's live validation was blocked on this and +is now unblocked — its acceptance case passes. diff --git a/docs/TESTING.md b/docs/TESTING.md index f1df3c1..aad5bb5 100644 --- a/docs/TESTING.md +++ b/docs/TESTING.md @@ -14,9 +14,12 @@ cargo clippy --workspace --all-targets -- -D warnings The account-free suite exercises the filesystem syscall contract against a temporary local directory. This validates the runner and covers creation/open flags, positioned and vectored I/O, truncation and sparse files, `mmap`, -`sendfile`, durability barriers, namespace/error semantics, open-file lifetime, -names and enumeration, and concurrent I/O. It never reads application state or -contacts Proton: +`sendfile`, `copy_file_range`, durability barriers, namespace/error semantics, +`renameat2` flags, open-file lifetime, names and enumeration, readdir stability +under concurrent mutation, metadata updates, extended attributes, allocation and +punched holes, refusal semantics for unimplemented operations, concurrent I/O, +and real application workloads. It never reads application state or contacts +Proton: ```console scripts/fuse-acceptance.sh --offline-only @@ -34,6 +37,85 @@ Use a dedicated test account without irreplaceable data: scripts/fuse-acceptance.sh --live /mnt/on-demand-testmount ``` +### The mount is diffed against an ordinary filesystem + +The local reference run is not only a self-test of the runner. Every case +records the facts it established, and a live run is compared against those +recordings, so a case fails when the mount *differs* from an ordinary +filesystem — not only when it raises. The report names the exact key: + +``` + [DIVERGENCE] live FUSE /mnt/testmount does not match the reference filesystem: + ! creation and open flags.truncated.size: live=6 reference=0 +``` + +Facts are recorded in one of two ways, chosen in the test body: + +- **compared** — a claim about filesystem semantics. The mount must agree. +- **noted** — context that legitimately differs: inode numbers, block counts, + reported mode bits, and which optional syscalls exist at all. Reported under + `[capabilities]`, never a failure. + +Because the choice is made at each call site, the set of accepted divergences +is visible in the test that accepts it. There is no separate allowlist to drift +out of date. When adding a case, prefer `record`; reach for `note` only when an +ordinary filesystem can do something a network filesystem genuinely cannot. + +### Operations the filesystem does not implement + +`git`, `rsync`, `cp -a`, `tar`, `df` and most installers probe for symlinks, +hard links, FIFOs, device nodes, `statfs`, file locks, `copy_file_range`, +`O_DIRECT` and `SEEK_HOLE` whether or not a filesystem offers them. The suite +asserts that each either works or is *refused cleanly* — a recognised errno, +never `EIO` and never a hang. A refusal that is merely unimplemented is fine; +one that is dirty breaks software that never asked for the feature. + +### Regressions from the bug ledger + +Cases named `regression B` reproduce a bug from `docs/BUGS.md` so it stays +fixed. B69 (identical rewrite must not fork a `(sync-conflict …)` copy), B70 (an +in-flight `.crdownload` must not be sealed as a revision) and B74 (a file +renamed after close must keep its content) need a real mount and a reachable +daemon; they are skipped, not failed, without one. + +B74 is currently **failing** — it is a live data-loss defect, not a regression +guard, and it was found by exactly this mechanism: the B70 case failed, and +narrowing it produced a smaller reproduction that has its own case now. Expect a +red run against a real mount until it is fixed. + +A regression case asserts against the *remote* outcome, so it must wait for the +queue to drain before reading. Reading straight back through the mount proves +nothing: the local attribute and content caches will happily return the bytes +that were just written even when nothing reached Drive, which is precisely how +B74 stayed invisible. + +### Timeouts, reports, and hung mounts + +A wedged FUSE operation cannot be interrupted from Python, so each case runs +under a soft alarm (`--timeout`, default 180s) that reports the case and lets +cleanup run, backed by a hard stop that dumps every thread's stack and exits. +That backstop is the only way to learn *where* a mount hung. A timeout ends the +run: later cases against a wedged mount produce noise, not information. + +```console +scripts/fuse-acceptance.sh --live /mnt/testmount \ + --timeout 300 --report-junit results.xml --budget 30 +``` + +`--report-json` also writes every recorded observation, which is what to attach +to a bug report. `--budget SECONDS` flags cases that pass but got slower — +B5 was a 186 ms-per-call regression that a pass/fail suite could not see. +`--list` prints every case and which targets it runs against, and `--fail-fast` +stops at the first failure. + +`--journal-check` fails the run if `proton-drive.service` logged any error while +it ran. A suite can pass every assertion while the daemon logs failures behind +it; that has happened, and it was caught only by reading the journal by hand. + +Abandoned `pdfs-acceptance-*` roots from a crashed run are removed at startup, +but only if they match the exact generated name and are over an hour old +(`PDFS_ACCEPTANCE_REAP_AGE`), so a concurrent run is never disturbed. + ### Automated mode matrix The managed live runner accepts two existing, empty, unmounted directories. It @@ -53,6 +135,16 @@ in `pdfs sync list`. Use `PDFS_ACCEPTANCE_SYNC_TIMEOUT` to extend transition timeouts and `PDFS_ACCEPTANCE_PDFS` to select a non-installed CLI binary, for example `target/debug/pdfs`. +### Durability drill + +`--durability` restarts `proton-drive.service` in the middle of the suite and +re-verifies the bytes written before it. `staging/` and `recovery/` can hold the +only copy of a file whose upload has not finished, so this is the case that +proves the queue is persistent rather than in-memory. It is opt-in because it +interrupts a running daemon; `PDFS_ACCEPTANCE_UNIT` selects a different unit. + +### Selecting cases + During development, `PDFS_ACCEPTANCE_ONLY` runs tests whose descriptive name contains the supplied text (for example `PDFS_ACCEPTANCE_ONLY=namespace`). A release acceptance run must leave it unset so the complete contract executes. diff --git a/scripts/fuse-acceptance.py b/scripts/fuse-acceptance.py index b7a0fc8..f7e29f3 100755 --- a/scripts/fuse-acceptance.py +++ b/scripts/fuse-acceptance.py @@ -1,21 +1,110 @@ #!/usr/bin/env python3 -"""Account-free filesystem API contract plus opt-in live FUSE acceptance.""" +"""Account-free filesystem API contract plus opt-in live FUSE acceptance. + +Every case runs first against a temporary local directory and then, when asked, +against the mount. The local run is not just a smoke test of the runner: its +observations become the reference that the mount is diffed against, so a case +fails when the mount *differs* from an ordinary filesystem, not only when it +raises. Differences that are legitimate (device ids, block counts, capabilities +a network filesystem cannot have) are recorded as notes instead of comparisons, +which makes the set of accepted divergences explicit and reviewable. +""" from __future__ import annotations import argparse import concurrent.futures +import contextlib +import ctypes import errno +import faulthandler +import fcntl import hashlib +import json import mmap import os from pathlib import Path +import re import shutil +import signal +import sqlite3 import stat import subprocess +import sys +import tarfile import tempfile import time import uuid +import xml.etree.ElementTree as ElementTree + +# A wedged FUSE operation is unkillable from Python, so every case runs under a +# soft alarm and a hard backstop. The soft alarm raises inside the main thread +# and lets cleanup and reporting run; the backstop dumps every thread's stack +# and exits, which is the only way to learn where a mount actually hung. +DEFAULT_TIMEOUT = int(os.environ.get("PDFS_ACCEPTANCE_TIMEOUT", "180")) +HARD_TIMEOUT_GRACE = 20 + +REFERENCE = "reference" +LIVE = "live" + +# Failing an unsupported operation is fine; failing it *dirtily* is not. These +# are the errno values that mean "this filesystem does not offer that", as +# opposed to EIO or a hang, which mean something is broken. +CLEAN_UNSUPPORTED = { + errno.EPERM, + errno.ENOSYS, + errno.EOPNOTSUPP, + errno.EACCES, + errno.EINVAL, + errno.ENOTTY, +} + +STALE_ROOT = re.compile(r"^pdfs-acceptance-[0-9a-f]{32}$") +ANSI = re.compile(r"\x1b\[[0-9;]*m") +ERROR_LEVEL = re.compile(r"\bERROR\b") + +FALLOC_FL_KEEP_SIZE = 0x01 +FALLOC_FL_PUNCH_HOLE = 0x02 +RENAME_NOREPLACE = 1 << 0 +RENAME_EXCHANGE = 1 << 1 +AT_FDCWD = -100 + +_libc = ctypes.CDLL(None, use_errno=True) + + +def _libc_call(name: str, argtypes: list, *args) -> None: + entry = getattr(_libc, name, None) + if entry is None: + raise OSError(errno.ENOSYS, f"{name} is unavailable in libc") + entry.argtypes = argtypes + entry.restype = ctypes.c_int + ctypes.set_errno(0) + if entry(*args) != 0: + code = ctypes.get_errno() + raise OSError(code, os.strerror(code)) + + +def fallocate(fd: int, mode: int, offset: int, length: int) -> None: + _libc_call( + "fallocate", + [ctypes.c_int, ctypes.c_int, ctypes.c_longlong, ctypes.c_longlong], + ctypes.c_int(fd), + ctypes.c_int(mode), + ctypes.c_longlong(offset), + ctypes.c_longlong(length), + ) + + +def renameat2(old: Path, new: Path, flags: int) -> None: + _libc_call( + "renameat2", + [ctypes.c_int, ctypes.c_char_p, ctypes.c_int, ctypes.c_char_p, ctypes.c_uint], + ctypes.c_int(AT_FDCWD), + os.fsencode(old), + ctypes.c_int(AT_FDCWD), + os.fsencode(new), + ctypes.c_uint(flags), + ) def check(condition: bool, message: str) -> None: @@ -32,6 +121,33 @@ def expect_errno(expected: set[int], operation, description: str) -> None: raise AssertionError(f"{description}: unexpectedly succeeded") +def check_bytes(actual: bytes, expected: bytes, message: str) -> None: + """Assert byte equality with a report that is actually diagnosable. + + "wrong bytes" is useless in a log: a truncated upload, a zero-filled gap and + a wholly different revision all produce it. Say which one it was. + """ + if actual == expected: + return + detail = f"{message}: got {len(actual)} bytes, expected {len(expected)}" + if len(actual) == len(expected): + offset = next(i for i, (a, b) in enumerate(zip(actual, expected)) if a != b) + run_end = offset + while run_end < len(actual) and actual[run_end] != expected[run_end]: + run_end += 1 + zeros = actual[offset:run_end] == bytes(run_end - offset) + detail += ( + f"; first difference at {offset} ({run_end - offset} bytes" + f"{', all zero' if zeros else ''})" + f"; got {actual[offset:offset + 16]!r} expected {expected[offset:offset + 16]!r}" + ) + elif expected.startswith(actual): + detail += "; got a truncated prefix of the expected content" + elif actual.startswith(expected): + detail += "; got the expected content plus trailing data" + raise AssertionError(detail) + + def read(path: Path) -> bytes: with path.open("rb", buffering=0) as file: return file.read() @@ -50,7 +166,143 @@ def write_durable(path: Path, data: bytes) -> None: os.close(fd) -def test_create_flags(root: Path) -> None: +class Observations: + """Per-case facts, split into what must match the reference and what may not. + + `record` is a claim about filesystem semantics: the mount has to agree with + an ordinary filesystem or the case fails. `note` is context — inode numbers, + block counts, which optional syscalls exist — that is reported in the diff + but never fails a run. Choosing between them at the call site is what keeps + the accepted-divergence list honest; there is no separate allowlist to drift. + """ + + def __init__(self) -> None: + self.compared: dict[str, object] = {} + self.noted: dict[str, object] = {} + self._scope = "" + + def scope(self, name: str) -> None: + self._scope = name + + def _key(self, key: str) -> str: + return f"{self._scope}.{key}" if self._scope else key + + def record(self, key: str, value) -> None: + self.compared[self._key(key)] = value + + def note(self, key: str, value) -> None: + self.noted[self._key(key)] = value + + def stat(self, key: str, path: Path) -> os.stat_result: + """Record the portable half of a stat and note the rest.""" + info = os.lstat(path) + self.record(f"{key}.type", stat.S_IFMT(info.st_mode)) + self.record(f"{key}.size", info.st_size) + self.record(f"{key}.is_dir", stat.S_ISDIR(info.st_mode)) + self.note(f"{key}.mode", stat.S_IMODE(info.st_mode)) + self.note(f"{key}.nlink", info.st_nlink) + self.note(f"{key}.blocks", info.st_blocks) + return info + + def probe(self, key: str, operation, allow: set[int] = CLEAN_UNSUPPORTED) -> Outcome: + """Run an optional operation; demand that failure at least be clean. + + The reference filesystem supports more than a network filesystem can, so + the *outcome* is a note. What is asserted is the shape of a refusal: a + recognised errno, never EIO and never a hang. + """ + try: + value = operation() + except OSError as error: + check( + error.errno in allow, + f"{key}: refused with errno {error.errno} ({errno.errorcode.get(error.errno)}); " + f"expected success or one of {sorted(allow)}", + ) + self.note(key, f"errno:{errno.errorcode.get(error.errno, error.errno)}") + return Outcome(False, None, error.errno) + self.note(key, "ok") + return Outcome(True, value, None) + + +class Outcome: + """The result of an optional operation: did it work, and if not, why.""" + + def __init__(self, ok: bool, value, code: int | None) -> None: + self.ok = ok + self.value = value + self.errno = code + + def __bool__(self) -> bool: + return self.ok + + +class Context: + """Everything a case may touch: its sandbox, its recorder, and the daemon.""" + + def __init__(self, root: Path, obs: Observations, kind: str, daemon=None) -> None: + self.root = root + self.obs = obs + self.kind = kind + self.daemon = daemon + + @property + def is_live(self) -> bool: + return self.kind == LIVE + + def record(self, key: str, value) -> None: + self.obs.record(key, value) + + def note(self, key: str, value) -> None: + self.obs.note(key, value) + + def probe(self, key: str, operation, allow: set[int] = CLEAN_UNSUPPORTED): + return self.obs.probe(key, operation, allow) + + def stat(self, key: str, path: Path) -> os.stat_result: + return self.obs.stat(key, path) + + def require_daemon(self) -> Daemon: + if self.daemon is None: + raise Skip("no reachable pdfs daemon; set PDFS_ACCEPTANCE_PDFS") + return self.daemon + + +class Skip(Exception): + """A case cannot run here, and that is not a failure.""" + + +class TestTimeout(Exception): + pass + + +@contextlib.contextmanager +def time_limit(seconds: int, label: str): + if seconds <= 0: + yield + return + + def on_alarm(_signum, _frame): + raise TestTimeout(f"{label} exceeded {seconds}s") + + previous = signal.signal(signal.SIGALRM, on_alarm) + faulthandler.dump_traceback_later(seconds + HARD_TIMEOUT_GRACE, exit=True) + signal.setitimer(signal.ITIMER_REAL, seconds) + try: + yield + finally: + signal.setitimer(signal.ITIMER_REAL, 0) + faulthandler.cancel_dump_traceback_later() + signal.signal(signal.SIGALRM, previous) + + +# -------------------------------------------------------------------------- +# Filesystem API contract +# -------------------------------------------------------------------------- + + +def test_create_flags(ctx: Context) -> None: + root = ctx.root path = root / "flags" fd = os.open(path, os.O_CREAT | os.O_EXCL | os.O_RDWR, 0o600) os.write(fd, b"abcdef") @@ -61,12 +313,15 @@ def test_create_flags(root: Path) -> None: os.write(fd, b"++") os.close(fd) check(read(path) == b"abcdef++", "O_APPEND ignored") + ctx.record("append", read(path)) fd = os.open(path, os.O_WRONLY | os.O_TRUNC) os.close(fd) check(path.stat().st_size == 0, "O_TRUNC did not truncate") + ctx.stat("truncated", path) -def test_positioned_and_vectored_io(root: Path) -> None: +def test_positioned_and_vectored_io(ctx: Context) -> None: + root = ctx.root path = root / "positioned.bin" base = bytes(range(256)) * 32768 write_durable(path, base) @@ -88,10 +343,14 @@ def test_positioned_and_vectored_io(root: Path) -> None: expected[65531:65543] = b"vector-write" got = read(path) check(got == expected, "positioned write damaged surrounding bytes") - print(f" sha256 {hashlib.sha256(got).hexdigest()}") + digest = hashlib.sha256(got).hexdigest() + ctx.record("digest", digest) + ctx.stat("file", path) + print(f" sha256 {digest}") -def test_resize_and_sparse_io(root: Path) -> None: +def test_resize_and_sparse_io(ctx: Context) -> None: + root = ctx.root path = root / "resize" write_durable(path, b"0123456789") os.truncate(path, 4) @@ -99,6 +358,7 @@ def test_resize_and_sparse_io(root: Path) -> None: os.truncate(path, 8193) data = read(path) check(data[:4] == b"0123" and data[4:] == bytes(8189), "grown range is not zero-filled") + ctx.record("grown.digest", hashlib.sha256(data).hexdigest()) sparse = root / "sparse" fd = os.open(sparse, os.O_CREAT | os.O_RDWR, 0o600) @@ -116,9 +376,11 @@ def test_resize_and_sparse_io(root: Path) -> None: check(os.pread(fd, 1, sparse.stat().st_size + 4096) == b"", "read beyond EOF not empty") finally: os.close(fd) + ctx.stat("sparse", sparse) -def test_mmap_and_copy_paths(root: Path) -> None: +def test_mmap_and_copy_paths(ctx: Context) -> None: + root = ctx.root source = root / "mapped" data = bytearray((b"mmap-and-sendfile\0" * 65536)[:1024 * 1024]) write_durable(source, data) @@ -147,11 +409,34 @@ def test_mmap_and_copy_paths(root: Path) -> None: os.close(src) os.close(dst) check(read(target) == data, "sendfile copy mismatch") + ctx.record("digest", hashlib.sha256(read(target)).hexdigest()) + + # coreutils reaches for copy_file_range before read/write, so a filesystem + # that neither implements it nor refuses it cleanly breaks plain `cp`. + ranged = root / "copy-file-range" + src = os.open(source, os.O_RDONLY) + dst = os.open(ranged, os.O_CREAT | os.O_WRONLY | os.O_TRUNC, 0o600) + try: + copied = ctx.probe( + "copy_file_range", + lambda: os.copy_file_range(src, dst, len(data)), + allow=CLEAN_UNSUPPORTED | {errno.EXDEV}, + ) + finally: + os.close(src) + os.close(dst) + if copied.ok and copied.value: + check( + read(ranged)[: copied.value] == data[: copied.value], + "copy_file_range produced wrong bytes", + ) -def test_namespace_and_errors(root: Path) -> None: +def test_namespace_and_errors(ctx: Context) -> None: + root = ctx.root a, b = root / "dir-a", root / "dir-b" - a.mkdir(); b.mkdir() + a.mkdir() + b.mkdir() (a / "nested").mkdir() write_durable(a / "nested" / "child", b"child") write_durable(root / "source", b"source") @@ -163,6 +448,7 @@ def test_namespace_and_errors(root: Path) -> None: check("moved" in os.listdir(b), "cross-directory move missing from readdir") os.rename(b / "moved", b / "moved") check(read(b / "moved") == b"source", "same-name rename changed data") + ctx.record("moved.bytes", read(b / "moved")) expect_errno({errno.ENOTEMPTY, errno.EEXIST}, lambda: os.rmdir(a), "rmdir(non-empty)") expect_errno({errno.ENOTDIR}, lambda: os.rmdir(b / "moved"), "rmdir(file)") @@ -172,17 +458,39 @@ def test_namespace_and_errors(root: Path) -> None: expect_errno({errno.ENOENT}, lambda: os.rename(root / "absent", root / "new"), "rename(absent)") empty_src, empty_dst = root / "empty-src", root / "empty-dst" - empty_src.mkdir(); empty_dst.mkdir() + empty_src.mkdir() + empty_dst.mkdir() os.replace(empty_src, empty_dst) check(empty_dst.is_dir() and not empty_src.exists(), "empty directory replacement failed") nonempty = root / "nonempty-dst" - nonempty.mkdir(); write_durable(nonempty / "keep", b"keep") - replacement = root / "replacement-dir"; replacement.mkdir() - expect_errno({errno.ENOTEMPTY, errno.EEXIST}, lambda: os.replace(replacement, nonempty), "replace non-empty dir") + nonempty.mkdir() + write_durable(nonempty / "keep", b"keep") + replacement = root / "replacement-dir" + replacement.mkdir() + expect_errno( + {errno.ENOTEMPTY, errno.EEXIST}, + lambda: os.replace(replacement, nonempty), + "replace non-empty dir", + ) check(read(nonempty / "keep") == b"keep", "failed replacement damaged destination") + # renameat2 flags: git and dpkg use NOREPLACE, and a filesystem that ignores + # the flag instead of refusing it turns a guarded rename into a clobber. + guard_src, guard_dst = root / "guard-src", root / "guard-dst" + write_durable(guard_src, b"guard-src") + write_durable(guard_dst, b"guard-dst") + guarded = ctx.probe( + "renameat2.noreplace", + lambda: renameat2(guard_src, guard_dst, RENAME_NOREPLACE), + allow=CLEAN_UNSUPPORTED | {errno.EEXIST}, + ) + check(not guarded.ok, "RENAME_NOREPLACE overwrote an existing destination") + check(read(guard_dst) == b"guard-dst", "RENAME_NOREPLACE damaged the destination") + ctx.probe("renameat2.exchange", lambda: renameat2(guard_src, guard_dst, RENAME_EXCHANGE)) -def test_open_lifetime(root: Path) -> None: + +def test_open_lifetime(ctx: Context) -> None: + root = ctx.root path = root / "open-unlink" data = b"held-open\0" * 4096 write_durable(path, data) @@ -199,11 +507,14 @@ def test_open_lifetime(root: Path) -> None: fd = os.open(old, os.O_RDWR) os.rename(old, new) os.pwrite(fd, b"after!", 0) - os.fsync(fd); os.close(fd) + os.fsync(fd) + os.close(fd) check(not old.exists() and read(new) == b"after!", "open handle did not follow rename") + ctx.record("renamed.bytes", read(new)) -def test_names_and_enumeration(root: Path) -> None: +def test_names_and_enumeration(ctx: Context) -> None: + root = ctx.root names = ["space name.txt", "unicodé-文件", ".hidden", "trailing.dot.", "case", "CASE"] for index, name in enumerate(names): write_durable(root / name, f"name-{index}".encode()) @@ -211,10 +522,301 @@ def test_names_and_enumeration(root: Path) -> None: check(set(names) <= listed, "directory enumeration omitted valid names") for name in listed: os.lstat(root / name) + ctx.record("case_sensitive", read(root / "case") != read(root / "CASE")) + ctx.record("names", sorted(name for name in names)) expect_errno({errno.ENAMETOOLONG}, lambda: os.open(root / ("x" * 256), os.O_CREAT), "overlong name") -def test_concurrency(root: Path) -> None: +def test_readdir_stability(ctx: Context) -> None: + """Enumeration must not duplicate or lose entries when the directory moves. + + readdir is served off the dispatch loop, so a pass that re-reads the child + list between two kernel-visible offsets can double-report or skip an entry. + A lister that sees one name twice deletes it twice. + """ + root = ctx.root + directory = root / "readdir-churn" + directory.mkdir() + stable = [f"stable-{index:03d}" for index in range(64)] + for name in stable: + write_durable(directory / name, b"x") + + seen: list[str] = [] + created: list[str] = [] + with os.scandir(directory) as entries: + for position, entry in enumerate(entries): + seen.append(entry.name) + # Mutate mid-enumeration: churn is the point. + if position == 8: + for index in range(4): + name = f"late-{index}" + write_durable(directory / name, b"y") + created.append(name) + if position == 24: + os.unlink(directory / stable[-1]) + + check(len(seen) == len(set(seen)), f"readdir reported duplicates: {sorted(seen)}") + survivors = set(os.listdir(directory)) + check(set(created) <= survivors, "entries created during enumeration were lost") + check(stable[-1] not in survivors, "entry unlinked during enumeration reappeared") + # An entry that existed unchanged for the whole pass must be reported once. + missing = set(stable[:-1]) - set(seen) + check(not missing, f"stable entries vanished from readdir: {sorted(missing)[:10]}") + ctx.record("no_duplicates", True) + ctx.record("survivors", len(survivors)) + + # rewinddir: a second full pass must agree with the settled directory. + check(set(os.listdir(directory)) == survivors, "second enumeration disagreed with the first") + + +def test_metadata_updates(ctx: Context) -> None: + """chmod/utimes must succeed even where they cannot be stored. + + Drive has no POSIX mode or owner, and `setattr` deliberately accepts and + ignores them. That is only correct if it reports success: `cp -p`, `tar -x` + and `rsync -a` all treat a refused chmod as a fatal error. + """ + root = ctx.root + path = root / "metadata" + write_durable(path, b"metadata") + + os.chmod(path, 0o640) + ctx.note("mode.after_chmod", stat.S_IMODE(os.lstat(path).st_mode)) + os.utime(path, (1_600_000_000, 1_600_000_123)) + ctx.note("mtime.after_utime", int(os.lstat(path).st_mtime)) + ctx.record("chmod_and_utime_succeeded", True) + + directory = root / "metadata-dir" + directory.mkdir() + os.chmod(directory, 0o750) + os.utime(directory, None) + + # A path-based truncate with no open handle is a separate write path (a + # shell's `> file`), and it is the one that has to work offline. + write_durable(path, b"0123456789") + os.truncate(path, 0) + check(read(path) == b"", "closed-file truncate to zero did not empty the file") + ctx.record("closed_truncate.size", os.lstat(path).st_size) + + with path.open("wb", buffering=0) as handle: + handle.write(b"redirect") + check(read(path) == b"redirect", "reopen-and-write after truncate lost bytes") + ctx.record("redirect.bytes", read(path)) + + +def test_extended_attributes(ctx: Context) -> None: + """The thumbnail xattr interface, including its two-call size protocol.""" + root = ctx.root + path = root / "attributes.txt" + write_durable(path, b"attributes") + + expect_errno( + {errno.ENODATA, errno.ENOATTR} if hasattr(errno, "ENOATTR") else {errno.ENODATA}, + lambda: os.getxattr(path, "user.proton.definitely-absent"), + "getxattr(unknown name)", + ) + ctx.probe("setxattr", lambda: os.setxattr(path, "user.pdfs.probe", b"v")) + + listed = os.listxattr(path) + ctx.note("listxattr", sorted(listed)) + # Thumbnails exist only for images and video; advertising them on a text + # file is what made `ls -l` issue a network round trip per file (B5). + check( + "user.proton.thumbnail" not in listed and "user.proton.preview" not in listed, + f"a text file advertised thumbnail attributes: {listed}", + ) + ctx.record("no_thumbnail_on_text", True) + + if not ctx.is_live: + return + # Opportunistic: if the mount holds a real image, exercise the size protocol + # against it. Nothing here creates one, because only the server can. + for candidate in _find_thumbnailable(ctx.root.parent): + names = os.listxattr(candidate) + if "user.proton.thumbnail" not in names: + continue + size = _xattr_size(candidate, "user.proton.thumbnail") + if size == 0: + continue + expect_errno( + {errno.ERANGE}, + lambda: _xattr_into(candidate, "user.proton.thumbnail", size - 1), + "getxattr(undersized buffer)", + ) + payload = os.getxattr(candidate, "user.proton.thumbnail") + check(len(payload) == size, "thumbnail length disagreed with its size probe") + ctx.note("thumbnail.bytes", size) + return + ctx.note("thumbnail.bytes", "no thumbnailable file found") + + +def _find_thumbnailable(directory: Path, limit: int = 200): + suffixes = {".jpg", ".jpeg", ".png", ".heic", ".mp4", ".mov", ".webp", ".gif"} + seen = 0 + try: + entries = sorted(directory.iterdir()) + except OSError: + return + for entry in entries: + if seen >= limit: + return + seen += 1 + if entry.is_file() and entry.suffix.lower() in suffixes: + yield entry + + +def _xattr_size(path: Path, name: str) -> int: + fd = os.open(path, os.O_RDONLY) + try: + return len(os.getxattr(fd, name)) + except OSError: + return 0 + finally: + os.close(fd) + + +def _xattr_into(path: Path, name: str, size: int) -> bytes: + """Force the undersized-buffer branch that Python's helper hides.""" + buffer = ctypes.create_string_buffer(max(size, 1)) + _libc.getxattr.argtypes = [ + ctypes.c_char_p, + ctypes.c_char_p, + ctypes.c_void_p, + ctypes.c_size_t, + ] + _libc.getxattr.restype = ctypes.c_long + ctypes.set_errno(0) + written = _libc.getxattr(os.fsencode(path), name.encode(), buffer, ctypes.c_size_t(size)) + if written < 0: + code = ctypes.get_errno() + raise OSError(code, os.strerror(code)) + return buffer.raw[:written] + + +def test_allocation_and_holes(ctx: Context) -> None: + """fallocate, including the punched hole that must survive the upload. + + A punched range reads as zero. The commit path gap-fills unauthored ranges + from the remote baseline, so if the hole is not claimed as authored the + original bytes come back and the hole silently disappears after close. + """ + root = ctx.root + path = root / "allocated" + write_durable(path, b"A" * (256 * 1024)) + + fd = os.open(path, os.O_RDWR) + try: + ctx.probe("posix_fallocate", lambda: os.posix_fallocate(fd, 0, 512 * 1024)) + os.fsync(fd) + finally: + os.close(fd) + ctx.note("allocated.size", os.lstat(path).st_size) + + holed = root / "punched" + write_durable(holed, b"B" * (192 * 1024)) + fd = os.open(holed, os.O_RDWR) + punched = False + try: + try: + fallocate(fd, FALLOC_FL_PUNCH_HOLE | FALLOC_FL_KEEP_SIZE, 64 * 1024, 64 * 1024) + punched = True + except OSError as error: + check( + error.errno in CLEAN_UNSUPPORTED, + f"punch hole refused with errno {error.errno}", + ) + ctx.note("punch_hole", f"errno:{errno.errorcode.get(error.errno)}") + if punched: + ctx.note("punch_hole", "ok") + check( + os.pread(fd, 64 * 1024, 64 * 1024) == bytes(64 * 1024), + "punched range did not read as zeros", + ) + os.fsync(fd) + finally: + os.close(fd) + + if punched: + check(os.lstat(holed).st_size == 192 * 1024, "punch hole with KEEP_SIZE changed the size") + expected = b"B" * (64 * 1024) + bytes(64 * 1024) + b"B" * (64 * 1024) + # Reopening is the assertion that matters: it is where a gap-fill from + # the remote baseline would undo the hole. + check(read(holed) == expected, "punched hole did not survive close and reopen") + # A note, not a comparison: whether the filesystem can punch at all is a + # capability. That it produced the right bytes when it did is asserted + # above, where the answer does not depend on the reference having agreed + # to punch too. + ctx.note("punched.digest", hashlib.sha256(read(holed)).hexdigest()) + + # tar and `cp --sparse` navigate with SEEK_HOLE/SEEK_DATA. + fd = os.open(holed, os.O_RDONLY) + try: + ctx.probe("seek_data", lambda: os.lseek(fd, 0, os.SEEK_DATA), allow=CLEAN_UNSUPPORTED | {errno.ENXIO}) + ctx.probe("seek_hole", lambda: os.lseek(fd, 0, os.SEEK_HOLE), allow=CLEAN_UNSUPPORTED | {errno.ENXIO}) + finally: + os.close(fd) + + +def test_unsupported_operations(ctx: Context) -> None: + """Operations the filesystem does not implement must fail cleanly, not hang. + + Every one of these is probed by ordinary tools — git, rsync, `cp -a`, `df`, + installers. An unimplemented FUSE handler that returns EIO, or blocks, turns + a graceful fallback into a hard failure in software that never asked for the + feature in the first place. + """ + root = ctx.root + target = root / "link-target" + write_durable(target, b"link-target") + + ctx.probe("symlink", lambda: os.symlink("link-target", root / "symlink")) + if (root / "symlink").is_symlink(): + check(os.readlink(root / "symlink") == "link-target", "readlink returned the wrong target") + ctx.probe("hardlink", lambda: os.link(target, root / "hardlink")) + ctx.probe("mkfifo", lambda: os.mkfifo(root / "fifo")) + ctx.probe("mknod", lambda: os.mknod(root / "node", 0o600 | stat.S_IFCHR, os.makedev(1, 3))) + + # df and any installer that checks free space call statfs. It must answer. + info = os.statvfs(root) + check(info.f_bsize > 0, "statfs reported a zero block size") + ctx.note("statfs.bsize", info.f_bsize) + ctx.note("statfs.blocks", info.f_blocks) + ctx.record("statfs_answers", True) + + check(os.access(root, os.R_OK | os.X_OK), "access(2) denied read/execute on the test root") + check(os.access(target, os.R_OK), "access(2) denied read on a readable file") + + fd = os.open(target, os.O_RDWR) + try: + ctx.probe("flock", lambda: fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)) + with contextlib.suppress(OSError): + fcntl.flock(fd, fcntl.LOCK_UN) + ctx.probe("posix_lock", lambda: fcntl.lockf(fd, fcntl.LOCK_EX | fcntl.LOCK_NB, 0, 1)) + with contextlib.suppress(OSError): + fcntl.lockf(fd, fcntl.LOCK_UN, 0, 1) + finally: + os.close(fd) + + direct = ctx.probe("o_direct", lambda: os.open(target, os.O_RDONLY | os.O_DIRECT)) + if direct.ok: + with contextlib.suppress(OSError): + # O_DIRECT demands an aligned buffer; mmap gives a page-aligned one. + with mmap.mmap(-1, 4096) as buffer: + os.preadv(direct.value, [buffer], 0) + os.close(direct.value) + + # TCGETS against a filesystem must answer ENOTTY, which is what stops fuser + # from logging a warning every time a program probes whether an fd is a tty. + fd = os.open(root, os.O_RDONLY) + try: + ctx.probe("ioctl_tcgets", lambda: fcntl.ioctl(fd, 0x5401, b"\0" * 64)) + finally: + os.close(fd) + + +def test_concurrency(ctx: Context) -> None: + root = ctx.root + def independent(index: int) -> None: data = bytes([index]) * (131072 + index) path = root / f"concurrent-{index}" @@ -229,9 +831,11 @@ def test_concurrency(root: Path) -> None: write_durable(shared, bytes(extent * 8)) fd = os.open(shared, os.O_RDWR) try: + def positioned(index: int) -> None: payload = bytes([index + 1]) * extent check(os.pwrite(fd, payload, index * extent) == extent, f"short concurrent pwrite {index}") + with concurrent.futures.ThreadPoolExecutor(max_workers=8) as pool: list(pool.map(positioned, range(8))) os.fsync(fd) @@ -239,39 +843,668 @@ def test_concurrency(root: Path) -> None: os.close(fd) expected = b"".join(bytes([i + 1]) * extent for i in range(8)) check(read(shared) == expected, "concurrent disjoint writes overlapped or vanished") + ctx.record("shared.digest", hashlib.sha256(read(shared)).hexdigest()) + + +def test_application_workloads(ctx: Context) -> None: + """Real tools, because they combine syscalls in ways a matrix will not. + + Each of these has broken on a network filesystem before: the editor save + dance (write temp, fsync, rename over), git's lock files and many small + objects, tar's metadata restore, sqlite's journal and locking. + """ + root = ctx.root + workloads = root / "workloads" + workloads.mkdir() + + # The atomic-save cycle every editor performs, repeated so a stale inode or + # a mishandled replace shows up rather than passing once by luck. + document = workloads / "document.txt" + for revision in range(5): + temporary = workloads / f".document.txt.{revision}.tmp" + payload = f"revision {revision}\n".encode() * 128 + write_durable(temporary, payload) + os.replace(temporary, document) + check(read(document) == payload, f"atomic save lost revision {revision}") + ctx.record("atomic_save.bytes", read(document)) + + # tar restores mode and mtime on extract; a filesystem that refuses either + # makes every extraction fail loudly. + tree = workloads / "tree" + (tree / "nested").mkdir(parents=True) + write_durable(tree / "nested" / "leaf", b"leaf" * 512) + write_durable(tree / "top", b"top") + archive = workloads / "tree.tar" + with tarfile.open(archive, "w") as bundle: + bundle.add(tree, arcname="tree") + extracted = workloads / "extracted" + extracted.mkdir() + with tarfile.open(archive) as bundle: + try: + bundle.extractall(extracted, filter="data") + except TypeError: # the extraction filter predates Python 3.12 + bundle.extractall(extracted) + check( + read(extracted / "tree" / "nested" / "leaf") == b"leaf" * 512, + "tar extraction produced wrong bytes", + ) + ctx.record("tar.leaf_size", os.lstat(extracted / "tree" / "nested" / "leaf").st_size) + + # sqlite exercises locking, journal creation and unlink, and fsync ordering. + database = workloads / "workload.db" + connection = sqlite3.connect(database) + try: + connection.execute("CREATE TABLE rows (id INTEGER PRIMARY KEY, value TEXT)") + with connection: + connection.executemany( + "INSERT INTO rows (value) VALUES (?)", [(f"row-{index}",) for index in range(500)] + ) + connection.execute("PRAGMA wal_checkpoint(TRUNCATE)") + count = connection.execute("SELECT COUNT(*) FROM rows").fetchone()[0] + finally: + connection.close() + check(count == 500, f"sqlite lost rows: {count}") + ctx.record("sqlite.rows", count) + connection = sqlite3.connect(database) + try: + check( + connection.execute("PRAGMA integrity_check").fetchone()[0] == "ok", + "sqlite integrity check failed after reopen", + ) + finally: + connection.close() + ctx.record("sqlite.integrity", "ok") + + if shutil.which("git"): + repository = workloads / "repository" + repository.mkdir() + _run_tool(["git", "init", "--quiet", "."], repository) + _run_tool(["git", "config", "user.email", "acceptance@example.invalid"], repository) + _run_tool(["git", "config", "user.name", "pdfs acceptance"], repository) + for index in range(3): + write_durable(repository / f"file-{index}", f"content {index}\n".encode() * 64) + _run_tool(["git", "add", "-A"], repository) + _run_tool(["git", "commit", "--quiet", "-m", f"commit {index}"], repository) + status = _run_tool(["git", "status", "--porcelain"], repository) + check(status.strip() == "", f"git saw a dirty tree after committing: {status!r}") + _run_tool(["git", "fsck", "--no-progress"], repository) + ctx.record("git.clean_status", True) + else: + ctx.note("git", "not installed") + + if shutil.which("rsync"): + mirror = workloads / "rsync-mirror" + mirror.mkdir() + _run_tool(["rsync", "-a", f"{tree}/", f"{mirror}/"], workloads) + check( + read(mirror / "nested" / "leaf") == b"leaf" * 512, + "rsync mirrored the wrong bytes", + ) + ctx.record("rsync.leaf_size", os.lstat(mirror / "nested" / "leaf").st_size) + else: + ctx.note("rsync", "not installed") + + +def _run_tool(command: list[str], cwd: Path) -> str: + result = subprocess.run(command, cwd=cwd, text=True, capture_output=True) + if result.returncode: + detail = (result.stderr or result.stdout).strip() + raise AssertionError(f"{' '.join(command)} failed in {cwd}: {detail}") + return result.stdout + + +# -------------------------------------------------------------------------- +# Regressions from docs/BUGS.md +# -------------------------------------------------------------------------- + + +def test_regression_b7_renamed_directory(ctx: Context) -> None: + """A renamed directory must stay traversable (B7). + + The reported failure was a directory that appeared in `ls` but could not be + entered, because rename replaced the inode the kernel still had cached. + """ + root = ctx.root + original = root / "b7 original dir" + original.mkdir() + write_durable(original / "child file.txt", b"b7-child") + (original / "sub dir").mkdir() + write_durable(original / "sub dir" / "deep", b"b7-deep") + + before = os.lstat(original).st_ino + renamed = root / "b7 renamed dïr ünï" + os.rename(original, renamed) + + check(renamed.is_dir(), "renamed directory is not a directory") + check(set(os.listdir(renamed)) == {"child file.txt", "sub dir"}, "renamed directory lost children") + check(read(renamed / "child file.txt") == b"b7-child", "child unreadable after directory rename") + check(read(renamed / "sub dir" / "deep") == b"b7-deep", "grandchild unreadable after rename") + ctx.record("children", sorted(os.listdir(renamed))) + ctx.note("inode_stable", os.lstat(renamed).st_ino == before) + + # Writing through the new path proves the entry is not merely readable. + write_durable(renamed / "child file.txt", b"b7-rewritten") + check(read(renamed / "child file.txt") == b"b7-rewritten", "write through renamed directory lost") + + +def test_regression_b69_identical_rewrite(ctx: Context) -> None: + """Rewriting identical bytes must not fork a `(sync-conflict …)` copy (B69). + + Revision identity used to be `(mtime, size)`, so a rewrite that changed the + mtime looked like a divergent revision and the sync engine kept both. + """ + daemon = ctx.require_daemon() + root = ctx.root + path = root / "b69-identical.bin" + payload = b"b69-identical-payload\n" * 4096 + + write_durable(path, payload) + daemon.wait_for_queue() + time.sleep(1) + write_durable(path, payload) + daemon.wait_for_queue() + + conflicts = _conflict_copies(root) + check(not conflicts, f"identical rewrite produced conflict copies: {conflicts}") + check_bytes(read(path), payload, "identical rewrite changed the file contents") + ctx.record("conflicts", []) + + +def test_regression_b70_transient_download_name(ctx: Context) -> None: + """An in-flight browser download must not be sealed as a revision (B70). + + A `.crdownload` is a partial file the browser renames on completion. Sealing + it produced an upload of the partial bytes and then a conflict fork against + the finished name. + """ + daemon = ctx.require_daemon() + root = ctx.root + partial = root / "b70-download.zip.crdownload" + final = root / "b70-download.zip" + payload = b"b70-final-payload\n" * 8192 + + # Write the partial in chunks with a pause, the way a download lands. + fd = os.open(partial, os.O_CREAT | os.O_WRONLY | os.O_TRUNC, 0o600) + try: + half = len(payload) // 2 + os.write(fd, payload[:half]) + os.fsync(fd) + time.sleep(2) + os.write(fd, payload[half:]) + os.fsync(fd) + finally: + os.close(fd) + os.rename(partial, final) + daemon.wait_for_queue() + + listing = set(os.listdir(root)) + check(final.name in listing, "completed download is missing after rename") + check(partial.name not in listing, "transient download name survived the rename") + conflicts = _conflict_copies(root) + check(not conflicts, f"download rename produced conflict copies: {conflicts}") + check_bytes(read(final), payload, "completed download has the wrong bytes") + ctx.record("final.digest", hashlib.sha256(read(final)).hexdigest()) + ctx.record("conflicts", []) + + +def test_regression_b74_rename_after_close(ctx: Context) -> None: + """Write, close, rename: the content must reach Drive (B74). + + Not a `.crdownload` and not an overwrite — the plain temp-file dance that + every editor, browser, `rsync` and `git` performs. Renaming a file whose + create is still queued lost the staged bytes and published an empty node, + so this is the narrowest reproduction of that loss. + """ + daemon = ctx.require_daemon() + root = ctx.root + payload = b"b74-rename-after-close\n" * 6144 + + temporary = root / "b74-staging.tmp" + final = root / "b74-final.bin" + write_durable(temporary, payload) + os.rename(temporary, final) + daemon.wait_for_queue() + # The queue reports empty the moment the create lands; the revision that + # carries the bytes can still be in flight behind it. + time.sleep(5) + daemon.wait_for_queue() + + check(final.is_file(), "renamed file is missing after the queue drained") + check_bytes(read(final), payload, "renamed file lost its content") + check(not _conflict_copies(root), "rename after close produced conflict copies") + ctx.record("digest", hashlib.sha256(read(final)).hexdigest()) + + +def _conflict_copies(root: Path) -> list[str]: + return sorted(name for name in os.listdir(root) if "(sync-conflict" in name) + + +def test_durability_across_restart(ctx: Context) -> None: + """Written bytes must survive a daemon restart (staging/ and recovery/). + + Those directories can hold the only copy of a file whose upload has not + finished. This is the one case that proves the queue is persistent rather + than merely in-memory. + """ + daemon = ctx.require_daemon() + if not daemon.may_restart: + raise Skip("restart drill not requested; pass --durability") + root = ctx.root + path = root / "durability.bin" + payload = hashlib.sha256(b"durability").digest() * 8192 + write_durable(path, payload) + daemon.wait_for_queue() + + mountpoint = daemon.enclosing_mount(root) + daemon.restart() + daemon.wait_for_mount(mountpoint) + daemon.wait_for_queue() + + check(path.is_file(), "file vanished across a daemon restart") + check(read(path) == payload, "file contents changed across a daemon restart") + check(not _conflict_copies(root), "restart produced conflict copies") + ctx.record("digest", hashlib.sha256(read(path)).hexdigest()) + + +class Case: + def __init__(self, name: str, run, kinds: tuple[str, ...] = (REFERENCE, LIVE)) -> None: + self.name = name + self.run = run + self.kinds = kinds TESTS = [ - ("creation and open flags", test_create_flags), - ("positioned and vectored I/O", test_positioned_and_vectored_io), - ("truncate, growth, sparse ranges, and EOF", test_resize_and_sparse_io), - ("mmap and zero-copy copy paths", test_mmap_and_copy_paths), - ("namespace operations and POSIX errors", test_namespace_and_errors), - ("open-handle lifetime across unlink and rename", test_open_lifetime), - ("names, lookup, stat, and enumeration", test_names_and_enumeration), - ("independent and shared-file concurrency", test_concurrency), + Case("creation and open flags", test_create_flags), + Case("positioned and vectored I/O", test_positioned_and_vectored_io), + Case("truncate, growth, sparse ranges, and EOF", test_resize_and_sparse_io), + Case("mmap and zero-copy copy paths", test_mmap_and_copy_paths), + Case("namespace operations and POSIX errors", test_namespace_and_errors), + Case("open-handle lifetime across unlink and rename", test_open_lifetime), + Case("names, lookup, stat, and enumeration", test_names_and_enumeration), + Case("readdir stability under concurrent mutation", test_readdir_stability), + Case("metadata updates and path-based truncate", test_metadata_updates), + Case("extended attributes and the thumbnail interface", test_extended_attributes), + Case("allocation, punched holes, and hole-seeking", test_allocation_and_holes), + Case("unsupported operations refuse cleanly", test_unsupported_operations), + Case("independent and shared-file concurrency", test_concurrency), + Case("application workloads (editor, tar, sqlite, git, rsync)", test_application_workloads), + Case("regression B7: renamed directory stays traversable", test_regression_b7_renamed_directory), + Case("regression B69: identical rewrite makes no conflict", test_regression_b69_identical_rewrite, (LIVE,)), + Case("regression B70: transient download name is not sealed", test_regression_b70_transient_download_name, (LIVE,)), + Case("regression B74: rename after close keeps its content", test_regression_b74_rename_after_close, (LIVE,)), + Case("durability across a daemon restart", test_durability_across_restart, (LIVE,)), ] -def run_contract(parent: Path, label: str) -> tuple[Path, str]: +# -------------------------------------------------------------------------- +# Runner +# -------------------------------------------------------------------------- + + +class Result: + def __init__(self, name: str, target: str, status: str, seconds: float, message: str = "") -> None: + self.name = name + self.target = target + self.status = status + self.seconds = seconds + self.message = message + + def as_dict(self) -> dict: + return { + "name": self.name, + "target": self.target, + "status": self.status, + "seconds": round(self.seconds, 3), + "message": self.message, + } + + +class Run: + def __init__(self, label: str, kind: str, root: Path) -> None: + self.label = label + self.kind = kind + self.root = root + self.results: list[Result] = [] + self.compared: dict[str, object] = {} + self.noted: dict[str, object] = {} + self.ran: set[str] = set() + self.digest = "" + + @property + def failed(self) -> bool: + return any(result.status in {"fail", "timeout"} for result in self.results) + + +REPORT: list[Run] = [] + + +def select_cases(kind: str) -> list[Case]: + selected = os.environ.get("PDFS_ACCEPTANCE_ONLY") + cases = [case for case in TESTS if kind in case.kinds] + if selected: + cases = [case for case in cases if selected.lower() in case.name.lower()] + check(bool(cases), f"PDFS_ACCEPTANCE_ONLY={selected!r} matched no tests for {kind}") + return cases + + +def reap_stale_roots(parent: Path, max_age: int) -> None: + """Remove abandoned roots from crashed runs so preconditions stay meaningful. + + Only exact `pdfs-acceptance-<32 hex>` names are eligible, and only once they + are older than the age cutoff, so a concurrent run's tree is never touched. + """ + if max_age <= 0: + return + cutoff = time.time() - max_age + try: + entries = list(parent.iterdir()) + except OSError as error: + print(f"WARNING: could not scan {parent} for stale roots: {error}") + return + for entry in entries: + if not STALE_ROOT.match(entry.name): + continue + try: + if not entry.is_dir() or entry.stat().st_mtime > cutoff: + continue + except OSError: + continue + print(f"[reap] removing abandoned acceptance root {entry}") + shutil.rmtree(entry, ignore_errors=True) + + +def run_contract( + parent: Path, + label: str, + kind: str = REFERENCE, + daemon=None, + timeout: int = DEFAULT_TIMEOUT, + fail_fast: bool = False, +) -> Run: + reap_stale_roots(parent, int(os.environ.get("PDFS_ACCEPTANCE_REAP_AGE", "3600"))) root = parent / f"pdfs-acceptance-{uuid.uuid4().hex}" root.mkdir() + run = Run(label, kind, root) + REPORT.append(run) print(f"[target] {label}: {parent}") + obs = Observations() try: - selected = os.environ.get("PDFS_ACCEPTANCE_ONLY") - tests = TESTS if not selected else [test for test in TESTS if selected.lower() in test[0].lower()] - check(bool(tests), f"PDFS_ACCEPTANCE_ONLY={selected!r} matched no tests") - for name, operation in tests: - print(f" [test] {name}") - operation(root) + for case in select_cases(kind): + obs.scope(case.name) + context = Context(root, obs, kind, daemon) + started = time.monotonic() + print(f" [test] {case.name}") + try: + with time_limit(timeout, case.name): + case.run(context) + except Skip as reason: + elapsed = time.monotonic() - started + print(f" SKIP {reason}") + run.results.append(Result(case.name, label, "skip", elapsed, str(reason))) + continue + except TestTimeout as reason: + elapsed = time.monotonic() - started + print(f" TIMEOUT {reason}") + run.results.append(Result(case.name, label, "timeout", elapsed, str(reason))) + # A timeout means the mount may be wedged; further cases would + # only produce noise, and cleanup already has to fight for it. + break + except BaseException as error: # noqa: BLE001 - reported, then continued + elapsed = time.monotonic() - started + detail = f"{type(error).__name__}: {error}" + print(f" FAIL {detail}") + run.results.append(Result(case.name, label, "fail", elapsed, detail)) + if fail_fast: + raise + continue + elapsed = time.monotonic() - started + print(f" ok {elapsed:.2f}s") + run.results.append(Result(case.name, label, "pass", elapsed)) + run.ran.add(case.name) + run.compared = dict(obs.compared) + run.noted = dict(obs.noted) digest_path = root / "positioned.bin" - digest = hashlib.sha256(read(digest_path)).hexdigest() if digest_path.exists() else "" - return root, digest + if digest_path.exists(): + run.digest = hashlib.sha256(read(digest_path)).hexdigest() + return run except BaseException: shutil.rmtree(root, ignore_errors=True) raise +def compare_runs(reference: Run, target: Run) -> list[str]: + """Diff the mount's recorded semantics against the local filesystem's. + + Only keys from cases that passed on both sides are compared: a case that + failed or was skipped has already been reported, and its partial + observations would just repeat that failure in a less useful form. + """ + shared = reference.ran & target.ran + divergences: list[str] = [] + for key, expected in reference.compared.items(): + case = key.split(".", 1)[0] + if case not in shared: + continue + if key not in target.compared: + divergences.append(f"{key}: missing on {target.label}") + continue + actual = target.compared[key] + if actual != expected: + divergences.append(f"{key}: {target.label}={actual!r} reference={expected!r}") + for key in target.compared: + case = key.split(".", 1)[0] + if case in shared and key not in reference.compared: + divergences.append(f"{key}: present on {target.label}, absent from reference") + return divergences + + +def note_capability_differences(reference: Run, target: Run) -> list[str]: + lines = [] + for key, expected in reference.noted.items(): + if key in target.noted and target.noted[key] != expected: + lines.append(f"{key}: {target.label}={target.noted[key]!r} reference={expected!r}") + return lines + + +def write_reports(json_path: Path | None, junit_path: Path | None) -> None: + if json_path: + payload = { + "runs": [ + { + "label": run.label, + "kind": run.kind, + "results": [result.as_dict() for result in run.results], + "observations": { + key: _jsonable(value) for key, value in sorted(run.compared.items()) + }, + "notes": {key: _jsonable(value) for key, value in sorted(run.noted.items())}, + } + for run in REPORT + ] + } + json_path.write_text(json.dumps(payload, indent=2, sort_keys=True)) + print(f"[report] wrote {json_path}") + if junit_path: + suites = ElementTree.Element("testsuites") + for run in REPORT: + suite = ElementTree.SubElement( + suites, + "testsuite", + name=run.label, + tests=str(len(run.results)), + failures=str(sum(1 for r in run.results if r.status in {"fail", "timeout"})), + skipped=str(sum(1 for r in run.results if r.status == "skip")), + time=f"{sum(r.seconds for r in run.results):.3f}", + ) + for result in run.results: + case = ElementTree.SubElement( + suite, + "testcase", + name=result.name, + classname=run.label, + time=f"{result.seconds:.3f}", + ) + if result.status in {"fail", "timeout"}: + failure = ElementTree.SubElement(case, "failure", type=result.status) + failure.text = result.message + elif result.status == "skip": + ElementTree.SubElement(case, "skipped", message=result.message) + ElementTree.ElementTree(suites).write(junit_path, encoding="unicode", xml_declaration=True) + print(f"[report] wrote {junit_path}") + + +def _jsonable(value): + if isinstance(value, (bytes, bytearray)): + return value.decode("utf-8", "backslashreplace") + if isinstance(value, (set, frozenset)): + return sorted(value) + return value + + +def report_timings(budget: float) -> list[str]: + slow = [] + for run in REPORT: + for result in run.results: + if result.status == "pass" and budget and result.seconds > budget: + slow.append(f"{run.label}: {result.name} took {result.seconds:.1f}s (budget {budget:.0f}s)") + return slow + + +class JournalWatch: + """Fail a run that leaves errors in the daemon's journal. + + A suite can pass every assertion while the daemon logs a stream of failures + behind it; that has happened here before, and it was only noticed by reading + the journal by hand. + """ + + def __init__(self, unit: str = "proton-drive.service") -> None: + self.unit = unit + self.since = 0.0 + + def start(self) -> None: + if shutil.which("journalctl") is None: + print("WARNING: journalctl is unavailable; skipping the journal check") + self.since = 0.0 + return + self.since = time.time() + + def errors(self) -> list[str]: + if not self.since: + return [] + result = subprocess.run( + [ + "journalctl", + "--user", + "-u", + self.unit, + "--since", + f"@{int(self.since)}", + "--no-pager", + "-o", + "cat", + ], + text=True, + capture_output=True, + ) + if result.returncode: + return [f"journalctl failed: {result.stderr.strip()}"] + # Filtering on syslog priority (`-p err`) finds nothing: the daemon writes + # tracing levels as text on stdout, which journald files at the unit's + # default priority. Match the level token instead, after stripping the + # ANSI colouring tracing emits when it thinks it has a terminal. + errors = [] + for line in result.stdout.splitlines(): + plain = ANSI.sub("", line) + if ERROR_LEVEL.search(plain): + errors.append(plain.strip()) + return errors + + +# -------------------------------------------------------------------------- +# Daemon control +# -------------------------------------------------------------------------- + + +class Daemon: + """The `pdfs` CLI, plus the unit control the durability drill needs.""" + + def __init__(self, timeout: int, may_restart: bool = False) -> None: + self.timeout = timeout + self.may_restart = may_restart + self.pdfs = os.environ.get("PDFS_ACCEPTANCE_PDFS", "pdfs") + self.unit = os.environ.get("PDFS_ACCEPTANCE_UNIT", "proton-drive.service") + + @classmethod + def discover(cls, timeout: int, may_restart: bool = False) -> Daemon | None: + daemon = cls(timeout, may_restart) + if shutil.which(daemon.pdfs) is None and not Path(daemon.pdfs).is_file(): + return None + try: + daemon.status() + except Exception: + return None + return daemon + + def command(self, *args: str, json_output: bool = False) -> str: + command = [self.pdfs] + if json_output: + command.append("--json") + command.extend(args) + result = subprocess.run(command, text=True, capture_output=True, timeout=self.timeout) + if result.returncode: + detail = result.stderr.strip() or result.stdout.strip() + raise RuntimeError(f"{' '.join(command)} failed: {detail}") + return result.stdout + + def status(self) -> dict: + return json.loads(self.command("status", json_output=True)) + + def wait_for_queue(self) -> None: + deadline = time.monotonic() + self.timeout + last = None + while time.monotonic() < deadline: + last = self.status().get("mount") or {} + if last.get("pending_uploads", 0) == 0 and last.get("pending_changes", 0) == 0: + return + time.sleep(1) + raise TimeoutError(f"daemon mutation queue did not drain: {last}") + + def enclosing_mount(self, path: Path) -> Path: + current = path.resolve() + while current != current.parent: + if os.path.ismount(current): + return current + current = current.parent + return current + + def restart(self) -> None: + print(f"[restart] systemctl --user restart {self.unit}") + result = subprocess.run( + ["systemctl", "--user", "restart", self.unit], + text=True, + capture_output=True, + timeout=self.timeout, + ) + if result.returncode: + raise RuntimeError(f"could not restart {self.unit}: {result.stderr.strip()}") + + def wait_for_mount(self, mountpoint: Path) -> None: + deadline = time.monotonic() + self.timeout + while time.monotonic() < deadline: + if os.path.ismount(mountpoint): + with contextlib.suppress(OSError): + os.listdir(mountpoint) + return + time.sleep(1) + raise TimeoutError(f"{mountpoint} did not come back within {self.timeout}s") + + +# -------------------------------------------------------------------------- +# Live and managed modes +# -------------------------------------------------------------------------- + + def wait_for_copy(root: Path, relative: Path, digest: str, timeout: int) -> None: deadline = time.monotonic() + timeout target = root / relative @@ -311,12 +1544,10 @@ class ManagedSyncPair: return result.stdout def folders(self) -> list[dict]: - import json value = json.loads(self.command("sync", "list", json_output=True)) return value["items"] def wait_for_queue(self) -> None: - import json deadline = time.monotonic() + self.timeout last = None while time.monotonic() < deadline: @@ -372,6 +1603,7 @@ class ManagedSyncPair: time.sleep(1) registered = {Path(item["local_path"]).resolve() for item in current} for path in self.paths: + reap_stale_roots(path, int(os.environ.get("PDFS_ACCEPTANCE_REAP_AGE", "3600"))) check(path.is_dir(), f"managed path is not a directory: {path}") check(not any(path.iterdir()), f"managed path is not empty: {path}") check(path not in registered, f"managed path is already registered: {path}") @@ -486,9 +1718,10 @@ def is_mountpoint(path: Path) -> bool: return os.path.ismount(path) -def run_managed_matrix(paths: list[Path]) -> None: +def run_managed_matrix(paths: list[Path], reference: Run, args) -> None: timeout = int(os.environ.get("PDFS_ACCEPTANCE_SYNC_TIMEOUT", "180")) pair = ManagedSyncPair(paths, timeout) + daemon = Daemon.discover(timeout, may_restart=args.durability) roots: list[Path] = [] try: pair.create() @@ -502,11 +1735,19 @@ def run_managed_matrix(paths: list[Path]) -> None: print(f"[matrix] first={modes[0]}, second={modes[1]}") pair.set_modes(modes) for path, mode in zip(pair.paths, modes, strict=True): - root, _ = run_contract(path, f"managed {mode}") - roots.append(root) + run = run_contract( + path, + f"managed {mode} {path.name}", + kind=LIVE, + daemon=daemon, + timeout=args.timeout, + fail_fast=args.fail_fast, + ) + roots.append(run.root) + summarize_divergence(reference, run) pair.wait_for_queue() - pair.remove_tree(root) - roots.remove(root) + pair.remove_tree(run.root) + roots.remove(run.root) pair.wait_for_queue() # Mirror changes are asynchronous; force and await a pass before # changing its mode so authored bytes/deletions cannot be lost. @@ -518,7 +1759,39 @@ def run_managed_matrix(paths: list[Path]) -> None: pair.cleanup() -def main() -> None: +def summarize_divergence(reference: Run, target: Run) -> None: + divergences = compare_runs(reference, target) + capabilities = note_capability_differences(reference, target) + if capabilities: + print(f" [capabilities] {target.label} differs from the reference filesystem:") + for line in capabilities: + print(f" - {line}") + if divergences: + print(f" [DIVERGENCE] {target.label} does not match the reference filesystem:") + for line in divergences: + print(f" ! {line}") + target.results.append( + Result( + "differential comparison against the reference filesystem", + target.label, + "fail", + 0.0, + "; ".join(divergences), + ) + ) + else: + print(f" [differential] {target.label} matches the reference filesystem") + target.results.append( + Result( + "differential comparison against the reference filesystem", + target.label, + "pass", + 0.0, + ) + ) + + +def main() -> int: parser = argparse.ArgumentParser(description=__doc__) live_mode = parser.add_mutually_exclusive_group() live_mode.add_argument("--live", nargs="+", type=Path, metavar="MOUNTPOINT") @@ -528,45 +1801,145 @@ def main() -> None: type=Path, metavar=("EMPTY_DIR_A", "EMPTY_DIR_B"), ) + parser.add_argument("--list", action="store_true", help="print case names and exit") + parser.add_argument("--timeout", type=int, default=DEFAULT_TIMEOUT, help="per-case seconds") + parser.add_argument("--fail-fast", action="store_true", help="stop at the first failure") + parser.add_argument("--report-json", type=Path, metavar="PATH") + parser.add_argument("--report-junit", type=Path, metavar="PATH") + parser.add_argument( + "--budget", + type=float, + default=float(os.environ.get("PDFS_ACCEPTANCE_BUDGET", "0")), + help="report any case slower than this many seconds", + ) + parser.add_argument( + "--journal-check", + action="store_true", + help="fail if the daemon logs errors during the run", + ) + parser.add_argument( + "--durability", + action="store_true", + help="restart the daemon mid-suite and verify written bytes survive", + ) args = parser.parse_args() + if args.list: + for case in TESTS: + print(f"{','.join(case.kinds):<18} {case.name}") + return 0 + + journal = JournalWatch(os.environ.get("PDFS_ACCEPTANCE_UNIT", "proton-drive.service")) + if args.journal_check: + journal.start() + with tempfile.TemporaryDirectory(prefix="pdfs-api-reference-") as directory: - reference, _ = run_contract(Path(directory), "local filesystem API reference (no account)") - shutil.rmtree(reference) - print("[pass] account-free filesystem API contract") + reference = run_contract( + Path(directory), + "local filesystem API reference (no account)", + kind=REFERENCE, + timeout=args.timeout, + fail_fast=args.fail_fast, + ) + shutil.rmtree(reference.root, ignore_errors=True) + if reference.failed: + print("FAIL: the account-free contract failed against an ordinary filesystem") + else: + print("[pass] account-free filesystem API contract") - if args.managed_live: - run_managed_matrix(args.managed_live) - print("PASS: offline API and managed live mode-matrix suites completed") - return + try: + if args.managed_live: + run_managed_matrix(args.managed_live, reference, args) + elif args.live: + run_live(args.live, reference, args) + finally: + outcome = finish(args, journal) + return outcome - if not args.live: - print("PASS: offline acceptance suite completed; live FUSE test was not requested") - return - completed: list[tuple[Path, Path, str]] = [] +def run_live(mountpoints: list[Path], reference: Run, args) -> None: + completed: list[tuple[Path, Run]] = [] convergence = os.environ.get("PDFS_ACCEPTANCE_CONVERGENCE", "0") == "1" + timeout = int(os.environ.get("PDFS_ACCEPTANCE_SYNC_TIMEOUT", "120")) + daemon = Daemon.discover(timeout, may_restart=args.durability) + if daemon is None: + print("WARNING: no reachable pdfs daemon; sync regressions will be skipped") try: # Views of one remote folder must not independently create the same # fixtures. Exercise the primary, then observe that exact tree through # every secondary. Without convergence mode, each mount is independent # and receives the full contract. - targets = args.live[:1] if convergence and len(args.live) > 1 else args.live + targets = mountpoints[:1] if convergence and len(mountpoints) > 1 else mountpoints for mountpoint in targets: - root, digest = run_contract(mountpoint.resolve(), "live FUSE") - completed.append((mountpoint.resolve(), root, digest)) - if convergence and len(args.live) > 1: + run = run_contract( + mountpoint.resolve(), + f"live FUSE {mountpoint}", + kind=LIVE, + daemon=daemon, + timeout=args.timeout, + fail_fast=args.fail_fast, + ) + completed.append((mountpoint.resolve(), run)) + summarize_divergence(reference, run) + if convergence and len(mountpoints) > 1: print(" [test] cross-mount remote convergence") - timeout = int(os.environ.get("PDFS_ACCEPTANCE_SYNC_TIMEOUT", "120")) - source_root, digest = completed[0][1], completed[0][2] - relative = source_root.relative_to(completed[0][0]) / "positioned.bin" - for mountpoint in args.live[1:]: - wait_for_copy(mountpoint.resolve(), relative, digest, timeout) + source = completed[0][1] + relative = source.root.relative_to(completed[0][0]) / "positioned.bin" + for mountpoint in mountpoints[1:]: + wait_for_copy(mountpoint.resolve(), relative, source.digest, timeout) finally: - for _, root, _ in completed: - shutil.rmtree(root, ignore_errors=True) - print("PASS: offline API and live FUSE acceptance suites completed") + for _, run in completed: + shutil.rmtree(run.root, ignore_errors=True) + + +def finish(args, journal: JournalWatch) -> int: + write_reports(args.report_json, args.report_junit) + + slow = report_timings(args.budget) + if slow: + print("[timing] cases over budget:") + for line in slow: + print(f" - {line}") + + journal_errors = journal.errors() if args.journal_check else [] + if journal_errors: + print(f"[journal] the daemon logged {len(journal_errors)} error(s) during the run:") + for line in journal_errors[:20]: + print(f" ! {line}") + + failures = [ + (run.label, result) + for run in REPORT + for result in run.results + if result.status in {"fail", "timeout"} + ] + print() + for run in REPORT: + counts: dict[str, int] = {} + for result in run.results: + counts[result.status] = counts.get(result.status, 0) + 1 + detail = ", ".join(f"{count} {status}" for status, count in sorted(counts.items())) + print(f"[summary] {run.label}: {detail}") + + if failures or journal_errors: + for label, result in failures: + print(f"FAIL {label}: {result.name}: {result.message}") + return 1 + print("PASS: acceptance suite completed") + return 0 if __name__ == "__main__": - main() + try: + sys.exit(main()) + except KeyboardInterrupt: + print("\ninterrupted", file=sys.stderr) + sys.exit(130) + except BrokenPipeError: + os.dup2(os.open(os.devnull, os.O_WRONLY), sys.stdout.fileno()) + sys.exit(0) + except (AssertionError, RuntimeError, TimeoutError, ValueError) as failure: + # Setup and selection problems are the operator's to fix; a traceback + # buries the one line that says what to change. + print(f"error: {failure}", file=sys.stderr) + sys.exit(2) diff --git a/scripts/fuse-acceptance.sh b/scripts/fuse-acceptance.sh index 6c5409a..de69865 100755 --- a/scripts/fuse-acceptance.sh +++ b/scripts/fuse-acceptance.sh @@ -4,44 +4,79 @@ set -Eeuo pipefail usage() { cat >&2 <<'EOF' -usage: scripts/fuse-acceptance.sh [--offline-only] - scripts/fuse-acceptance.sh --live MOUNTPOINT [MOUNTPOINT ...] - scripts/fuse-acceptance.sh --managed-live EMPTY_DIR EMPTY_DIR +usage: scripts/fuse-acceptance.sh [--offline-only] [OPTION ...] + scripts/fuse-acceptance.sh --live MOUNTPOINT [MOUNTPOINT ...] [OPTION ...] + scripts/fuse-acceptance.sh --managed-live EMPTY_DIR EMPTY_DIR [OPTION ...] scripts/fuse-acceptance.sh MOUNTPOINT [MOUNTPOINT ...] # compatibility -The local reference suite always runs first and requires no account. --live then -runs the same contract on each mount. Set PDFS_ACCEPTANCE_CONVERGENCE=1 only -when all live mountpoints show the same remote folder. +The local reference suite always runs first and requires no account; the mount +is then diffed against its recorded behaviour. --live runs the same contract on +each mount. Set PDFS_ACCEPTANCE_CONVERGENCE=1 only when all live mountpoints +show the same remote folder. + +Options are forwarded to the Python runner. The useful ones: + --list print every case and the targets it runs against + --timeout SECONDS per-case limit (default 180); a hung case dumps stacks + --fail-fast stop at the first failure instead of continuing + --report-json PATH machine-readable results and recorded observations + --report-junit PATH JUnit XML for CI + --budget SECONDS report any case slower than this + --journal-check fail if the daemon logs errors during the run + --durability restart the daemon mid-suite and re-verify bytes EOF exit 2 } command -v python3 >/dev/null || { echo "python3 is required" >&2; exit 2; } script_dir="$(cd -- "$(dirname -- "${BASH_SOURCE[0]}")" && pwd)" +runner=(python3 -u "$script_dir/fuse-acceptance.py") + +mountpoints=() +passthrough=() +# Paths come first and options after them, so the first dash-prefixed argument +# ends the path list and everything from there is forwarded untouched. +collect_paths() { + while (( $# > 0 )) && [[ "$1" != -* ]]; do + mountpoints+=("$1") + shift + done + passthrough=("$@") +} case "${1:-}" in -h|--help) usage ;; --offline-only|"") - (( $# <= 1 )) || usage - exec python3 -u "$script_dir/fuse-acceptance.py" + shift || true + exec "${runner[@]}" "$@" + ;; + --live) + shift + collect_paths "$@" + (( ${#mountpoints[@]} > 0 )) || usage ;; - --live) shift ; (( $# > 0 )) || usage ;; --managed-live) shift - (( $# == 2 )) || usage - exec python3 -u "$script_dir/fuse-acceptance.py" --managed-live "$@" + (( $# >= 2 )) || usage + first="$1"; second="$2"; shift 2 + exec "${runner[@]}" --managed-live "$first" "$second" "$@" + ;; + --*) + # Bare options with no mode: an offline run with those options. + exec "${runner[@]}" "$@" + ;; + *) + collect_paths "$@" ;; - --*) usage ;; esac command -v findmnt >/dev/null || { echo "findmnt is required for --live" >&2; exit 2; } -for mountpoint in "$@"; do +for mountpoint in "${mountpoints[@]}"; do [[ -d "$mountpoint" ]] || { echo "not a directory: $mountpoint" >&2; exit 2; } done -fstype="$(findmnt -T "$1" -n -o FSTYPE | head -n 1)" +fstype="$(findmnt -T "${mountpoints[0]}" -n -o FSTYPE | head -n 1)" [[ "$fstype" == fuse* ]] || { - echo "refusing primary non-FUSE path $1 (type: ${fstype:-unknown})" >&2 + echo "refusing primary non-FUSE path ${mountpoints[0]} (type: ${fstype:-unknown})" >&2 exit 2 } -exec python3 -u "$script_dir/fuse-acceptance.py" --live "$@" +exec "${runner[@]}" --live "${mountpoints[@]}" "${passthrough[@]}"