mirror of
https://github.com/narrrl/proton-drive-linux.git
synced 2026-11-04 16:56:59 +01:00
nvrl
51800c77d7
Nothing stopped `/`, the home folder, the Proton Drive mount, the app's own state or cache, or a folder nested in another root from becoming a sync root, and a restore took a path from another machine's profile as it was. One validator, `syncroot::problem`, now runs where roots are added and restored, on paths resolved through symlinks with `..` refused (B234).
1401 lines
60 KiB
Rust
1401 lines
60 KiB
Rust
//! This machine as a Proton Drive Device, and the local folders it keeps in
|
|
//! step with the remote.
|
|
//!
|
|
//! A sync folder runs in one of two modes. `mirror` keeps a real local tree that
|
|
//! the engine in [`super::sync`] reconciles both ways; `ondemand` evicts that
|
|
//! tree and mounts a second FUSE session over the same path, rooted at the
|
|
//! folder`s remote node. Flipping between them is the delicate part — see
|
|
//! [`Core::apply_sync_folder_mode`].
|
|
|
|
use std::collections::HashMap;
|
|
use std::path::{Path, PathBuf};
|
|
use std::sync::Arc;
|
|
|
|
use parking_lot::Mutex;
|
|
use pdfs_core::batch;
|
|
use pdfs_core::control::{
|
|
ActivityKind, DeviceInfo, JobItem, MountKind, MountSpec, SyncFolderInfo, SyncPhase,
|
|
SyncProgress,
|
|
};
|
|
use pdfs_core::db::{StoredDevice, StoredSyncFolder};
|
|
use pdfs_core::mounts::MountMode;
|
|
use pdfs_core::profile::PROFILE_DIR_NAME;
|
|
use pdfs_core::{CoreError, CoreResult};
|
|
use proton_drive_rs::proton_sdk::ids::{DeviceUid, NodeUid};
|
|
use proton_drive_rs::{DeviceType, Node};
|
|
use std::sync::OnceLock;
|
|
use tracing::{info, warn};
|
|
|
|
use super::sync::{self, base_name};
|
|
use super::{
|
|
Core, SecondaryInsertRejection, SecondaryMount, SwitchBlocked, clear_stale_mount,
|
|
device_type_str, dir_is_empty, evict_dir_contents, fuse_connection_id, is_stale_mount,
|
|
now_secs, parse_uid, spawn_session, sync_folder_info, this_hostname,
|
|
};
|
|
|
|
/// A mode that this daemon is allowed to apply.
|
|
///
|
|
/// Persisted and control-protocol values first pass through [`MountMode`] so
|
|
/// forward-compatible unknown values never reach either destructive transition.
|
|
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
|
|
pub(crate) enum ApplicableMountMode {
|
|
Mirror,
|
|
OnDemand,
|
|
}
|
|
|
|
impl ApplicableMountMode {
|
|
fn parse(value: &str) -> Result<Self, String> {
|
|
match MountMode::from(value) {
|
|
MountMode::Mirror => Ok(Self::Mirror),
|
|
MountMode::OnDemand => Ok(Self::OnDemand),
|
|
MountMode::Unknown => Err(format!("unknown mode {value:?} (want mirror|ondemand)")),
|
|
}
|
|
}
|
|
|
|
const fn as_str(self) -> &'static str {
|
|
match self {
|
|
Self::Mirror => "mirror",
|
|
Self::OnDemand => "ondemand",
|
|
}
|
|
}
|
|
}
|
|
|
|
fn select_mode_transition(
|
|
current: &str,
|
|
target: &str,
|
|
) -> Result<Option<ApplicableMountMode>, String> {
|
|
let current = ApplicableMountMode::parse(current)
|
|
.map_err(|error| format!("unsupported current mode: {error}"))?;
|
|
let target = ApplicableMountMode::parse(target)
|
|
.map_err(|error| format!("unsupported target mode: {error}"))?;
|
|
Ok((current != target).then_some(target))
|
|
}
|
|
|
|
fn restore_snapshot_matches(snapshot: &StoredSyncFolder, current: &StoredSyncFolder) -> bool {
|
|
current.id == snapshot.id
|
|
&& current.local_path == snapshot.local_path
|
|
&& current.remote_uid == snapshot.remote_uid
|
|
&& current.remote_share_id == snapshot.remote_share_id
|
|
&& current.mode == "ondemand"
|
|
}
|
|
|
|
fn pending_intent_matches(pending: Option<&str>, expected: Option<&str>) -> bool {
|
|
expected.is_none_or(|expected| pending == Some(expected))
|
|
}
|
|
|
|
fn commit_after_teardown<T, E>(
|
|
teardown: Result<(), E>,
|
|
commit: impl FnOnce() -> T,
|
|
) -> Result<T, E> {
|
|
teardown?;
|
|
Ok(commit())
|
|
}
|
|
|
|
/// 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)
|
|
}
|
|
|
|
/// The paths no sync root may overlap, with the mountpoint as the config says
|
|
/// now.
|
|
pub(crate) fn sync_root_reserved() -> pdfs_core::syncroot::Reserved {
|
|
let config = pdfs_core::config::AppDirs::new()
|
|
.map(|dirs| dirs.load_config())
|
|
.unwrap_or_default();
|
|
pdfs_core::syncroot::Reserved::current(&config)
|
|
}
|
|
|
|
/// 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}")))
|
|
}
|
|
|
|
/// The name a folder moved in from `device` takes when `name` is already
|
|
/// taken under this device's root, or `None` when it can keep its own.
|
|
fn migrated_name(name: &str, device: &str, taken: &[String]) -> Option<String> {
|
|
let free = |candidate: &str| !taken.iter().any(|t| t == candidate);
|
|
if free(name) {
|
|
return None;
|
|
}
|
|
let base = format!("{name} ({device})");
|
|
if free(&base) {
|
|
return Some(base);
|
|
}
|
|
(2..)
|
|
.map(|n| format!("{base} {n}"))
|
|
.find(|candidate| free(candidate))
|
|
}
|
|
|
|
impl Core {
|
|
// ---- devices ----------------------------------------------------------
|
|
|
|
/// Unified local-location listing for the control protocol.
|
|
pub(crate) fn list_locations(&self) -> CoreResult<Vec<MountSpec>> {
|
|
let progress = self.sync_progress.lock();
|
|
let mut locations = self
|
|
.db
|
|
.mount_list()
|
|
.map_err(|e| CoreError::internal(format!("db: {e:?}")))?;
|
|
for location in &mut locations {
|
|
location.mounted = self.states.is_mounted_at(Path::new(&location.local_path));
|
|
if let MountKind::Device { sync_folder_id } = &location.kind {
|
|
location.progress = progress.get(sync_folder_id).cloned();
|
|
location.paused = self.sync_folder_paused(*sync_folder_id);
|
|
}
|
|
}
|
|
Ok(locations)
|
|
}
|
|
|
|
/// List the account's registered devices, flagging the one *this* machine
|
|
/// syncs to so a front-end can treat it as more than another computer in the
|
|
/// list — deleting it takes this machine's synced folders down with it.
|
|
pub(crate) fn list_devices(&self) -> CoreResult<Vec<DeviceInfo>> {
|
|
let devices = self
|
|
.rt
|
|
.block_on(self.drive.enumerate_devices())
|
|
.map_err(|e| CoreError::from_api(&e, "list devices"))?;
|
|
// 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(),
|
|
last_sync: d.last_sync_time,
|
|
}
|
|
})
|
|
.collect())
|
|
}
|
|
|
|
/// Rename a device by its uid.
|
|
pub(crate) fn rename_device(&self, uid: &str, name: &str) -> CoreResult<()> {
|
|
if name.is_empty() {
|
|
return Err(CoreError::invalid("device name must not be empty"));
|
|
}
|
|
let device_uid = DeviceUid::from(uid);
|
|
self.rt
|
|
.block_on(self.client.rename_device(&device_uid, name))
|
|
.map_err(|e| CoreError::from_api(&e, "rename device"))?;
|
|
Ok(())
|
|
}
|
|
|
|
/// Delete (deregister) a device by its uid.
|
|
pub(crate) fn delete_device(&self, uid: &str) -> CoreResult<()> {
|
|
let device_uid = DeviceUid::from(uid);
|
|
self.rt
|
|
.block_on(self.client.delete_device(&device_uid))
|
|
.map_err(|e| CoreError::from_api(&e, "delete device"))?;
|
|
Ok(())
|
|
}
|
|
|
|
/// Adopt an existing device as *this* machine's, pinning it in `config.json`
|
|
/// so hostname changes and reinstalls stop registering duplicates.
|
|
/// `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.drive.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}"))
|
|
}
|
|
|
|
/// The device a restore reads from: this machine's own for `None`, else
|
|
/// the registered device with that uid.
|
|
///
|
|
/// Another device is looked up without touching the cached device row —
|
|
/// restoring its folders here must not make this machine back up as it.
|
|
/// Its folders are then attached with that device's share id, so they sync
|
|
/// against the other computer's tree.
|
|
pub(crate) fn restore_source(&self, device: Option<&str>) -> CoreResult<StoredDevice> {
|
|
let Some(uid) = device else {
|
|
return self.ensure_device();
|
|
};
|
|
let remote = self
|
|
.rt
|
|
.block_on(self.drive.enumerate_devices())
|
|
.map_err(|e| CoreError::from_api(&e, "enumerate devices"))?;
|
|
let d = remote
|
|
.into_iter()
|
|
.find(|d| d.uid.to_string() == uid)
|
|
.ok_or_else(|| CoreError::not_found(format!("no device {uid} in this account")))?;
|
|
Ok(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("(unnamed device)")
|
|
.to_string(),
|
|
created: d.creation_time,
|
|
})
|
|
}
|
|
|
|
/// Move every folder another device backs up into this machine's own
|
|
/// device, so this machine carries on backing them up as itself. The other
|
|
/// device stays registered, without those folders.
|
|
///
|
|
/// Device roots all live on the main volume, so each folder is one
|
|
/// server-side move: nothing is downloaded or uploaded, and no quota is
|
|
/// used twice. A folder whose name is taken here keeps its content under
|
|
/// "<name> (<device>)". A failed folder does not stop the others, and a
|
|
/// second run picks up whatever is still left on the other device.
|
|
///
|
|
/// Nothing is synced to this disk here. The moved folders are now this
|
|
/// device's, so the ordinary restore offers them. Only folders that were
|
|
/// already synced here from the other device are touched: their uid
|
|
/// survives the move but their share does not.
|
|
pub(crate) fn migrate_device(&self, uid: &str) -> CoreResult<String> {
|
|
let source = self.restore_source(Some(uid))?;
|
|
let target = self.ensure_device()?;
|
|
if source.uid == target.uid {
|
|
return Err(CoreError::invalid(format!(
|
|
"{} is this computer's own device",
|
|
source.name
|
|
)));
|
|
}
|
|
let bad_root = |d: &StoredDevice| {
|
|
parse_uid(&d.root_uid)
|
|
.ok_or_else(|| CoreError::internal(format!("bad device root uid: {}", d.root_uid)))
|
|
};
|
|
let (source_root, target_root) = (bad_root(&source)?, bad_root(&target)?);
|
|
if source_root.volume_id != target_root.volume_id {
|
|
return Err(CoreError::invalid(
|
|
"the two devices are on different Proton Drive volumes; copy it instead",
|
|
));
|
|
}
|
|
|
|
let mut taken: Vec<String> = self
|
|
.device_root_nodes(&target_root)?
|
|
.into_iter()
|
|
.filter(|n| !n.trashed)
|
|
.map(|n| n.name)
|
|
.collect();
|
|
let synced = self
|
|
.db
|
|
.sync_folder_list()
|
|
.map_err(|e| CoreError::internal(format!("db: {e:?}")))?;
|
|
|
|
let mut moved = 0usize;
|
|
let mut renamed: Vec<String> = Vec::new();
|
|
let mut left: Vec<String> = Vec::new();
|
|
let mut failed: Vec<String> = Vec::new();
|
|
for node in self.device_root_nodes(&source_root)? {
|
|
if node.trashed || node.name == PROFILE_DIR_NAME {
|
|
continue;
|
|
}
|
|
// A device root only ever holds folders (B68); anything else was
|
|
// put there by hand and is left for the user to deal with.
|
|
if !node.is_folder() {
|
|
left.push(node.name);
|
|
continue;
|
|
}
|
|
let new_name = migrated_name(&node.name, &source.name, &taken);
|
|
if let Err(e) = self.move_rename_remote(&node.uid, &target_root, new_name.as_deref()) {
|
|
warn!(name = %node.name, error = %e, "migrate: could not move folder");
|
|
failed.push(format!("{}: {e}", node.name));
|
|
continue;
|
|
}
|
|
let landed = new_name.clone().unwrap_or_else(|| node.name.clone());
|
|
if new_name.is_some() {
|
|
renamed.push(format!("{} -> {landed}", node.name));
|
|
}
|
|
taken.push(landed);
|
|
moved += 1;
|
|
|
|
let uid = node.uid.to_string();
|
|
for folder in synced.iter().filter(|f| f.remote_uid == uid) {
|
|
if let Err(e) = self.db.sync_folder_set_share(folder.id, &target.share_id) {
|
|
warn!(id = folder.id, error = ?e, "migrate: could not re-point sync folder");
|
|
}
|
|
self.sync_now(Some(folder.id));
|
|
}
|
|
}
|
|
self.invalidate_children_of(&source_root);
|
|
self.invalidate_children_of(&target_root);
|
|
|
|
let mut message = format!("moved {moved} folder(s) from {}", source.name);
|
|
if !renamed.is_empty() {
|
|
message.push_str(&format!("; renamed: {}", renamed.join(", ")));
|
|
}
|
|
if !left.is_empty() {
|
|
message.push_str(&format!("; left behind: {}", left.join(", ")));
|
|
}
|
|
if !failed.is_empty() {
|
|
message.push_str(&format!("; failed: {}", failed.join(", ")));
|
|
}
|
|
info!(from = %source.name, to = %target.name, moved, failed = failed.len(),
|
|
"device migrated");
|
|
if moved == 0 && !failed.is_empty() {
|
|
return Err(CoreError::conflict(message));
|
|
}
|
|
Ok(message)
|
|
}
|
|
|
|
// ---- device folder sync -------------------------
|
|
|
|
/// 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.
|
|
pub(crate) fn ensure_device(&self) -> CoreResult<StoredDevice> {
|
|
let name = this_hostname();
|
|
// 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.drive.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
|
|
// "parent node is not a folder". Re-register in that case.
|
|
if let Some(dev) = self
|
|
.db
|
|
.device_get()
|
|
.map_err(|e| CoreError::internal(format!("db: {e:?}")))?
|
|
{
|
|
if remote.iter().any(|d| d.uid.to_string() == dev.uid) {
|
|
return Ok(dev);
|
|
}
|
|
warn!(uid = %dev.uid, "cached device is gone remotely; re-registering");
|
|
}
|
|
|
|
// Recover: an existing remote Linux device with the same name is ours.
|
|
let existing = remote.into_iter().find(|d| {
|
|
d.device_type == DeviceType::Linux && d.name.as_deref().ok() == Some(name.as_str())
|
|
});
|
|
let dev = match existing {
|
|
Some(d) => StoredDevice {
|
|
uid: d.uid.to_string(),
|
|
share_id: d.share_id.to_string(),
|
|
root_uid: d.root_folder_uid.to_string(),
|
|
name,
|
|
created: d.creation_time,
|
|
},
|
|
None => {
|
|
let d = self
|
|
.rt
|
|
.block_on(self.drive.create_device(&name, DeviceType::Linux))
|
|
.map_err(|e| CoreError::from_api(&e, "create device"))?;
|
|
StoredDevice {
|
|
uid: d.uid.to_string(),
|
|
share_id: d.share_id.to_string(),
|
|
root_uid: d.root_folder_uid.to_string(),
|
|
name,
|
|
created: d.creation_time,
|
|
}
|
|
}
|
|
};
|
|
self.db
|
|
.device_set(&dev)
|
|
.map_err(|e| CoreError::internal(format!("db: {e:?}")))?;
|
|
Ok(dev)
|
|
}
|
|
|
|
/// An untrashed folder named `name` directly under the device root, if one
|
|
/// already exists.
|
|
pub(crate) fn find_device_child_folder(
|
|
&self,
|
|
root_uid: &NodeUid,
|
|
name: &str,
|
|
) -> CoreResult<Option<NodeUid>> {
|
|
Ok(self
|
|
.device_root_nodes(root_uid)?
|
|
.into_iter()
|
|
.find(|n| n.is_folder() && !n.trashed && n.name == name)
|
|
.map(|n| n.uid))
|
|
}
|
|
|
|
/// Everything directly under a device root, trashed nodes included.
|
|
fn device_root_nodes(&self, root_uid: &NodeUid) -> CoreResult<Vec<Node>> {
|
|
let uids = self
|
|
.rt
|
|
.block_on(self.drive.enumerate_folder_children_node_uids(root_uid))
|
|
.map_err(|e| CoreError::from_api(&e, "list device root"))?;
|
|
if uids.is_empty() {
|
|
return Ok(Vec::new());
|
|
}
|
|
self.rt
|
|
.block_on(self.drive.enumerate_nodes(&uids))
|
|
.map_err(|e| CoreError::from_api(&e, "resolve device root children"))
|
|
}
|
|
|
|
/// Add a local folder to this device's sync set: register the device if
|
|
/// needed, create a matching folder under the device root, upload the local
|
|
/// 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 meta = std::fs::metadata(local)
|
|
.map_err(|e| CoreError::internal(format!("stat {}: {e}", local.display())))?;
|
|
if !meta.is_dir() {
|
|
return Err(CoreError::invalid(format!(
|
|
"{} is not a directory",
|
|
local.display()
|
|
)));
|
|
}
|
|
let local = local
|
|
.canonicalize()
|
|
.map_err(|e| CoreError::internal(format!("canonicalize {}: {e}", local.display())))?;
|
|
let local_str = local.to_string_lossy().to_string();
|
|
|
|
// `/`, the home folder, the mount, this app's own data, and a folder
|
|
// inside or around another root are refused (`syncroot::problem`). A
|
|
// duplicate is one of those, refused here for a clear error before
|
|
// the UNIQUE constraint would catch it.
|
|
let others = self
|
|
.db
|
|
.sync_folder_list()
|
|
.map_err(|e| CoreError::internal(format!("db: {e:?}")))?
|
|
.iter()
|
|
.map(|f| PathBuf::from(&f.local_path))
|
|
.collect::<Vec<_>>();
|
|
let reserved = sync_root_reserved();
|
|
if let Some(problem) = pdfs_core::syncroot::problem(&local, &others, &reserved) {
|
|
return Err(CoreError::invalid(problem));
|
|
}
|
|
|
|
let name = local
|
|
.file_name()
|
|
.and_then(|n| n.to_str())
|
|
.ok_or_else(|| {
|
|
CoreError::invalid(format!("unusable folder name: {}", local.display()))
|
|
})?
|
|
.to_string();
|
|
|
|
// Syncing a local `.proton-drive-linux` would land on the folder holding
|
|
// this device's own profile (`find_device_child_folder` reuses by name),
|
|
// and the sync engine would then delete or overwrite it.
|
|
if name == pdfs_core::profile::PROFILE_DIR_NAME {
|
|
return Err(CoreError::invalid(format!(
|
|
"{name} is reserved for this device's profile backup"
|
|
)));
|
|
}
|
|
|
|
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))
|
|
})?;
|
|
|
|
// The synced folder's remote root: the folder under the device root named
|
|
// after the local basename. Reuse an existing one rather than creating a
|
|
// second folder with the same name — re-adding a folder (after a removal, or
|
|
// after a failed add that had already created it) must land back on the
|
|
// original, not leave the user with two "Downloads" in their Drive. The
|
|
// reconcile treats an existing remote tree correctly: unmatched paths read as
|
|
// a conflict, not as data loss.
|
|
let remote_root = match self.find_device_child_folder(&root_uid, &name)? {
|
|
Some(uid) => {
|
|
info!(name, "reusing existing device folder");
|
|
uid
|
|
}
|
|
None => self
|
|
.rt
|
|
.block_on(self.drive.create_folder(&root_uid, &name, Some(now_secs())))
|
|
.map_err(|e| CoreError::from_api(&e, &format!("create device folder {name}")))?,
|
|
};
|
|
|
|
let id = self
|
|
.db
|
|
.sync_folder_add(&local_str, &remote_root.to_string(), &device.share_id)
|
|
.map_err(|e| CoreError::internal(format!("db: {e:?}")))?;
|
|
|
|
// Hand the initial upload to the sync engine: an empty baseline against a
|
|
// full local tree reconciles as "upload everything", and the folder is
|
|
// added to the filesystem watch set in the same pass.
|
|
let _ = self.sync_tx.send(sync::SyncMsg::Rewatch);
|
|
let _ = self.sync_tx.send(sync::SyncMsg::Reconcile(id));
|
|
|
|
info!(local = %local.display(), id, "added sync folder");
|
|
self.db
|
|
.sync_folder_get(id)
|
|
.map_err(|e| CoreError::internal(format!("db: {e:?}")))?
|
|
.ok_or_else(|| CoreError::not_found("sync folder vanished after insert"))
|
|
}
|
|
|
|
/// List this device's synced folders for the front-ends, each carrying the
|
|
/// live progress of its pass when one is running.
|
|
pub(crate) fn list_sync_folders(&self) -> CoreResult<Vec<SyncFolderInfo>> {
|
|
let progress = self.sync_progress.lock();
|
|
Ok(self
|
|
.db
|
|
.sync_folder_list()
|
|
.map_err(|e| CoreError::internal(format!("db: {e:?}")))?
|
|
.into_iter()
|
|
.map(|f| {
|
|
let live = progress.get(&f.id).cloned();
|
|
let paused = self.sync_folder_paused(f.id);
|
|
sync_folder_info(f, live, paused)
|
|
})
|
|
.collect())
|
|
}
|
|
|
|
/// Everything the daemon is chewing on that isn't moving bytes, for
|
|
/// `GetQueueStatus`: the registered jobs (bulk-upload scans, the local index)
|
|
/// plus a synthesized job per running sync pass, so one Activity view answers
|
|
/// "is anything still happening?" without also polling `ListSyncFolders`.
|
|
///
|
|
/// The sync passes are folded in here rather than tracked as registry jobs
|
|
/// because the Devices page needs them per folder anyway
|
|
/// ([`SyncFolderInfo::progress`]) — this keeps one source of truth and hits
|
|
/// the db only while a pass is actually running.
|
|
pub(crate) fn jobs_snapshot(&self) -> Vec<JobItem> {
|
|
let mut jobs = self.transfers.jobs_snapshot();
|
|
let mut passes: Vec<(i64, SyncProgress)> = self
|
|
.sync_progress
|
|
.lock()
|
|
.iter()
|
|
.map(|(id, p)| (*id, p.clone()))
|
|
.collect();
|
|
if passes.is_empty() {
|
|
return jobs;
|
|
}
|
|
passes.sort_by_key(|(id, _)| *id);
|
|
|
|
let names: HashMap<i64, String> = self
|
|
.db
|
|
.sync_folder_list()
|
|
.unwrap_or_default()
|
|
.into_iter()
|
|
.map(|f| (f.id, f.local_path))
|
|
.collect();
|
|
for (id, p) in passes {
|
|
// The row is titled with the folder's own name; the full local path
|
|
// is what the Devices page shows, and is far too long for this line.
|
|
let folder = names
|
|
.get(&id)
|
|
.and_then(|path| Path::new(path).file_name())
|
|
.map(|n| n.to_string_lossy().into_owned())
|
|
.unwrap_or_else(|| "folder".to_string());
|
|
jobs.push(match p.phase {
|
|
// The scan's total is the last pass's baseline, so a folder that
|
|
// has never synced still reports indeterminate (`total: 0`) — but
|
|
// every later pass has a real bar. A grown folder can push `done`
|
|
// past the estimate; clamp so the row never reads "600 of 500".
|
|
SyncPhase::Scanning => JobItem {
|
|
title: format!("Checking {folder}"),
|
|
detail: "Looking for changes".to_string(),
|
|
done: p.done as u64,
|
|
total: if p.total == 0 {
|
|
0
|
|
} else {
|
|
p.total.max(p.done) as u64
|
|
},
|
|
// Most passes find nothing; only one that queues work is
|
|
// the drive syncing.
|
|
background: true,
|
|
},
|
|
SyncPhase::Applying => JobItem {
|
|
title: format!("Syncing {folder}"),
|
|
detail: p.current.clone(),
|
|
done: p.done as u64,
|
|
total: p.total.max(p.done) as u64,
|
|
background: false,
|
|
},
|
|
});
|
|
}
|
|
jobs
|
|
}
|
|
|
|
/// The lock guarding sync-folder `id` against concurrent reconcile/mode-switch.
|
|
pub(crate) fn sync_lock(&self, id: i64) -> Arc<Mutex<()>> {
|
|
self.sync_locks.lock().entry(id).or_default().clone()
|
|
}
|
|
|
|
/// Remove a synced folder from the sync set. `delete_remote` also deletes its
|
|
/// folder under the device root; otherwise the cloud copy is left in place.
|
|
pub(crate) fn remove_sync_folder(&self, id: i64, delete_remote: bool) -> CoreResult<()> {
|
|
let lock = self.sync_lock(id);
|
|
let _guard = lock.lock();
|
|
let folder = self
|
|
.db
|
|
.sync_folder_get(id)
|
|
.map_err(|e| CoreError::internal(format!("db: {e:?}")))?
|
|
.ok_or_else(|| CoreError::not_found(format!("no synced folder with id {id}")))?;
|
|
// An `ondemand` folder *is* a FUSE mount over its local path, so dropping
|
|
// only the row would strand the mount: the path would keep serving a
|
|
// 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.
|
|
// Taken in its own statement so the registry lock is released before the
|
|
// teardown: an edition-2024 `if let` chain holds the guard for the whole
|
|
// body, and `teardown` unmounts and joins the session thread — which can
|
|
// be inside a multi-second transfer. The shutdown path takes the same
|
|
// lock, so holding it here risks systemd's stop timeout landing mid-
|
|
// unmount and stranding the endpoint.
|
|
let taken = self.mounts.lock().remove(&id);
|
|
if let Some(mount) = taken
|
|
&& let Err(error) = commit_after_teardown(mount.teardown(), || ())
|
|
{
|
|
let _ = self.db.sync_folder_set_state(id, "error", now_secs());
|
|
return Err(CoreError::internal(format!(
|
|
"cannot remove synced folder: unmount {} failed: {error}",
|
|
folder.local_path
|
|
)));
|
|
}
|
|
info!(id, path = %folder.local_path, "unmounted on-demand folder");
|
|
if !self
|
|
.db
|
|
.sync_folder_remove(id)
|
|
.map_err(|e| CoreError::internal(format!("db: {e:?}")))?
|
|
{
|
|
return Err(CoreError::invalid(format!("no synced folder with id {id}")));
|
|
}
|
|
drop(_guard);
|
|
self.clear_sync_folder_paused(id);
|
|
// Stop watching the folder we just dropped.
|
|
let _ = self.sync_tx.send(sync::SyncMsg::Rewatch);
|
|
if delete_remote
|
|
&& let Some(uid) = parse_uid(&folder.remote_uid)
|
|
&& let Err(e) = self
|
|
.rt
|
|
.block_on(self.drive.trash_nodes(&[uid]))
|
|
.and_then(batch::into_unit)
|
|
{
|
|
warn!(id, error = %e, "delete remote device folder failed");
|
|
}
|
|
Ok(())
|
|
}
|
|
|
|
/// Trigger a reconcile: one folder by id, or every folder when `id` is `None`.
|
|
pub(crate) fn sync_now(&self, id: Option<i64>) {
|
|
let _ = match id {
|
|
Some(id) => self.sync_tx.send(sync::SyncMsg::Reconcile(id)),
|
|
None => self.sync_tx.send(sync::SyncMsg::ReconcileAll),
|
|
};
|
|
}
|
|
|
|
/// A sibling Core for a secondary FUSE session rooted at an `ondemand` sync
|
|
/// folder. It shares everything with this one,
|
|
/// the tree included, except what belongs to a session: its root and its
|
|
/// kernel channel, both filled in when the session is built.
|
|
pub(crate) fn fork_state(&self) -> Core {
|
|
let mut fork = self.clone();
|
|
fork.primary = false;
|
|
fork.root = Arc::new(OnceLock::new());
|
|
fork.notifier = Arc::new(OnceLock::new());
|
|
fork.session_live = Arc::new(std::sync::atomic::AtomicBool::new(false));
|
|
fork
|
|
}
|
|
|
|
/// Ask for a synced folder to move to `mode`, applying it now if the folder is
|
|
/// free and safe to switch, and **queueing** it otherwise. Returns the human
|
|
/// message for the reply.
|
|
///
|
|
/// Queueing rather than rejecting is what makes the toggle usable: a folder
|
|
/// that syncs continuously (a busy Downloads folder) is almost never caught in
|
|
/// the narrow window where it is both unlocked and `idle`, so a `try_lock`
|
|
/// rejection asks the user to keep retrying until they get lucky. Instead the
|
|
/// intent is recorded and the engine applies it at the end of the pass already
|
|
/// running — which, seeing a queued `ondemand`, also stops pulling down files
|
|
/// it is about to evict ([`Core::push_pass`]).
|
|
pub(crate) fn request_sync_folder_mode(&self, id: i64, mode: &str) -> CoreResult<String> {
|
|
// Preserve the existing control behavior: reject an unsupported request
|
|
// before looking up the folder, then validate the persisted source too.
|
|
let target_mode = ApplicableMountMode::parse(mode).map_err(CoreError::invalid)?;
|
|
let folder = self
|
|
.db
|
|
.sync_folder_get(id)
|
|
.map_err(|e| CoreError::internal(format!("db: {e:?}")))?
|
|
.ok_or_else(|| CoreError::not_found(format!("no synced folder with id {id}")))?;
|
|
let current_mode = ApplicableMountMode::parse(&folder.mode)
|
|
.map_err(|error| CoreError::invalid(format!("unsupported current mode: {error}")))?;
|
|
// A switch waits on the folder's reconcile, which a pause skips; refuse
|
|
// it up front rather than queue it indefinitely.
|
|
if current_mode != target_mode && self.sync_folder_paused(id) {
|
|
return Err(CoreError::invalid(
|
|
"this folder is paused; resume it before changing its mode",
|
|
));
|
|
}
|
|
if current_mode == target_mode {
|
|
let mode_str = target_mode.as_str();
|
|
let lock = self.sync_lock(id);
|
|
match lock.try_lock() {
|
|
Some(_guard) => {
|
|
let current = self
|
|
.db
|
|
.sync_folder_get(id)
|
|
.map_err(|e| CoreError::internal(format!("db: {e:?}")))?
|
|
.ok_or_else(|| {
|
|
CoreError::not_found(format!("no synced folder with id {id}"))
|
|
})?;
|
|
let locked_mode =
|
|
ApplicableMountMode::parse(¤t.mode).map_err(|error| {
|
|
CoreError::invalid(format!("unsupported current mode: {error}"))
|
|
})?;
|
|
if locked_mode == target_mode {
|
|
self.db
|
|
.sync_folder_set_pending_mode(id, None)
|
|
.map_err(|e| CoreError::internal(format!("db: {e:?}")))?;
|
|
let _ = self.sync_tx.send(sync::SyncMsg::Rewatch);
|
|
return Ok(if current.pending_mode.is_some() {
|
|
format!("staying {mode_str}")
|
|
} else {
|
|
format!("already {mode_str}")
|
|
});
|
|
}
|
|
}
|
|
None => {
|
|
// A transition or reconcile owns the folder lock. Record
|
|
// this request instead of clearing the old target: if a
|
|
// queued switch already passed its recheck, its mode write
|
|
// preserves this opposite marker and the engine switches
|
|
// back next.
|
|
self.db
|
|
.sync_folder_set_pending_mode(id, Some(mode_str))
|
|
.map_err(|e| CoreError::internal(format!("db: {e:?}")))?;
|
|
let _ = self.sync_tx.send(sync::SyncMsg::Reconcile(id));
|
|
return Ok(format!("staying {mode_str}"));
|
|
}
|
|
}
|
|
}
|
|
let mode = target_mode;
|
|
let mode_str = target_mode.as_str();
|
|
|
|
match self.apply_sync_folder_mode(id, mode, None) {
|
|
Ok(message) => Ok(message),
|
|
// Not switchable this instant — remember the intent instead of making
|
|
// the user retry, and kick a pass to clear whatever is in the way.
|
|
Err(SwitchBlocked::NotNow) => {
|
|
self.db
|
|
.sync_folder_set_pending_mode(id, Some(mode_str))
|
|
.map_err(|e| CoreError::internal(format!("db: {e:?}")))?;
|
|
let _ = self.sync_tx.send(sync::SyncMsg::Reconcile(id));
|
|
Ok(match mode {
|
|
ApplicableMountMode::OnDemand => format!(
|
|
"{} will go on-demand once its local changes are uploaded",
|
|
base_name(&folder.local_path)
|
|
),
|
|
ApplicableMountMode::Mirror => format!(
|
|
"{} will start mirroring once the current sync finishes",
|
|
base_name(&folder.local_path)
|
|
),
|
|
})
|
|
}
|
|
Err(SwitchBlocked::Superseded) => Err(CoreError::internal(
|
|
"direct mode request was unexpectedly superseded",
|
|
)),
|
|
Err(SwitchBlocked::Failed(e)) => Err(CoreError::internal(e)),
|
|
}
|
|
}
|
|
|
|
/// Apply a queued mode switch if the folder has one and is now able to take it.
|
|
/// Called by the sync engine after every pass, so a switch the user asked for
|
|
/// mid-sync lands as soon as the pass that blocked it is done. A folder that is
|
|
/// still not ready (its push failed, so the local copy is not safe to evict)
|
|
/// keeps its `pending_mode` and is retried after the next pass.
|
|
pub(crate) fn settle_pending_mode(&self, id: i64) {
|
|
let Ok(Some(folder)) = self.db.sync_folder_get(id) else {
|
|
return;
|
|
};
|
|
let Some(expected_mode) = folder.pending_mode.as_deref() else {
|
|
return;
|
|
};
|
|
let mode = match select_mode_transition(&folder.mode, expected_mode) {
|
|
Ok(Some(mode)) => mode,
|
|
Ok(None) => {
|
|
let _ = self.db.sync_folder_clear_pending_mode_if(id, expected_mode);
|
|
return;
|
|
}
|
|
Err(error) => {
|
|
warn!(
|
|
id,
|
|
current_mode = %folder.mode,
|
|
pending_mode = %expected_mode,
|
|
%error,
|
|
"withdrawing unsafe queued mode transition"
|
|
);
|
|
let _ = self.db.sync_folder_clear_pending_mode_if(id, expected_mode);
|
|
return;
|
|
}
|
|
};
|
|
let mode_str = mode.as_str();
|
|
match self.apply_sync_folder_mode(id, mode, Some(expected_mode)) {
|
|
Ok(message) => {
|
|
info!(id, mode = mode_str, "applied queued mode switch");
|
|
self.log_activity(ActivityKind::Sync, &message, "", true);
|
|
}
|
|
// Still blocked: the pass could not get everything up, so the local copy
|
|
// is not safe to evict yet. Leave the request standing — the next pass
|
|
// (poll, or the retry the engine schedules) tries again.
|
|
Err(SwitchBlocked::NotNow) => {
|
|
info!(id, mode = mode_str, "queued mode switch still waiting");
|
|
}
|
|
Err(SwitchBlocked::Superseded) => {
|
|
info!(id, mode = mode_str, "queued mode switch was replaced");
|
|
}
|
|
Err(SwitchBlocked::Failed(e)) => {
|
|
warn!(id, mode = mode_str, error = %e, "queued mode switch failed; withdrawing");
|
|
let _ = self.db.sync_folder_clear_pending_mode_if(id, expected_mode);
|
|
self.log_activity(
|
|
ActivityKind::Sync,
|
|
format!("couldn't switch {}", base_name(&folder.local_path)),
|
|
e,
|
|
false,
|
|
);
|
|
}
|
|
}
|
|
}
|
|
|
|
/// Flip a synced folder between `mirror` (full local copy + two-way sync) and
|
|
/// `ondemand` (a FUSE mount over the local path; no local storage). Returns a
|
|
/// human message on success.
|
|
///
|
|
/// - **mirror→ondemand**: require the folder in-sync (`idle`), stop watching it,
|
|
/// evict the local files to reclaim disk, then mount a secondary `ProtonFs`
|
|
/// rooted at the folder's remote node over its local path.
|
|
/// - **ondemand→mirror**: unmount, clear the stale baseline (the local tree was
|
|
/// evicted), then hand the folder back to the engine, which re-downloads it.
|
|
///
|
|
/// [`SwitchBlocked::NotNow`] means "not yet, try after a pass" and is never an
|
|
/// error the user needs to see — callers queue on it.
|
|
pub(crate) fn apply_sync_folder_mode(
|
|
&self,
|
|
id: i64,
|
|
mode: ApplicableMountMode,
|
|
expected_pending: Option<&str>,
|
|
) -> Result<String, SwitchBlocked> {
|
|
let mode_str = mode.as_str();
|
|
// Hold the folder's lock for the whole switch so no reconcile pass can be
|
|
// running over the tree we are about to evict (or start while we mount over
|
|
// it). A pass in flight holds the lock for its full duration, so `try_lock`
|
|
// failing is exactly "still syncing" — and it is the only reliable signal:
|
|
// the `state` column is still `idle` in the window between `add_sync_folder`
|
|
// inserting the row and the engine picking it up.
|
|
let lock = self.sync_lock(id);
|
|
let Some(_guard) = lock.try_lock() else {
|
|
return Err(SwitchBlocked::NotNow);
|
|
};
|
|
// Re-read under the lock: a pass that finished while we waited may have
|
|
// changed `state`.
|
|
let folder = self
|
|
.db
|
|
.sync_folder_get(id)
|
|
.map_err(|e| SwitchBlocked::Failed(format!("db: {e:?}")))?
|
|
.ok_or_else(|| SwitchBlocked::Failed(format!("no synced folder with id {id}")))?;
|
|
if !pending_intent_matches(folder.pending_mode.as_deref(), expected_pending) {
|
|
return Err(SwitchBlocked::Superseded);
|
|
}
|
|
let current_mode = ApplicableMountMode::parse(&folder.mode)
|
|
.map_err(|e| SwitchBlocked::Failed(format!("unsupported current mode: {e}")))?;
|
|
if current_mode == mode {
|
|
return Ok(format!("already {mode_str}"));
|
|
}
|
|
let local = PathBuf::from(&folder.local_path);
|
|
|
|
match mode {
|
|
ApplicableMountMode::OnDemand => {
|
|
// Only flip a folder that is fully in sync — a failed reconcile means
|
|
// local edits could still be un-uploaded, and we are about to delete
|
|
// the local copy. Not an error: a pass makes this true, and the queued
|
|
// request is applied once one does.
|
|
if folder.state != "idle" {
|
|
return Err(SwitchBlocked::NotNow);
|
|
}
|
|
// `idle` only says the last pass applied everything it tried.
|
|
// It never tries an ignored path, a symlink or a folder it
|
|
// cannot read, it defers a file open for writing, and a file can
|
|
// change after it ends. The eviction below deletes all of those
|
|
// for good, so it waits until nothing here is missing from Drive
|
|
// (B198).
|
|
let baseline = self
|
|
.db
|
|
.sync_entries(id)
|
|
.map_err(|e| SwitchBlocked::Failed(format!("db: {e:?}")))?;
|
|
if let Some(reason) =
|
|
sync::mirror_tree_unsynced(&local, &baseline, &self.ignore_rules(&local))
|
|
{
|
|
return Err(SwitchBlocked::Failed(format!(
|
|
"going online-only would delete what is only on this computer: {reason}"
|
|
)));
|
|
}
|
|
let root_uid = parse_uid(&folder.remote_uid).ok_or_else(|| {
|
|
SwitchBlocked::Failed(format!("bad remote uid: {}", folder.remote_uid))
|
|
})?;
|
|
let root = self
|
|
.rt
|
|
.block_on(self.drive.enumerate_nodes(std::slice::from_ref(&root_uid)))
|
|
.map_err(|e| SwitchBlocked::Failed(format!("fetch remote root: {e}")))?
|
|
.into_iter()
|
|
.next()
|
|
.ok_or_else(|| SwitchBlocked::Failed("remote folder not found".to_string()))?;
|
|
|
|
// Persist the mode *before* touching the local tree, and never roll it
|
|
// back on failure.
|
|
//
|
|
// The eviction below deletes every local file in the folder. While the
|
|
// row still says `mirror`, that empty directory is a valid input to the
|
|
// sync engine, which reads it as "the user deleted all of this" and
|
|
// trashes the whole folder on Drive. So the window between the evict and
|
|
// the mode write is a window in which any failure — a busy mountpoint, a
|
|
// missing fuse module, a dead DB — destroys the folder. Writing the mode
|
|
// first closes it: the engine skips non-mirror folders, so the worst a
|
|
// failure can now leave behind is an `ondemand` row with no mount, which
|
|
// is inert. Inert and marked `error` is recoverable; trashed is not.
|
|
//
|
|
// The same reasoning forbids reverting to `mirror` when a step below
|
|
// fails, since eviction may already have deleted part of the tree.
|
|
// Recovery is the user's explicit switch back to `mirror`, whose arm
|
|
// clears the baseline and re-downloads.
|
|
if let Err(e) = self.db.sync_folder_set_mode(id, "ondemand") {
|
|
return Err(SwitchBlocked::Failed(format!("db: {e:?}")));
|
|
}
|
|
|
|
// Reclaim the disk before mounting. Once FUSE covers `local`, a
|
|
// directory walk addresses the remote namespace, so evicting at
|
|
// that point sends unlink/rmdir requests to Drive and leaves the
|
|
// underlying mirror hidden and untouched.
|
|
if let Err(e) = evict_dir_contents(&local) {
|
|
warn!(id, path = %local.display(), error = %e, "evict local dir contents failed");
|
|
}
|
|
let mount = self.spawn_ondemand_mount(&local, root).map_err(|e| {
|
|
SwitchBlocked::Failed(format!(
|
|
"mount ondemand failed after local cache eviction; switch back to mirror to restore it: {e}"
|
|
))
|
|
})?;
|
|
self.publish_secondary_mount(id, mount)
|
|
.map_err(SwitchBlocked::Failed)?;
|
|
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");
|
|
Ok(format!("{} is now on-demand", local.display()))
|
|
}
|
|
ApplicableMountMode::Mirror => {
|
|
// ondemand→mirror: tear down the secondary mount first. Taken
|
|
// before the teardown so the registry lock is not held across
|
|
// the unmount and thread join (see `remove_sync_folder`).
|
|
let taken = self.mounts.lock().remove(&id);
|
|
if let Some(mount) = taken
|
|
&& let Err(error) = commit_after_teardown(mount.teardown(), || ())
|
|
{
|
|
let _ = self.db.sync_folder_set_state(id, "error", now_secs());
|
|
return Err(SwitchBlocked::Failed(format!(
|
|
"unmount {} failed; folder remains on-demand: {error}",
|
|
local.display()
|
|
)));
|
|
}
|
|
// The evicted local tree makes the old baseline lie ("everything
|
|
// deleted locally"); clear it so the reconcile is a pure download.
|
|
self.db
|
|
.sync_entries_clear(id)
|
|
.map_err(|e| SwitchBlocked::Failed(format!("db: {e:?}")))?;
|
|
self.db
|
|
.sync_folder_set_mode(id, "mirror")
|
|
.map_err(|e| SwitchBlocked::Failed(format!("db: {e:?}")))?;
|
|
let _ = self.sync_tx.send(sync::SyncMsg::Rewatch);
|
|
let _ = self.sync_tx.send(sync::SyncMsg::Reconcile(id));
|
|
info!(id, path = %local.display(), "restored sync folder to mirror");
|
|
Ok(format!(
|
|
"{} is mirroring again; downloading",
|
|
local.display()
|
|
))
|
|
}
|
|
}
|
|
}
|
|
|
|
/// Spawn a secondary FUSE session rooted at `root` over `local` from a
|
|
/// [`Core::fork_state`] sibling. Clears any stale kernel mount first (a crashed run can leave
|
|
/// one, which would fail the fresh mount with EBUSY).
|
|
pub(crate) fn spawn_ondemand_mount(
|
|
&self,
|
|
local: &Path,
|
|
root: Node,
|
|
) -> CoreResult<SecondaryMount> {
|
|
let core = self.fork_state();
|
|
let session_live = core.session_live.clone();
|
|
// The fork owns this state/path registration. `spawn_session` only
|
|
// marks it live after FUSE has spawned successfully.
|
|
core.register_state(local);
|
|
let session = spawn_session(&core, local, root)
|
|
.map_err(|e| CoreError::internal(format!("mount {}: {e}", local.display())))?;
|
|
Ok(SecondaryMount::new(
|
|
session,
|
|
fuse_connection_id(local),
|
|
session_live,
|
|
))
|
|
}
|
|
|
|
/// Publish one newly spawned secondary or explicitly tear it down when
|
|
/// shutdown has closed publication or another session already owns the id.
|
|
fn publish_secondary_mount(&self, id: i64, mount: SecondaryMount) -> Result<(), String> {
|
|
let rejected = self.mounts.lock().insert(id, mount);
|
|
let Err((reason, mount)) = rejected else {
|
|
return Ok(());
|
|
};
|
|
let reason = match reason {
|
|
SecondaryInsertRejection::Closed => "daemon is shutting down",
|
|
SecondaryInsertRejection::Duplicate => "folder already has a live mount",
|
|
};
|
|
match mount.teardown() {
|
|
Ok(()) => Err(reason.to_string()),
|
|
Err(error) => Err(format!("{reason}; rejected mount teardown failed: {error}")),
|
|
}
|
|
}
|
|
|
|
fn mark_restore_error_if_current(&self, snapshot: &StoredSyncFolder) {
|
|
let lock = self.sync_lock(snapshot.id);
|
|
let _guard = lock.lock();
|
|
if self
|
|
.db
|
|
.sync_folder_get(snapshot.id)
|
|
.ok()
|
|
.flatten()
|
|
.is_some_and(|current| restore_snapshot_matches(snapshot, ¤t))
|
|
{
|
|
let _ = self
|
|
.db
|
|
.sync_folder_set_state(snapshot.id, "error", now_secs());
|
|
}
|
|
}
|
|
|
|
/// Re-establish FUSE mounts for folders left in `ondemand` mode across a daemon
|
|
/// restart (their local dirs are empty on disk — the files live in the cloud).
|
|
/// Best-effort per folder: a missing local path or a failed remote fetch marks
|
|
/// the folder `error` and moves on rather than aborting the rest. Runs on its
|
|
/// own thread from `mount` so the network fetches never block startup.
|
|
pub(crate) fn restore_ondemand_mounts(&self) {
|
|
let folders = match self.db.sync_folder_list() {
|
|
Ok(f) => f,
|
|
Err(e) => {
|
|
warn!(error = ?e, "restore on-demand: cannot list folders");
|
|
return;
|
|
}
|
|
};
|
|
for folder in folders {
|
|
if folder.mode != "ondemand" {
|
|
continue;
|
|
}
|
|
let Some(root_uid) = parse_uid(&folder.remote_uid) else {
|
|
warn!(id = folder.id, "restore on-demand: bad remote uid");
|
|
self.mark_restore_error_if_current(&folder);
|
|
continue;
|
|
};
|
|
let root = match self
|
|
.rt
|
|
.block_on(self.drive.enumerate_nodes(std::slice::from_ref(&root_uid)))
|
|
{
|
|
Ok(v) => match v.into_iter().next() {
|
|
Some(n) => n,
|
|
None => {
|
|
warn!(id = folder.id, "restore on-demand: remote folder gone");
|
|
self.mark_restore_error_if_current(&folder);
|
|
continue;
|
|
}
|
|
},
|
|
Err(e) => {
|
|
warn!(id = folder.id, error = %e, "restore on-demand: fetch remote failed");
|
|
self.mark_restore_error_if_current(&folder);
|
|
continue;
|
|
}
|
|
};
|
|
let lock = self.sync_lock(folder.id);
|
|
let _guard = lock.lock();
|
|
let current = match self.db.sync_folder_get(folder.id) {
|
|
Ok(Some(current)) if restore_snapshot_matches(&folder, ¤t) => current,
|
|
Ok(_) => {
|
|
info!(
|
|
id = folder.id,
|
|
"restore on-demand: folder changed while fetching"
|
|
);
|
|
continue;
|
|
}
|
|
Err(error) => {
|
|
warn!(
|
|
id = folder.id,
|
|
?error,
|
|
"restore on-demand: cannot re-read folder"
|
|
);
|
|
continue;
|
|
}
|
|
};
|
|
if !self.mounts.lock().is_accepting() {
|
|
info!(id = folder.id, "restore on-demand: shutdown in progress");
|
|
continue;
|
|
}
|
|
let local = PathBuf::from(¤t.local_path);
|
|
// A daemon killed before it could unmount can leave a dead FUSE
|
|
// connection over the path. Clear it only after the row identity
|
|
// and mode have been revalidated under the folder lock.
|
|
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
|
|
.db
|
|
.sync_folder_set_state(folder.id, "error", now_secs());
|
|
continue;
|
|
}
|
|
match dir_is_empty(&local) {
|
|
Ok(true) => {}
|
|
Ok(false) => {
|
|
warn!(
|
|
id = folder.id,
|
|
path = %local.display(),
|
|
"restore on-demand: local dir is not empty; refusing to mount over it"
|
|
);
|
|
let _ = self
|
|
.db
|
|
.sync_folder_set_state(folder.id, "error", now_secs());
|
|
continue;
|
|
}
|
|
Err(e) => {
|
|
warn!(id = folder.id, path = %local.display(), error = %e, "restore on-demand: cannot read local dir");
|
|
let _ = self
|
|
.db
|
|
.sync_folder_set_state(folder.id, "error", now_secs());
|
|
continue;
|
|
}
|
|
}
|
|
match self.spawn_ondemand_mount(&local, root) {
|
|
Ok(mount) => match self.publish_secondary_mount(folder.id, mount) {
|
|
Ok(()) => {
|
|
let _ = self.db.sync_folder_set_state(folder.id, "idle", now_secs());
|
|
info!(id = folder.id, path = %local.display(), "remounted on-demand folder");
|
|
}
|
|
Err(error) => {
|
|
warn!(id = folder.id, %error, "restore on-demand: mount rejected");
|
|
}
|
|
},
|
|
Err(e) => {
|
|
warn!(id = folder.id, error = %e, "restore on-demand: mount failed");
|
|
let _ = self
|
|
.db
|
|
.sync_folder_set_state(folder.id, "error", now_secs());
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod tests {
|
|
use pdfs_core::db::StoredSyncFolder;
|
|
|
|
use super::{
|
|
ApplicableMountMode, commit_after_teardown, migrated_name, pending_intent_matches,
|
|
restore_snapshot_matches, select_mode_transition,
|
|
};
|
|
|
|
#[test]
|
|
fn migrated_folder_keeps_its_name_when_it_is_free() {
|
|
let taken = vec!["Pictures".to_string()];
|
|
assert_eq!(migrated_name("Documents", "DESKTOP-1", &taken), None);
|
|
}
|
|
|
|
#[test]
|
|
fn migrated_folder_is_named_after_its_device_on_a_clash() {
|
|
let taken = vec!["Documents".to_string()];
|
|
assert_eq!(
|
|
migrated_name("Documents", "DESKTOP-1", &taken).as_deref(),
|
|
Some("Documents (DESKTOP-1)")
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn migrated_folder_is_numbered_when_the_device_name_is_taken_too() {
|
|
let taken = vec![
|
|
"Documents".to_string(),
|
|
"Documents (DESKTOP-1)".to_string(),
|
|
"Documents (DESKTOP-1) 2".to_string(),
|
|
];
|
|
assert_eq!(
|
|
migrated_name("Documents", "DESKTOP-1", &taken).as_deref(),
|
|
Some("Documents (DESKTOP-1) 3")
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn unknown_mode_cannot_select_a_destructive_transition() {
|
|
let mut unmounts = 0;
|
|
let mut baseline_clears = 0;
|
|
let mut mode_writes = 0;
|
|
if let Ok(mode) = ApplicableMountMode::parse("streaming") {
|
|
match mode {
|
|
ApplicableMountMode::Mirror => {
|
|
unmounts += 1;
|
|
baseline_clears += 1;
|
|
mode_writes += 1;
|
|
}
|
|
ApplicableMountMode::OnDemand => {
|
|
mode_writes += 1;
|
|
}
|
|
}
|
|
}
|
|
assert_eq!((unmounts, baseline_clears, mode_writes), (0, 0, 0));
|
|
assert_eq!(
|
|
ApplicableMountMode::parse("streaming"),
|
|
Err("unknown mode \"streaming\" (want mirror|ondemand)".to_string())
|
|
);
|
|
assert_eq!(
|
|
ApplicableMountMode::parse("mirror"),
|
|
Ok(ApplicableMountMode::Mirror)
|
|
);
|
|
assert_eq!(
|
|
ApplicableMountMode::parse("ondemand"),
|
|
Ok(ApplicableMountMode::OnDemand)
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn unknown_source_mode_with_pending_mirror_cannot_select_a_transition() {
|
|
let mut destructive_helper_reached = false;
|
|
let selected = select_mode_transition("streaming", "mirror");
|
|
if let Ok(Some(_)) = selected {
|
|
destructive_helper_reached = true;
|
|
}
|
|
|
|
assert!(!destructive_helper_reached);
|
|
assert_eq!(
|
|
selected,
|
|
Err(
|
|
"unsupported current mode: unknown mode \"streaming\" (want mirror|ondemand)"
|
|
.to_string()
|
|
)
|
|
);
|
|
}
|
|
|
|
fn folder(local_path: &str, remote_uid: &str, mode: &str) -> StoredSyncFolder {
|
|
StoredSyncFolder {
|
|
id: 7,
|
|
local_path: local_path.to_string(),
|
|
remote_uid: remote_uid.to_string(),
|
|
remote_share_id: "share".to_string(),
|
|
mode: mode.to_string(),
|
|
pending_mode: None,
|
|
state: "idle".to_string(),
|
|
last_sync: 0,
|
|
}
|
|
}
|
|
|
|
#[test]
|
|
fn stale_restore_identity_or_mode_is_rejected() {
|
|
let snapshot = folder("/mnt/folder", "volume~node", "ondemand");
|
|
assert!(restore_snapshot_matches(
|
|
&snapshot,
|
|
&folder("/mnt/folder", "volume~node", "ondemand")
|
|
));
|
|
assert!(!restore_snapshot_matches(
|
|
&snapshot,
|
|
&folder("/mnt/replaced", "volume~node", "ondemand")
|
|
));
|
|
assert!(!restore_snapshot_matches(
|
|
&snapshot,
|
|
&folder("/mnt/folder", "volume~other", "ondemand")
|
|
));
|
|
assert!(!restore_snapshot_matches(
|
|
&snapshot,
|
|
&folder("/mnt/folder", "volume~node", "mirror")
|
|
));
|
|
}
|
|
|
|
#[test]
|
|
fn teardown_failure_prevents_follow_up_state_changes() {
|
|
let mut commits = 0;
|
|
let result = commit_after_teardown::<_, &str>(Err("busy"), || {
|
|
commits += 1;
|
|
});
|
|
assert_eq!(result, Err("busy"));
|
|
assert_eq!(commits, 0);
|
|
|
|
commit_after_teardown::<_, &str>(Ok(()), || commits += 1).unwrap();
|
|
assert_eq!(commits, 1);
|
|
}
|
|
|
|
#[test]
|
|
fn canceled_or_replaced_queued_intent_cannot_transition() {
|
|
assert!(pending_intent_matches(Some("ondemand"), Some("ondemand")));
|
|
assert!(!pending_intent_matches(None, Some("ondemand")));
|
|
assert!(!pending_intent_matches(Some("mirror"), Some("ondemand")));
|
|
assert!(pending_intent_matches(Some("mirror"), None));
|
|
}
|
|
}
|