feat(sync): attach local folders to existing My files folders

This commit is contained in:
danielebrox committed 2026-11-02 19:23:49 +01:00
1 parent 800705ccb0
commit 5ca0272f15
13 files changed
+453 -1

No files matched your search

+25
View File
@@ -696,6 +696,15 @@ enum SyncCmd {
/// Local directory to sync.
path: PathBuf,
},
/// Sync a real local folder with an existing folder in My files.
/// The first pass preserves local-only files and keeps differing versions.
Attach {
/// Existing local directory; it remains a real local directory.
path: PathBuf,
/// Existing folder in My files, relative to the cloud mount or inside it.
#[arg(long)]
to: PathBuf,
},
/// List this machine's synced folders.
List,
/// Remove a synced folder by id.
@@ -1116,6 +1125,22 @@ fn cmd_sync(action: SyncCmd) -> Result<()> {
.to_string();
ok_or_bail(control_request(CtlRequest::AddSyncFolder { local_path })?)?
}
SyncCmd::Attach { path, to } => {
let abs = std::fs::canonicalize(&path)
.with_context(|| format!("resolve {}", path.display()))?;
let local_path = abs
.to_str()
.ok_or_else(|| anyhow!("path is not valid UTF-8"))?
.to_owned();
let remote_path = to
.to_str()
.ok_or_else(|| anyhow!("remote path is not valid UTF-8"))?
.to_owned();
ok_or_bail(control_request(CtlRequest::AttachSyncFolder {
local_path,
remote_path,
})?)?
}
SyncCmd::List => {
let response = control_request(CtlRequest::ListSyncFolders)?;
if emit_json(&response)? {
+7
View File
@@ -469,6 +469,13 @@ pub enum Request {
/// Add a local folder to this machine's device, uploading its tree and
/// registering the device on first use. Replies with [`Response::Ok`].
AddSyncFolder { local_path: String },
/// Keep a real local directory in step with an existing folder in My files.
/// First sync unions one-sided files and preserves differing versions.
/// Replies with [`Response::Ok`]; no device backup is created.
AttachSyncFolder {
local_path: String,
remote_path: String,
},
/// List this device's synced folders. Replies with [`Response::SyncFolders`].
ListSyncFolders,
/// Remove a synced folder by id; `delete_remote` also trashes its cloud copy.
+136
View File
@@ -0,0 +1,136 @@
//! Explicitly bind an existing My files folder to a real local mirror.
//!
//! Unlike a FUSE mount, this never covers or evicts the local directory.
//! The existing three-way engine performs the initial, non-destructive union.
use super::{Core, parse_uid, sync};
use pdfs_core::db::StoredSyncFolder;
use pdfs_core::{CoreError, CoreResult};
use proton_drive_rs::proton_sdk::ids::NodeUid;
use std::collections::HashSet;
use std::path::{Component, Path, PathBuf};
impl Core {
/// A bounded, server-verified ancestry walk. Do not decide ownership or
/// remote-root overlap from a possibly incomplete cached folder listing.
fn attach_ancestors(&self, start: &NodeUid) -> CoreResult<Vec<NodeUid>> {
let mut ancestors = Vec::new();
let mut seen = HashSet::new();
let mut next = Some(start.clone());
while let Some(uid) = next {
if !seen.insert(uid.clone()) || ancestors.len() >= 256 {
return Err(CoreError::invalid(
"invalid or excessively deep remote folder ancestry",
));
}
ancestors.push(uid.clone());
if uid == self.primary_root_uid {
break;
}
let nodes = self
.rt
.block_on(self.drive.enumerate_nodes(std::slice::from_ref(&uid)))
.map_err(|e| CoreError::from_api(&e, "verify sync folder ancestry"))?;
let node = nodes
.into_iter()
.next()
.ok_or_else(|| CoreError::not_found("remote sync folder is missing"))?;
if node.trashed {
return Err(CoreError::invalid("remote sync folder is in the trash"));
}
next = node.parent_uid;
}
Ok(ancestors)
}
pub(crate) fn attach_sync_folder(
&self,
local: &Path,
remote: &Path,
mountpoint: &Path,
) -> CoreResult<StoredSyncFolder> {
let _registration = super::devices::SYNC_ROOT_REGISTRATION.lock();
if !self.primary {
return Err(CoreError::invalid("attach must use My files"));
}
if remote.is_absolute()
|| remote
.components()
.any(|c| !matches!(c, Component::Normal(_) | Component::CurDir))
{
return Err(CoreError::invalid(
"remote folder must stay inside My files",
));
}
let local = local
.canonicalize()
.map_err(|e| CoreError::invalid(format!("resolve {}: {e}", local.display())))?;
if !local.is_dir() {
return Err(CoreError::invalid("local sync path is not a directory"));
}
let existing = self
.db
.sync_folder_list()
.map_err(|e| CoreError::internal(format!("db: {e:?}")))?;
let others = existing
.iter()
.map(|f| PathBuf::from(&f.local_path))
.collect::<Vec<_>>();
let mut reserved = crate::devices::sync_root_reserved();
reserved.mountpoint = Some(
mountpoint
.canonicalize()
.map_err(|e| CoreError::internal(format!("resolve mountpoint: {e}")))?,
);
if let Some(problem) = pdfs_core::syncroot::problem(&local, &others, &reserved) {
return Err(CoreError::invalid(problem));
}
let (ino, remote_uid) = self.resolve(remote)?;
let is_folder = self
.state()
.entries
.get(&ino)
.is_some_and(|e| e.node.is_folder() && !e.node.trashed);
if !is_folder {
return Err(CoreError::invalid("remote sync path is not a folder"));
}
let ancestors = self.attach_ancestors(&remote_uid)?;
if remote_uid == self.primary_root_uid || !ancestors.contains(&self.primary_root_uid) {
return Err(CoreError::invalid(
"choose an owned folder below My files, not its root or a shared folder",
));
}
for other in &existing {
let uid = parse_uid(&other.remote_uid)
.ok_or_else(|| CoreError::internal("invalid existing remote sync root"))?;
if uid == remote_uid
|| ancestors.contains(&uid)
|| self.attach_ancestors(&uid)?.contains(&remote_uid)
{
return Err(CoreError::invalid(
"remote folder overlaps an existing synced folder",
));
}
}
let share_id = self
.rt
.block_on(self.drive.context_share_id(&self.primary_root_uid))
.map_err(|e| CoreError::from_api(&e, "resolve My files share"))?
.to_string();
if share_id.is_empty() {
return Err(CoreError::invalid("My files share is unavailable"));
}
let id = self
.db
.sync_folder_add(&local.to_string_lossy(), &remote_uid.to_string(), &share_id)
.map_err(|e| CoreError::internal(format!("db: {e:?}")))?;
// An empty baseline makes one-sided files new, never deletions. The
// engine hashes matching-size files and preserves divergent versions.
let _ = self.sync_tx.send(sync::SyncMsg::Rewatch);
let _ = self.sync_tx.send(sync::SyncMsg::Reconcile(id));
self.db
.sync_folder_get(id)
.map_err(|e| CoreError::internal(format!("db: {e:?}")))?
.ok_or_else(|| CoreError::not_found("new sync folder disappeared"))
}
}
+26
View File
@@ -385,6 +385,7 @@ fn touched_topics(request: &CtlRequest) -> &'static [Topic] {
R::RenameDevice { .. } | R::DeleteDevice { .. } => &[Topic::Devices],
R::AdoptDevice { .. } | R::MigrateDevice { .. } => &[Topic::Devices, Topic::Locations],
R::AddSyncFolder { .. }
| R::AttachSyncFolder { .. }
| R::RemoveSyncFolder { .. }
| R::SetSyncFolderMode { .. }
| R::SetSyncFolderPaused { .. }
@@ -1397,6 +1398,31 @@ fn handle_control_conn(core: &Core, username: &str, mountpoint: &Path, stream: U
message: format!("adding {local_path}"),
}
}
Ok(CtlRequest::AttachSyncFolder {
local_path,
remote_path,
}) => {
let result = rel_to_mount(mountpoint, &remote_path).and_then(|remote| {
core.attach_sync_folder(Path::new(&local_path), &remote, mountpoint)
});
match result {
Ok(folder) => {
core.log_activity(
ActivityKind::Upload,
&local_path,
"attached existing My files folder",
true,
);
CtlResponse::Ok {
message: format!(
"sync folder {} attached: {}",
folder.id, folder.local_path
),
}
}
Err(error) => CtlResponse::error(error),
}
}
Ok(CtlRequest::ListSyncFolders) => match core.list_sync_folders() {
Ok(items) => CtlResponse::SyncFolders { items },
Err(e) => CtlResponse::error(e),
+6
View File
@@ -33,6 +33,11 @@ use super::{
now_secs, parse_uid, spawn_session, sync_folder_info, this_hostname,
};
/// Registration validates local and remote root overlap before inserting a
/// mapping. Serialize all registration paths so concurrent control requests
/// cannot both validate against the same old set of roots.
pub(super) static SYNC_ROOT_REGISTRATION: Mutex<()> = Mutex::new(());
/// A mode that this daemon is allowed to apply.
///
/// Persisted and control-protocol values first pass through [`MountMode`] so
@@ -513,6 +518,7 @@ impl Core {
/// tree into it once, and record the mapping. Phase 1 is a one-shot upload —
/// the two-way engine (Phase 2) reconciles later changes.
pub(crate) fn add_sync_folder(&self, local: &Path) -> CoreResult<StoredSyncFolder> {
let _registration = SYNC_ROOT_REGISTRATION.lock();
let meta = std::fs::metadata(local)
.map_err(|e| CoreError::internal(format!("stat {}: {e}", local.display())))?;
if !meta.is_dir() {
+7 -1
View File
@@ -26,7 +26,7 @@ use futures::StreamExt as _;
use futures::stream::BoxStream;
use proton_drive_rs::proton_sdk::account::Quota;
use proton_drive_rs::proton_sdk::error::Result;
use proton_drive_rs::proton_sdk::ids::{DriveEventId, NodeUid};
use proton_drive_rs::proton_sdk::ids::{DriveEventId, NodeUid, ShareId};
use proton_drive_rs::{
AvailableNameTarget, Device, DeviceType, DriveEvent, DriveEventScopeId, Node, NodeMoveItem,
ProtonDriveClient, RevisionReader, Thumbnail,
@@ -43,6 +43,8 @@ pub(crate) type OutcomeStream<'a> = BoxStream<'a, Result<(NodeUid, Result<()>)>>
pub(crate) trait DriveApi: Send + Sync {
async fn get_my_files_folder(&self) -> Result<Node>;
async fn context_share_id(&self, uid: &NodeUid) -> Result<ShareId>;
async fn get_node(&self, uid: &NodeUid) -> Result<Option<Node>>;
async fn enumerate_nodes(&self, uids: &[NodeUid]) -> Result<Vec<Node>>;
@@ -305,6 +307,10 @@ impl RevisionRead for RevisionReader {
#[async_trait]
impl DriveApi for ProtonDriveClient {
async fn context_share_id(&self, uid: &NodeUid) -> Result<ShareId> {
ProtonDriveClient::context_share_id(self, uid).await
}
async fn get_my_files_folder(&self) -> Result<Node> {
ProtonDriveClient::get_my_files_folder(self).await
}
+1
View File
@@ -79,6 +79,7 @@ use proton_drive_rs::{
};
mod albums;
mod attach;
mod background;
mod conflicts;
mod control;
+1
View File
@@ -389,6 +389,7 @@ impl Core {
items: &[pdfs_core::control::RestoreItem],
device: Option<&str>,
) -> CoreResult<RestoreOutcome> {
let _registration = super::devices::SYNC_ROOT_REGISTRATION.lock();
if items.is_empty() {
return Err(CoreError::invalid("nothing to restore"));
}
+214
View File
@@ -2912,6 +2912,220 @@ mod tests {
let _ = std::fs::remove_dir_all(&dir);
}
#[test]
#[ignore = "mounts FUSE: run explicitly on an isolated fake Drive"]
fn attaching_my_files_preserves_local_files_and_syncs_both_ways() {
use std::os::unix::fs::MetadataExt;
let drive = FakeDrive::new();
let dir = scratch("attach-my-files");
let daemon = Daemon::start(&dir, drive.client(1, Faults::lan())).unwrap();
std::fs::create_dir(daemon.mountpoint.join("Work")).unwrap();
assert!(wait_until(Duration::from_secs(30), || {
daemon.pending().is_ok_and(|items| items.is_empty()) && drive.lookup("Work").is_some()
}));
drive.device().write("Work/cloud.txt", b"from cloud");
drive.device().write("Work/common.txt", b"remote version");
drive.device().write("Work/identical.txt", b"same bytes");
let local = dir.join("local");
std::fs::create_dir(&local).unwrap();
std::fs::write(local.join("local.txt"), b"only on this computer").unwrap();
std::fs::write(local.join("common.txt"), b"local version").unwrap();
std::fs::write(local.join("identical.txt"), b"same bytes").unwrap();
daemon
.request(&Request::AttachSyncFolder {
local_path: local.to_str().unwrap().into(),
remote_path: "Work".into(),
})
.unwrap();
let folder = match daemon.request(&Request::ListSyncFolders).unwrap() {
Response::SyncFolders { items } => {
items
.into_iter()
.find(|f| f.local_path == local.to_str().unwrap())
.unwrap()
.id
}
other => panic!("unexpected reply: {other:?}"),
};
assert!(wait_until(Duration::from_secs(30), || {
drive.lookup("Work/local.txt").is_some()
&& std::fs::read(local.join("cloud.txt")).is_ok_and(|b| b == b"from cloud")
&& std::fs::read(local.join("common.txt")).is_ok_and(|b| b == b"remote version")
&& !conflict_copies(&local, "common").is_empty()
}));
assert_ne!(
std::fs::metadata(&local).unwrap().dev(),
std::fs::metadata(&daemon.mountpoint).unwrap().dev()
);
assert!(
conflict_copies(&local, "common")
.iter()
.any(|p| std::fs::read(p).is_ok_and(|b| b == b"local version"))
);
assert!(
daemon
.request(&Request::AttachSyncFolder {
local_path: local.to_str().unwrap().into(),
remote_path: "Work".into()
})
.is_err()
);
let duplicate = dir.join("duplicate");
std::fs::create_dir(&duplicate).unwrap();
assert!(
daemon
.request(&Request::AttachSyncFolder {
local_path: duplicate.to_str().unwrap().into(),
remote_path: "Work".into()
})
.is_err()
);
assert!(
daemon
.request(&Request::AttachSyncFolder {
local_path: duplicate.to_str().unwrap().into(),
remote_path: "../Work".into()
})
.is_err()
);
// Parent/child mappings, the account root and files cannot be attached.
std::fs::create_dir(daemon.mountpoint.join("Work/Sub")).unwrap();
assert!(wait_until(Duration::from_secs(30), || {
daemon.pending().is_ok_and(|items| items.is_empty())
&& drive.lookup("Work/Sub").is_some()
}));
for remote in ["Work/Sub", ".", "Work/cloud.txt"] {
assert!(
daemon
.request(&Request::AttachSyncFolder {
local_path: duplicate.to_str().unwrap().into(),
remote_path: remote.into(),
})
.is_err()
);
}
// Local overlap and symlink aliases into the actual cloud mount are refused.
let alias = dir.join("cloud-alias");
std::os::unix::fs::symlink(daemon.mountpoint.join("Work"), &alias).unwrap();
std::fs::create_dir_all(local.join("Sub")).unwrap();
for path in [
local.join("Sub"),
dir.clone(),
daemon.mountpoint.join("Work"),
alias,
] {
let error = daemon
.request(&Request::AttachSyncFolder {
local_path: path.to_string_lossy().into_owned(),
remote_path: "Work/Sub".into(),
})
.unwrap_err()
.to_string();
assert!(error.contains("overlaps"), "{error}");
}
assert!(conflict_copies(&local, "identical").is_empty());
assert_eq!(
std::fs::read(local.join("identical.txt")).unwrap(),
b"same bytes"
);
// Simultaneous requests cannot register two local trees for one cloud root.
std::fs::create_dir(daemon.mountpoint.join("Other")).unwrap();
assert!(wait_until(Duration::from_secs(30), || {
daemon.pending().is_ok_and(|items| items.is_empty()) && drive.lookup("Other").is_some()
}));
let candidate_a = dir.join("candidate-a");
let candidate_b = dir.join("candidate-b");
std::fs::create_dir(&candidate_a).unwrap();
std::fs::create_dir(&candidate_b).unwrap();
let outcomes = std::thread::scope(|scope| {
let a = scope.spawn(|| {
daemon
.request(&Request::AttachSyncFolder {
local_path: candidate_a.to_str().unwrap().into(),
remote_path: "Other".into(),
})
.map_err(|e| e.to_string())
});
let b = scope.spawn(|| {
daemon
.request(&Request::AttachSyncFolder {
local_path: candidate_b.to_str().unwrap().into(),
remote_path: "Other".into(),
})
.map_err(|e| e.to_string())
});
[a.join().unwrap(), b.join().unwrap()]
});
assert_eq!(
outcomes.iter().filter(|result| result.is_ok()).count(),
1,
"exactly one mapping may own a remote root: {outcomes:?}"
);
drive
.device()
.write("Work/cloud.txt", b"new cloud revision");
daemon
.request(&Request::SyncNow { id: Some(folder) })
.unwrap();
assert!(wait_until(Duration::from_secs(30), || std::fs::read(
local.join("cloud.txt")
)
.is_ok_and(|b| b == b"new cloud revision")));
std::fs::write(local.join("local.txt"), b"new local revision").unwrap();
daemon
.request(&Request::SyncNow { id: Some(folder) })
.unwrap();
assert!(wait_until(Duration::from_secs(30), || drive
.tree()
.get("Work/local.txt")
== Some(&Entry::File(Arc::new(b"new local revision".to_vec())))));
assert!(wait_until(Duration::from_secs(30), || listed_folder(
&daemon, folder
)
.is_some_and(|f| f.state == "idle")));
daemon
.request(&Request::SetSyncFolderPaused {
id: folder,
paused: true,
})
.unwrap();
let before = drive.tree();
for entry in std::fs::read_dir(&local).unwrap() {
let path = entry.unwrap().path();
if path.is_dir() {
std::fs::remove_dir_all(path).unwrap();
} else {
std::fs::remove_file(path).unwrap();
}
}
daemon
.request(&Request::SetSyncFolderPaused {
id: folder,
paused: false,
})
.unwrap();
daemon
.request(&Request::SyncNow { id: Some(folder) })
.unwrap();
assert!(wait_until(Duration::from_secs(30), || listed_folder(
&daemon, folder
)
.is_some_and(|f| f.state == "error")));
assert_eq!(
drive.tree(),
before,
"the wipe guard must preserve the remote tree"
);
assert!(matches!(daemon.stop(), Ok(MountOutcome::Shutdown)));
let db = Db::open(&dir.join("pdfs.db")).unwrap();
assert_eq!(
db.sync_folder_get(folder).unwrap().unwrap().remote_share_id,
"my-files-share"
);
drop(db);
let _ = std::fs::remove_dir_all(&dir);
}
/// Register `local` as a synced folder in `mode` and wait until the daemon
/// lists it that way, returning its id.
fn sync_folder(daemon: &Daemon, local: &Path, mode: &str) -> i64 {
+8
View File
@@ -1123,6 +1123,14 @@ fn delete_one(server: &mut Server, uid: &NodeUid) -> Result<()> {
#[async_trait]
impl DriveApi for FakeClient {
async fn context_share_id(&self, uid: &NodeUid) -> Result<ShareId> {
self.request(false, |server| {
assert_eq!(uid, &server.root);
Ok(ShareId::new("my-files-share"))
})
.await
}
async fn get_my_files_folder(&self) -> Result<Node> {
self.request(false, |server| {
let root = server.root.clone();