nvrl/proton-drive-linux
1
0 Fork 0
mirror of https://github.com/narrrl/proton-drive-linux.git synced 2026-11-03 16:41:26 +01:00
Code Issues Packages Projects Releases Wiki Activity

fixed race conditions + extendet test suite

Browse Source
This commit is contained in:
nvrl committed 2026-07-26 15:56:58 +02:00
1 parent a091c2f34f
commit 8470662c54
12 files changed
+2287 -243
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

crates/pdfs-fuse/src/background.rs
+95 -70
View File
@@ -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<State>,
content: &ContentCache,
pending: &Mutex<HashMap<NodeUid, PendingRevision>>,
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<u64> = 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<u64> = 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<Mutex<State>>,
content: Arc<ContentCache>,
db: Arc<Db>,
pending: Arc<Mutex<HashMap<NodeUid, PendingRevision>>>,
notifier: Notifier,
core: Core,
) {
let db = core.db.clone();
let mut cursor: Option<DriveEventId> = 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, &notifier, 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");
crates/pdfs-fuse/src/devices.rs
+5
View File
@@ -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
}
crates/pdfs-fuse/src/drain.rs
+67 -43
View File
@@ -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
crates/pdfs-fuse/src/filesystem.rs
+118
View File
@@ -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
crates/pdfs-fuse/src/lib.rs
+192 -20
View File
@@ -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<Mutex<HashMap<i64, Arc<Mutex<()>>>>>,
/// 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<StateRegistry>,
}
/// 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<Mutex<State>>, Arc<OnceLock<Notifier>>);
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<Mutex<State>>,
/// This mount's kernel notification channel — the same cell as
/// [`Core::notifier`], still empty until its session is spawned.
notifier: Arc<OnceLock<Notifier>>,
}
/// 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<Vec<MountedState>>);
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<Mutex<State>>, notifier: Arc<OnceLock<Notifier>>) {
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<LiveMount> {
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
crates/pdfs-fuse/src/mount.rs
+5 -9
View File
@@ -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
crates/pdfs-fuse/src/sharing.rs
+28 -10
View File
@@ -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<String> {
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<String> {
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("/"))
}
crates/pdfs-fuse/src/sweep.rs
+21 -6
View File
@@ -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