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

added functionality to open shared files with you + fixed bug where mid transfer the SIGTERM signal can corrupt the mount

Browse Source
This commit is contained in:
nvrl committed 2026-07-20 22:30:16 +02:00
1 parent e0f0c7e0b7
commit bf625bf8d9
15 files changed
+1621 -136
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

crates/pdfs-fuse/src/control.rs
+62 -12
View File
@@ -73,18 +73,24 @@ fn handle_control_conn(core: &Core, username: &str, mountpoint: &Path, stream: U
}
Ok(CtlRequest::Pin { path }) => match rel_to_mount(mountpoint, &path) {
Ok(rel) => match core.pin(&rel) {
Ok(name) => CtlResponse::Ok {
message: format!("pinned {name}"),
},
Ok(name) => {
core.touch_profile();
CtlResponse::Ok {
message: format!("pinned {name}"),
}
}
Err(e) => CtlResponse::error(e),
},
Err(e) => CtlResponse::error(e),
},
Ok(CtlRequest::Unpin { path }) => match rel_to_mount(mountpoint, &path) {
Ok(rel) => match core.unpin(&rel) {
Ok(name) => CtlResponse::Ok {
message: format!("unpinned {name}"),
},
Ok(name) => {
core.touch_profile();
CtlResponse::Ok {
message: format!("unpinned {name}"),
}
}
Err(e) => CtlResponse::error(e),
},
Err(e) => CtlResponse::error(e),
@@ -456,9 +462,12 @@ fn handle_control_conn(core: &Core, username: &str, mountpoint: &Path, stream: U
cfg.cache_budget = Some(bytes);
dirs.save_config(&cfg)
}) {
Ok(Ok(())) => CtlResponse::Ok {
message: format!("cache budget set to {bytes} bytes"),
},
Ok(Ok(())) => {
core.touch_profile();
CtlResponse::Ok {
message: format!("cache budget set to {bytes} bytes"),
}
}
Ok(Err(e)) => CtlResponse::error(CoreError::internal(format!(
"budget applied but config write failed: {e}"
))),
@@ -483,6 +492,10 @@ fn handle_control_conn(core: &Core, username: &str, mountpoint: &Path, stream: U
},
Err(e) => CtlResponse::error(e),
},
Ok(CtlRequest::AdoptDevice { uid }) => match core.adopt_device(uid.as_deref()) {
Ok(message) => CtlResponse::Ok { message },
Err(e) => CtlResponse::error(e),
},
Ok(CtlRequest::AddSyncFolder { local_path }) => {
// Registering the device and uploading a folder tree far outlasts the
// socket read timeout, so ack immediately and work on a background
@@ -494,6 +507,7 @@ fn handle_control_conn(core: &Core, username: &str, mountpoint: &Path, stream: U
let started = Instant::now();
match core.add_sync_folder(&path) {
Ok(folder) => {
core.touch_profile();
let name = Path::new(&folder.local_path)
.file_name()
.and_then(|n| n.to_str())
@@ -522,15 +536,19 @@ fn handle_control_conn(core: &Core, username: &str, mountpoint: &Path, stream: U
},
Ok(CtlRequest::RemoveSyncFolder { id, delete_remote }) => {
match core.remove_sync_folder(id, delete_remote) {
Ok(()) => CtlResponse::Ok {
message: "removed synced folder".to_string(),
},
Ok(()) => {
core.touch_profile();
CtlResponse::Ok {
message: "removed synced folder".to_string(),
}
}
Err(e) => CtlResponse::error(e),
}
}
Ok(CtlRequest::SetSyncFolderMode { id, mode }) => {
match core.request_sync_folder_mode(id, &mode) {
Ok(message) => {
core.touch_profile();
core.log_activity(ActivityKind::Upload, &message, "", true);
CtlResponse::Ok { message }
}
@@ -546,6 +564,28 @@ fn handle_control_conn(core: &Core, username: &str, mountpoint: &Path, stream: U
},
}
}
Ok(CtlRequest::ListRestorableFolders) => match core.list_restorable_folders() {
Ok(items) => CtlResponse::RestorableFolders { items },
Err(e) => CtlResponse::error(e),
},
Ok(CtlRequest::RestoreSyncFolders { items }) => {
// Like AddSyncFolder: each restored folder downloads a whole tree,
// which outlasts the socket timeout. Ack, then work.
let core = core.clone();
std::thread::spawn(move || match core.restore_sync_folders(&items) {
Ok(message) => {
core.touch_profile();
core.log_activity(ActivityKind::Download, &message, "", true);
}
Err(e) => {
warn!(error = %e, "restore failed");
core.log_activity(ActivityKind::Download, "restore folders", &e, false);
}
});
CtlResponse::Ok {
message: "restoring folders".to_string(),
}
}
Ok(CtlRequest::ShareNode {
path,
emails,
@@ -640,6 +680,16 @@ fn handle_control_conn(core: &Core, username: &str, mountpoint: &Path, stream: U
Ok(entries) => CtlResponse::Entries { entries },
Err(e) => CtlResponse::error(e),
},
Ok(CtlRequest::ListSharedFolder { uid }) => match core.list_shared_folder(&uid) {
Ok(entries) => CtlResponse::Entries { entries },
Err(e) => CtlResponse::error(e),
},
Ok(CtlRequest::OpenSharedFile { uid }) => match core.open_shared_file(&uid) {
Ok(p) => CtlResponse::FilePath {
path: p.display().to_string(),
},
Err(e) => CtlResponse::error(e),
},
Ok(CtlRequest::LeaveShared { uid }) => match core.leave_shared(&uid) {
Ok(()) => {
core.log_activity(ActivityKind::Unshare, "shared item", "left", true);
crates/pdfs-fuse/src/devices.rs
+127 -9
View File
@@ -26,9 +26,34 @@ use tracing::{info, warn};
use super::sync::{self, base_name};
use super::{
BackgroundSession, Core, ProtonFs, State, SwitchBlocked, clear_stale_mount, device_type_str,
dir_is_empty, evict_dir_contents, now_secs, parse_uid, sync_folder_info, this_hostname,
dir_is_empty, evict_dir_contents, fuse_connection_id, is_stale_mount, now_secs, parse_uid,
sync_folder_info, this_hostname, umount_session_unblocked,
};
/// The device uid this machine has been adopted into, from `config.json`.
///
/// Read fresh each time rather than cached on [`Core`]: adoption rewrites the
/// config from the control socket, and the next `ensure_device` must see it
/// without a daemon restart.
pub(crate) fn adopted_device_uid() -> Option<String> {
let uid = pdfs_core::config::AppDirs::new()
.ok()?
.load_config()
.device_uid?;
let uid = uid.trim().to_string();
(!uid.is_empty()).then_some(uid)
}
/// Pin (or, with `None`, unpin) the adopted device uid in `config.json`.
pub(crate) fn set_adopted_device_uid(uid: Option<&str>) -> CoreResult<()> {
let dirs = pdfs_core::config::AppDirs::new()
.map_err(|e| CoreError::internal(format!("config dirs: {e}")))?;
let mut cfg = dirs.load_config();
cfg.device_uid = uid.map(|u| u.to_string());
dirs.save_config(&cfg)
.map_err(|e| CoreError::internal(format!("write config: {e}")))
}
impl Core {
// ---- devices ----------------------------------------------------------
@@ -43,12 +68,14 @@ impl Core {
// No cached device row yet means this machine syncs nothing, so none of
// the listed devices is ours.
let this_uid = self.db.device_get().ok().flatten().map(|d| d.uid);
let pinned = adopted_device_uid();
Ok(devices
.into_iter()
.map(|d| {
let uid = d.uid.to_string();
DeviceInfo {
this_device: this_uid.as_deref() == Some(uid.as_str()),
adopted: pinned.as_deref() == Some(uid.as_str()),
uid,
name: d.name.unwrap_or_else(|_| "(unnamed device)".to_string()),
device_type: device_type_str(d.device_type).to_string(),
@@ -79,21 +106,97 @@ impl Core {
Ok(())
}
/// Adopt an existing device as *this* machine's, pinning it in `config.json`
/// so hostname changes and reinstalls stop registering duplicates
/// (features.md 5.1). `None` clears the pin and returns to hostname matching.
///
/// Refuses a uid that is not in the account: the whole point of the pin is
/// that it is authoritative, so accepting a bad one would defer the failure
/// to the next `ensure_device` with no obvious cause.
pub(crate) fn adopt_device(&self, uid: Option<&str>) -> CoreResult<String> {
let Some(uid) = uid else {
set_adopted_device_uid(None)?;
return Ok("adoption cleared; device now resolved by hostname".to_string());
};
let remote = self
.rt
.block_on(self.client.enumerate_devices())
.map_err(|e| CoreError::from_api(&e, "enumerate devices"))?;
let dev = remote
.into_iter()
.find(|d| d.uid.to_string() == uid)
.ok_or_else(|| CoreError::not_found(format!("no device {uid} in this account")))?;
let name = dev
.name
.as_deref()
.ok()
.map(|n| n.to_string())
.unwrap_or_else(this_hostname);
set_adopted_device_uid(Some(uid))?;
// Write the row through too, so callers that read the cached device
// (and `ListSyncFolders`' "this device" flag) agree immediately rather
// than after the next `ensure_device`.
self.db
.device_set(&StoredDevice {
uid: uid.to_string(),
share_id: dev.share_id.to_string(),
root_uid: dev.root_folder_uid.to_string(),
name: name.clone(),
created: dev.creation_time,
})
.map_err(|e| CoreError::internal(format!("db: {e:?}")))?;
info!(uid, name, "adopted device");
Ok(format!("adopted device {name}"))
}
// ---- device folder sync (devices.md, Phase 1) -------------------------
/// Auto-register (or recover) this machine as a Proton Drive Device, caching
/// it so restarts reuse the same device. Recovery matches an existing remote
/// Linux device by name before creating a new one, so a lost local record
/// doesn't orphan the device's root folder.
///
/// Resolution order: the *adopted* uid pinned in `config.json`, then the
/// cached DB row (validated remotely), then a hostname match, then create.
/// The pin exists because the hostname heuristic silently registers a second
/// device after a rename or reinstall, orphaning the first one's folders
/// (features.md 5.1).
pub(crate) fn ensure_device(&self) -> CoreResult<StoredDevice> {
let name = this_hostname();
// Enumerate the remote devices once: used both to validate any cached
// record and to recover an existing device by name.
// Enumerate the remote devices once: used to validate the pin, validate
// any cached record, and recover an existing device by name.
let remote = self
.rt
.block_on(self.client.enumerate_devices())
.map_err(|e| CoreError::from_api(&e, "enumerate devices"))?;
// An adopted uid is an explicit instruction, so it outranks everything
// and it fails loudly: falling back to the heuristic here would create
// the duplicate device the user adopted specifically to avoid.
if let Some(pin) = adopted_device_uid() {
let d = remote
.iter()
.find(|d| d.uid.to_string() == pin)
.ok_or_else(|| {
CoreError::not_found(format!(
"adopted device {pin} not found in this account; \
run `pdfs devices adopt --clear` to return to hostname matching"
))
})?;
let dev = StoredDevice {
uid: d.uid.to_string(),
share_id: d.share_id.to_string(),
root_uid: d.root_folder_uid.to_string(),
name: d.name.as_deref().ok().unwrap_or(&name).to_string(),
created: d.creation_time,
};
self.db
.device_set(&dev)
.map_err(|e| CoreError::internal(format!("db: {e:?}")))?;
return Ok(dev);
}
// A cached device is only trustworthy if it still exists remotely. A
// device deleted from another client (or the web UI) leaves a stale row
// whose root folder is gone, so creating folders under it fails with
@@ -346,8 +449,8 @@ impl Core {
// folder the daemon no longer tracks, and nothing would ever unmount it.
// Tear it down first — including before trashing the remote tree it
// serves, which would otherwise leave it answering for deleted nodes.
if let Some(session) = self.mounts.lock().remove(&id) {
if let Err(e) = session.umount_and_join() {
if let Some((session, conn)) = self.mounts.lock().remove(&id) {
if let Err(e) = umount_session_unblocked(session, conn) {
warn!(id, error = %e, "unmount on-demand folder failed");
} else {
info!(id, path = %folder.local_path, "unmounted on-demand folder");
@@ -599,7 +702,9 @@ impl Core {
return Err(SwitchBlocked::Failed(e.to_string()));
}
};
self.mounts.lock().insert(id, session);
self.mounts
.lock()
.insert(id, (session, fuse_connection_id(&local)));
let _ = self.sync_tx.send(sync::SyncMsg::Rewatch);
self.db.sync_folder_set_state(id, "idle", now_secs()).ok();
info!(id, path = %local.display(), "mounted sync folder on-demand");
@@ -607,8 +712,8 @@ impl Core {
}
_ => {
// ondemand→mirror: tear down the secondary mount first.
if let Some(session) = self.mounts.lock().remove(&id)
&& let Err(e) = session.umount_and_join()
if let Some((session, conn)) = self.mounts.lock().remove(&id)
&& let Err(e) = umount_session_unblocked(session, conn)
{
warn!(id, error = %e, "unmount on-demand folder failed");
}
@@ -676,6 +781,17 @@ impl Core {
continue;
}
let local = PathBuf::from(&folder.local_path);
// A daemon killed before it could unmount (systemd's stop timeout,
// say) leaves this path attached to a dead FUSE connection: it still
// exists, but every stat on it fails with ENOTCONN — so `is_dir()`
// reads it as *missing* and the folder gets parked in `error` on
// every subsequent start, with no way back but a manual
// `fusermount -u`. Detach the corpse first; the directory
// underneath then restores normally.
if is_stale_mount(&local) {
warn!(id = folder.id, path = %local.display(), "restore on-demand: clearing stale mount");
clear_stale_mount(&local);
}
if !local.is_dir() {
warn!(id = folder.id, path = %local.display(), "restore on-demand: local path missing");
let _ = self
@@ -737,7 +853,9 @@ impl Core {
};
match self.spawn_ondemand_mount(&local, root) {
Ok(session) => {
self.mounts.lock().insert(folder.id, session);
self.mounts
.lock()
.insert(folder.id, (session, fuse_connection_id(&local)));
let _ = self.db.sync_folder_set_state(folder.id, "idle", now_secs());
info!(id = folder.id, path = %local.display(), "remounted on-demand folder");
}
crates/pdfs-fuse/src/lib.rs
+110 -28
View File
@@ -75,6 +75,7 @@ use control::run_control_socket;
mod devices;
mod drain;
mod photos;
mod profile;
mod sharing;
mod state;
mod sync;
@@ -506,8 +507,10 @@ struct Core {
/// id (devices.md Phase 3). Each is a `ProtonFs` rooted at the folder's remote
/// node, mounted over its local path, sharing this Core's client/cache/db but
/// with its own inode space (`fork_state`). Held so we can unmount on toggle
/// back to `mirror` and on daemon shutdown.
mounts: Arc<Mutex<HashMap<i64, BackgroundSession>>>,
/// back to `mirror` and on daemon shutdown. The `u32` is the FUSE connection
/// id (see [`fuse_connection_id`]), captured at mount time so teardown can
/// abort a mid-transfer connection instead of blocking on it.
mounts: Arc<Mutex<HashMap<i64, SecondaryMount>>>,
/// Per-sync-folder locks, held for a whole reconcile pass and for a whole
/// mode switch. A `mirror→ondemand` flip evicts the local tree and mounts
/// FUSE over it, so it must never overlap a pass that is walking and
@@ -3317,22 +3320,25 @@ impl ProtonFs {
let st = self.core.state.lock();
st.children.get(&parent).and_then(|kids| {
kids.iter().copied().find_map(|ino| {
st.entries.get(&ino).filter(|e| e.node.name == name).map(|e| {
// A `lookup` reply carries attrs with the same TTL a
// `getattr` would, so `ls -l` — which is one `lookup`
// per entry and no `getattr` at all — takes its sizes
// from here. Resolving only in `getattr` left the whole
// listing provisional (bugs.md B14).
let provisional = matches!(
&e.node.kind,
NodeKind::File {
claimed_size: None,
..
}
)
.then(|| (e.parent, e.uid.clone()));
(ino, self.attr(ino, &e.node), provisional)
})
st.entries
.get(&ino)
.filter(|e| e.node.name == name)
.map(|e| {
// A `lookup` reply carries attrs with the same TTL a
// `getattr` would, so `ls -l` — which is one `lookup`
// per entry and no `getattr` at all — takes its sizes
// from here. Resolving only in `getattr` left the whole
// listing provisional (bugs.md B14).
let provisional = matches!(
&e.node.kind,
NodeKind::File {
claimed_size: None,
..
}
)
.then(|| (e.parent, e.uid.clone()));
(ino, self.attr(ino, &e.node), provisional)
})
})
})
};
@@ -3351,9 +3357,9 @@ impl ProtonFs {
if !off_loop {
let fs = self.clone();
let name = name.to_string();
self.core
.workers
.run(Lane::Meta, move || fs.serve_lookup(parent, &name, reply, true));
self.core.workers.run(Lane::Meta, move || {
fs.serve_lookup(parent, &name, reply, true)
});
return;
}
self.core.upgrade_sizes_for_parent(ino, &uid, grandparent);
@@ -3932,9 +3938,9 @@ impl Filesystem for ProtonFs {
return;
}
let fs = self.clone();
self.core
.workers
.run(Lane::Meta, move || fs.serve_lookup(parent, &name, reply, true));
self.core.workers.run(Lane::Meta, move || {
fs.serve_lookup(parent, &name, reply, true)
});
}
fn getattr(&self, _req: &Request, ino: INodeNo, _fh: Option<FileHandle>, reply: ReplyAttr) {
@@ -5198,6 +5204,66 @@ pub enum MountOutcome {
Unmounted,
}
/// Whether `path` is a mountpoint whose FUSE connection is dead — the state a
/// daemon killed before it could unmount leaves behind.
///
/// The kernel answers every operation on such a path with `ENOTCONN`, which the
/// ordinary existence checks (`is_dir`, `exists`) report as plain `false`: the
/// path looks *absent* rather than broken. Callers that would otherwise treat
/// that as "nothing to mount here" use this to tell the two apart.
pub(crate) fn is_stale_mount(path: &Path) -> bool {
matches!(
std::fs::metadata(path).map_err(|e| e.raw_os_error()),
Err(Some(libc::ENOTCONN))
)
}
/// A secondary (on-demand sync folder) FUSE session, paired with its FUSE
/// connection id so teardown can abort a mid-transfer connection rather than
/// block on it. See [`Core::mounts`] and [`umount_session_unblocked`].
type SecondaryMount = (BackgroundSession, Option<u32>);
/// The kernel's id for the FUSE connection backing the mount at `mountpoint` —
/// the directory name under `/sys/fs/fuse/connections`, which is the minor
/// number of the mountpoint's device. Must be read *while still mounted*; after
/// unmount the path resolves to the underlying directory on another device.
/// `None` when the path can't be stat'd.
fn fuse_connection_id(mountpoint: &Path) -> Option<u32> {
use std::os::unix::fs::MetadataExt;
let dev = std::fs::metadata(mountpoint).ok()?.dev();
Some(libc::minor(dev))
}
/// Force the kernel to abort FUSE connection `id`, erroring every in-flight
/// request.
///
/// On a stop signal we unmount lazily (`MNT_DETACH`) so the call succeeds even
/// mid-transfer — but detach only removes the mountpoint; it does *not* end the
/// connection while a request is still in flight. fuser's session loop then
/// blocks on `/dev/fuse` waiting for a next request that never comes, so `join`
/// hangs — long enough during a transfer that systemd's stop timeout SIGKILLs
/// the daemon mid-unmount, stranding the on-demand mounts as dead endpoints.
/// Writing the connection's `abort` file makes the pending reads fail with
/// `ENODEV`, so the loop returns and `join` completes at once. Best-effort:
/// there is nothing more to do on the shutdown path if it fails.
fn abort_fuse_connection(id: u32) {
let path = format!("/sys/fs/fuse/connections/{id}/abort");
if std::fs::write(&path, b"1").is_ok() {
info!(id, "aborted FUSE connection to unblock unmount");
}
}
/// Unmount a background session that may be mid-transfer without wedging: abort
/// its connection first (so the session loop exits promptly), then lazily
/// unmount and join. `conn` is the id captured at mount time; `None` skips the
/// abort and just unmounts (a healthy idle mount joins on its own `Destroy`).
fn umount_session_unblocked(session: BackgroundSession, conn: Option<u32>) -> std::io::Result<()> {
if let Some(id) = conn {
abort_fuse_connection(id);
}
session.umount_and_join()
}
/// Best-effort teardown of a stale mount left behind by a crashed daemon. A
/// previous run that died without unmounting leaves the kernel mount in place,
/// so the fresh `Session::new` below would fail with EBUSY ("Device or resource
@@ -5419,6 +5485,9 @@ pub fn mount(
// for the event task. `spawn` runs the session loop on its own background
// thread; we then wait here for either a stop signal or the mount ending.
let bg = Session::new(fs, mountpoint, &config)?.spawn()?;
// The connection id, captured now while the mount is live, so a stop signal
// mid-transfer can abort it rather than block `join` (see `abort_fuse_connection`).
let main_conn = fuse_connection_id(mountpoint);
let notifier = bg.notifier();
// Same channel, kept on the `Core` so background work (a size upgrade, say)
// can invalidate kernel-cached metadata without threading a handle through.
@@ -5470,7 +5539,7 @@ pub fn mount(
match sig_rx.recv_timeout(Duration::from_millis(500)) {
Ok(()) => {
info!("stop requested; unmounting");
if let Err(e) = bg.umount_and_join() {
if let Err(e) = umount_session_unblocked(bg, main_conn) {
warn!(error = %e, "umount_and_join failed");
}
break MountOutcome::Shutdown;
@@ -5495,8 +5564,8 @@ pub fn mount(
// Unmount every on-demand sync folder too, or the kernel mounts linger as
// stale and the next start fails with EBUSY (devices.md Phase 3).
let secondaries: Vec<_> = core.mounts.lock().drain().collect();
for (id, session) in secondaries {
if let Err(e) = session.umount_and_join() {
for (id, (session, conn)) in secondaries {
if let Err(e) = umount_session_unblocked(session, conn) {
warn!(id, error = %e, "unmount on-demand folder failed");
}
}
@@ -5880,7 +5949,20 @@ mod pending_size_tests {
#[cfg(test)]
mod tests {
use super::{Intervals, conflict_name};
use super::{Intervals, conflict_name, is_stale_mount};
/// 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
/// that are simply missing.
#[test]
fn is_stale_mount_is_narrow() {
let dir = std::env::temp_dir().join(format!("pdfs-stale-{}", std::process::id()));
std::fs::create_dir_all(&dir).unwrap();
assert!(!is_stale_mount(&dir));
assert!(!is_stale_mount(&dir.join("no-such-entry")));
let _ = std::fs::remove_dir_all(&dir);
}
/// Flatten `segments` into a readable form for assertions.
fn segs(iv: &Intervals, start: u64, end: u64) -> Vec<(u64, u64, bool)> {
crates/pdfs-fuse/src/profile.rs
+394
View File
@@ -0,0 +1,394 @@
//! Reading and writing this machine's `profile.json` in its device root, and
//! the restore that rebuilds a machine from one (features.md 5.2 and 5.3).
//!
//! The document itself is defined in [`pdfs_core::profile`]; this is the half
//! that talks to Drive. Saving is best-effort and never blocks a user action:
//! the profile is a convenience for a *future* machine, so failing to write it
//! must not fail the folder add that triggered it.
use std::io::Cursor;
use std::path::{Path, PathBuf};
use std::sync::OnceLock;
use std::sync::atomic::{AtomicBool, Ordering};
use pdfs_core::control::RestorableFolder;
use pdfs_core::profile::{PROFILE_FILE_NAME, PROFILE_VERSION, Profile, ProfileFolder, ProfilePin};
use pdfs_core::{CoreError, CoreResult};
use proton_drive_rs::proton_sdk::ids::NodeUid;
use tracing::{info, warn};
use super::sync;
use super::{Core, now_secs, parse_uid, this_hostname};
/// Coalescing flags for background profile saves.
///
/// A single burst of user actions (adding three folders, pinning a tree) should
/// cost one upload, not one per action. `SAVING` admits one worker at a time and
/// `DIRTY` records that something changed while it was busy, so the worker takes
/// exactly one more lap afterwards.
///
/// Process-global rather than a [`Core`] field because on-demand mounts fork
/// their own `Core` (`Core::fork_state`) and all of them describe the same
/// machine — there is one profile per daemon, so there is one flag pair.
static SAVE_STATE: OnceLock<(AtomicBool, AtomicBool)> = OnceLock::new();
fn save_state() -> &'static (AtomicBool, AtomicBool) {
SAVE_STATE.get_or_init(|| (AtomicBool::new(false), AtomicBool::new(false)))
}
impl Core {
/// Note that the arrangement this machine's profile describes has changed,
/// and upload a fresh profile in the background.
///
/// Fire-and-forget by design: callers are mutation paths (add/remove a sync
/// folder, switch a mode, pin a node, save settings) whose success does not
/// depend on the backup succeeding.
pub(crate) fn touch_profile(&self) {
let (saving, dirty) = save_state();
dirty.store(true, Ordering::SeqCst);
// Someone is already uploading; they will see DIRTY and take another lap.
if saving.swap(true, Ordering::SeqCst) {
return;
}
let core = self.clone();
std::thread::spawn(move || {
let (saving, dirty) = save_state();
while dirty.swap(false, Ordering::SeqCst) {
if let Err(e) = core.save_profile() {
warn!(error = %e.message, "profile backup failed");
break;
}
}
saving.store(false, Ordering::SeqCst);
});
}
/// Serialize this machine's arrangement and upload it to `profile.json` in
/// the device root, replacing any existing one.
///
/// Does nothing when no device is registered: a machine that has never
/// synced a folder has nothing to describe, and registering a device just to
/// hold an empty profile would put a stray computer in the user's Drive.
pub(crate) fn save_profile(&self) -> CoreResult<()> {
let Some(device) = self
.db
.device_get()
.map_err(|e| CoreError::internal(format!("db: {e:?}")))?
else {
return Ok(());
};
let root_uid = parse_uid(&device.root_uid).ok_or_else(|| {
CoreError::internal(format!("bad device root uid: {}", device.root_uid))
})?;
let profile = self.build_profile(&device.uid)?;
let bytes = profile.to_bytes().map_err(CoreError::internal)?;
let len = bytes.len() as i64;
// Replace the existing document rather than adding a second file of the
// same name: the profile is a single mutable record, and a device root
// with `profile.json` and `profile (1).json` in it teaches the restore
// nothing about which to trust.
let existing = self.find_device_child_file(&root_uid, PROFILE_FILE_NAME)?;
match existing {
Some(uid) => self
.rt
.block_on(self.client.upload_new_revision_from(
&uid,
Cursor::new(bytes),
len,
Vec::new(),
Some(now_secs()),
))
.map(|_| ())
.map_err(|e| CoreError::from_api(&e, "upload profile revision"))?,
None => {
self.rt
.block_on(self.client.upload_file_replacing_draft_from(
&root_uid,
PROFILE_FILE_NAME,
"application/json",
Cursor::new(bytes),
len,
Vec::new(),
Some(now_secs()),
false,
))
.map_err(|e| CoreError::from_api(&e, "upload profile"))?;
}
}
info!(folders = profile.folders.len(), "profile backed up");
Ok(())
}
/// Snapshot the local arrangement into a [`Profile`].
fn build_profile(&self, device_uid: &str) -> CoreResult<Profile> {
let folders = self
.db
.sync_folder_list()
.map_err(|e| CoreError::internal(format!("db: {e:?}")))?
.into_iter()
.map(|f| ProfileFolder {
remote_uid: f.remote_uid,
local_path: f.local_path,
// A queued switch is where the user asked the folder to be, so
// that is what the next machine should set up.
mode: f.pending_mode.unwrap_or(f.mode),
})
.collect();
let pins = self
.db
.pin_list()
.map_err(|e| CoreError::internal(format!("db: {e:?}")))?
.into_iter()
.map(|(uid, path, recursive)| ProfilePin {
uid,
path,
recursive,
})
.collect();
let cfg = pdfs_core::config::AppDirs::new()
.map(|d| d.load_config())
.unwrap_or_default();
Ok(Profile {
version: PROFILE_VERSION,
device_uid: device_uid.to_string(),
hostname: this_hostname(),
saved_at: now_secs(),
folders,
pins,
ignore_patterns: cfg.ignore_patterns,
cache_budget: cfg.cache_budget,
mountpoint: cfg.mountpoint,
})
}
/// An untrashed *file* named `name` directly under the device root.
///
/// The sibling of `find_device_child_folder`; kept separate because a folder
/// named `profile.json` must not be mistaken for the profile.
fn find_device_child_file(
&self,
root_uid: &NodeUid,
name: &str,
) -> CoreResult<Option<NodeUid>> {
let uids = self
.rt
.block_on(self.client.enumerate_folder_children_node_uids(root_uid))
.map_err(|e| CoreError::from_api(&e, "list device root"))?;
if uids.is_empty() {
return Ok(None);
}
let nodes = self
.rt
.block_on(self.client.enumerate_nodes(&uids))
.map_err(|e| CoreError::from_api(&e, "resolve device root children"))?;
Ok(nodes
.into_iter()
.find(|n| !n.is_folder() && !n.trashed && n.name == name)
.map(|n| n.uid))
}
/// Download and parse the profile in the device root, if there is one.
fn load_profile(&self, root_uid: &NodeUid) -> CoreResult<Option<Profile>> {
let Some(uid) = self.find_device_child_file(root_uid, PROFILE_FILE_NAME)? else {
return Ok(None);
};
let mut buf: Vec<u8> = Vec::new();
self.rt
.block_on(self.client.download_file_to(&uid, &mut buf))
.map_err(|e| CoreError::from_api(&e, "download profile"))?;
Profile::parse(&buf).map(Some).map_err(CoreError::invalid)
}
// ---- restore (features.md 5.2) ----------------------------------------
/// The folders under this machine's device that could be synced here, each
/// carrying a proposed local path.
///
/// The proposal comes from the profile when one exists and its path is
/// usable on this machine, and from `~/<name>` otherwise — a restored
/// `/home/other-user/Documents` would be wrong on a machine where that user
/// does not exist, and silently creating it would be worse.
pub(crate) fn list_restorable_folders(&self) -> CoreResult<Vec<RestorableFolder>> {
let device = self.ensure_device()?;
let root_uid = parse_uid(&device.root_uid).ok_or_else(|| {
CoreError::internal(format!("bad device root uid: {}", device.root_uid))
})?;
let profile = match self.load_profile(&root_uid) {
Ok(p) => p,
// A missing or unreadable profile is not fatal: the remote tree
// alone is enough to offer a restore, just with plainer defaults.
Err(e) => {
warn!(error = %e.message, "could not read profile; proposing defaults");
None
}
};
let uids = self
.rt
.block_on(self.client.enumerate_folder_children_node_uids(&root_uid))
.map_err(|e| CoreError::from_api(&e, "list device root"))?;
let nodes = if uids.is_empty() {
Vec::new()
} else {
self.rt
.block_on(self.client.enumerate_nodes(&uids))
.map_err(|e| CoreError::from_api(&e, "resolve device root children"))?
};
// Folders already synced here are not restorable — they are already
// restored — but they are still listed, flagged, so the picker can show
// the whole device rather than a confusing subset.
let existing: Vec<String> = self
.db
.sync_folder_list()
.map_err(|e| CoreError::internal(format!("db: {e:?}")))?
.into_iter()
.map(|f| f.remote_uid)
.collect();
Ok(nodes
.into_iter()
.filter(|n| n.is_folder() && !n.trashed)
.map(|n| {
let uid = n.uid.to_string();
let saved = profile
.as_ref()
.and_then(|p| p.folders.iter().find(|f| f.remote_uid == uid));
let proposed = saved
.map(|f| PathBuf::from(&f.local_path))
.filter(|p| path_is_usable_here(p))
.unwrap_or_else(|| default_local_path(&n.name));
RestorableFolder {
already_synced: existing.iter().any(|e| e == &uid),
remote_uid: uid,
name: n.name,
local_path: proposed.to_string_lossy().to_string(),
mode: saved
.map(|f| f.mode.clone())
.unwrap_or_else(|| "mirror".into()),
}
})
.collect())
}
/// Attach remote device folders to local paths and let the sync engine fill
/// them in.
///
/// The download is not written here: a new sync-folder row starts with an
/// empty baseline, and an empty baseline against an empty local directory
/// and a populated remote already classifies as "download everything". The
/// restore is therefore just bookkeeping plus a `Reconcile`, and it reuses
/// the engine's conflict handling for the case where the local directory
/// turns out not to be empty after all.
pub(crate) fn restore_sync_folders(
&self,
items: &[pdfs_core::control::RestoreItem],
) -> CoreResult<String> {
if items.is_empty() {
return Err(CoreError::invalid("nothing to restore"));
}
let device = self.ensure_device()?;
let existing = self
.db
.sync_folder_list()
.map_err(|e| CoreError::internal(format!("db: {e:?}")))?;
let mut restored = 0usize;
let mut skipped: Vec<String> = Vec::new();
for item in items {
let local = PathBuf::from(&item.local_path);
if !local.is_absolute() {
skipped.push(format!("{}: local path must be absolute", item.local_path));
continue;
}
if existing.iter().any(|f| f.remote_uid == item.remote_uid) {
skipped.push(format!("{}: already synced", item.local_path));
continue;
}
if existing
.iter()
.any(|f| Path::new(&f.local_path) == local.as_path())
{
skipped.push(format!("{}: path already in use", item.local_path));
continue;
}
if let Err(e) = std::fs::create_dir_all(&local) {
skipped.push(format!("{}: {e}", item.local_path));
continue;
}
let mode = match item.mode.as_str() {
"mirror" | "ondemand" => item.mode.as_str(),
other => {
skipped.push(format!("{}: unknown mode {other}", item.local_path));
continue;
}
};
let id = self
.db
.sync_folder_add(&local.to_string_lossy(), &item.remote_uid, &device.share_id)
.map_err(|e| CoreError::internal(format!("db: {e:?}")))?;
// `ondemand` is applied through the normal request path, which
// mounts the FUSE session and handles the folder being busy — the
// restore does not get its own copy of that logic.
if mode == "ondemand" {
let _ = self.request_sync_folder_mode(id, "ondemand");
}
let _ = self.sync_tx.send(sync::SyncMsg::Reconcile(id));
restored += 1;
}
let _ = self.sync_tx.send(sync::SyncMsg::Rewatch);
let mut message = format!("restored {restored} folder(s)");
if !skipped.is_empty() {
message.push_str(&format!("; skipped: {}", skipped.join(", ")));
}
info!(restored, skipped = skipped.len(), "restore complete");
Ok(message)
}
}
/// Whether a path recorded on another machine can be reused verbatim here.
///
/// Requires the *parent* to exist: `~/Documents` on a machine with the same
/// username is a fine proposal even though the directory itself is gone (the
/// restore creates it), but `/home/someone-else/Documents` is not.
fn path_is_usable_here(path: &Path) -> bool {
path.is_absolute() && path.parent().is_some_and(|p| p.is_dir())
}
/// `~/<name>`, or `/<name>` if the home directory cannot be determined.
fn default_local_path(name: &str) -> PathBuf {
match pdfs_core::config::AppDirs::new()
.ok()
.and_then(|d| d.home_dir())
{
Some(home) => home.join(name),
None => PathBuf::from("/").join(name),
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn foreign_home_is_not_reused() {
assert!(!path_is_usable_here(Path::new(
"/home/definitely-not-a-user-here/Documents"
)));
assert!(!path_is_usable_here(Path::new("relative/path")));
}
/// A path whose parent exists but which does not itself is exactly the
/// restore case — the directory is about to be created.
#[test]
fn missing_leaf_under_existing_parent_is_reused() {
let dir = std::env::temp_dir().join("pdfs-restore-probe");
std::fs::create_dir_all(&dir).unwrap();
assert!(path_is_usable_here(&dir.join("Documents")));
let _ = std::fs::remove_dir_all(&dir);
}
}
crates/pdfs-fuse/src/sharing.rs
+84 -25
View File
@@ -15,7 +15,34 @@ use pdfs_core::{CoreError, CoreResult};
use proton_drive_rs::proton_sdk::ids::NodeUid;
use proton_drive_rs::{MemberRole, NodeKind};
use super::{Core, ROOT_INO, parse_uid, public_link_info, role_from_str, role_to_str};
use proton_drive_rs::Node;
use super::{Core, ROOT_INO, node_size, parse_uid, public_link_info, role_from_str, role_to_str};
/// A node from a *shared* listing as a [`DirEntry`]. Shared nodes live outside
/// the mount, so the entry is uid-addressed: `path` is empty, and the local-state
/// flags (`pinned`, `cached`) are false — they describe my own tree, not this one.
fn shared_entry(n: Node) -> DirEntry {
let is_dir = n.is_folder();
let size = match &n.kind {
NodeKind::File {
claimed_size,
total_size_on_storage,
..
} => claimed_size.unwrap_or(*total_size_on_storage).max(0) as u64,
NodeKind::Folder => 0,
};
DirEntry {
name: n.name,
is_dir,
size,
modified: n.modification_time,
pinned: false,
cached: false,
uid: n.uid.to_string(),
path: String::new(),
}
}
impl Core {
// ---- sharing a node ---------------------------------------------------
@@ -232,30 +259,62 @@ impl Core {
.rt
.block_on(self.client.enumerate_nodes(&uids))
.map_err(|e| CoreError::from_api(&e, "enumerate nodes"))?;
Ok(nodes
.into_iter()
.map(|n| {
let is_dir = n.is_folder();
let size = match &n.kind {
NodeKind::File {
claimed_size,
total_size_on_storage,
..
} => claimed_size.unwrap_or(*total_size_on_storage).max(0) as u64,
NodeKind::Folder => 0,
};
DirEntry {
name: n.name,
is_dir,
size,
modified: 0,
pinned: false,
cached: false,
uid: n.uid.to_string(),
path: String::new(),
}
})
.collect())
Ok(nodes.into_iter().map(shared_entry).collect())
}
/// List the children of a folder shared with me, addressed by uid.
///
/// A shared subtree has no path in this account's mount — the FUSE tree only
/// spans my own volume — so browsing into one is uid-addressed the whole way
/// down, each listing handing the front-end the uids it descends with next.
pub(crate) fn list_shared_folder(&self, uid: &str) -> CoreResult<Vec<DirEntry>> {
let uid =
parse_uid(uid).ok_or_else(|| CoreError::invalid(format!("invalid uid: {uid}")))?;
let child_uids = self
.rt
.block_on(self.client.enumerate_folder_children_node_uids(&uid))
.map_err(|e| CoreError::from_api(&e, "enumerate shared children"))?;
if child_uids.is_empty() {
return Ok(Vec::new());
}
let nodes = self
.rt
.block_on(self.client.enumerate_nodes(&child_uids))
.map_err(|e| CoreError::from_api(&e, "enumerate nodes"))?;
Ok(nodes.into_iter().map(shared_entry).collect())
}
/// Download a file shared with me into the content cache, returning its
/// on-disk path (served from cache when a fresh blob already exists).
///
/// The by-uid twin of [`Core::open_file`]: same cache and same tracked
/// download, but the node is fetched from the server rather than resolved
/// through the inode tree, which does not cover other people's volumes.
///
/// [`Core::open_file`]: super::Core::open_file
pub(crate) fn open_shared_file(&self, uid: &str) -> CoreResult<std::path::PathBuf> {
let uid =
parse_uid(uid).ok_or_else(|| CoreError::invalid(format!("invalid uid: {uid}")))?;
let node = self
.rt
.block_on(self.client.get_node(&uid))
.map_err(|e| CoreError::from_api(&e, "get shared node"))?
.ok_or_else(|| CoreError::not_found("shared node not found"))?;
if !node.is_file() {
return Err(CoreError::invalid("not a regular file"));
}
let size = node_size(&node);
let mtime = node.modification_time;
if let Some(p) = self.cache.cached_content_path(&uid, mtime, size) {
return Ok(p);
}
let bytes = self
.download_file_tracked(&uid, &node.name, size)
.map_err(|e| CoreError::from_api(&e, "download"))?;
self.cache
.store(&uid, mtime, size, &bytes)
.map_err(|e| CoreError::internal(format!("cache store: {e}")))?;
Ok(self.cache.content_path(&uid))
}
/// Leave a shared node by its uid.
crates/pdfs-fuse/src/sync.rs
+12 -1
View File
@@ -2094,7 +2094,8 @@ mod tests {
let remote: HashMap<String, RemoteItem> = HashMap::new();
let baseline: HashMap<String, StoredSyncEntry> = HashMap::new();
let order = classification_order(&local, &remote, &baseline, &rules_for(&["node_modules/"]));
let order =
classification_order(&local, &remote, &baseline, &rules_for(&["node_modules/"]));
assert_eq!(
order,
@@ -2181,6 +2182,16 @@ mod tests {
assert!(guard_local_wipe(&base, &fresh).is_err());
}
/// A restored folder (features.md 5.2) starts with an empty baseline against
/// an empty local directory and a full remote. That must reconcile as
/// "download everything" rather than tripping the wipe guard — otherwise
/// every restore would wedge its folder on the first pass.
#[test]
fn guard_local_wipe_is_inert_for_a_restored_folder() {
let empty: HashMap<String, ()> = HashMap::new();
assert!(guard_local_wipe(&empty, &empty).is_ok());
}
#[test]
fn parent_and_base_split_relative_paths() {
assert_eq!(parent_rel("a"), "");