mirror of
https://github.com/narrrl/proton-drive-linux.git
synced 2026-11-03 16:41:26 +01:00
nvrl
97277d41dd
A switch asked for before any pass saw the folder empty found it still idle, so the check for what only this computer has ran at once and refused it for good: every synced path read as "deleted here but not on Proton Drive yet". The account run hit it in the B255 case. A local copy gone in full is the wipe guard's case, which the push pass of a queued switch settles without trashing anything on Drive. The switch now returns NotNow for it, so the request is queued and that pass runs first. A partial deletion is still refused as before.
1451 lines
63 KiB
Rust
1451 lines
63 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:?}")))?;
|
|
let rules = self.ignore_rules(&local);
|
|
// Emptied before any pass saw it, the folder is still `idle`, and
|
|
// the check below would refuse the switch for good. The push pass
|
|
// of a queued switch forgets those paths instead (`docs/BUGS.md`
|
|
// B255).
|
|
if sync::mirror_tree_gone(&local, &baseline, &rules) {
|
|
return Err(SwitchBlocked::NotNow);
|
|
}
|
|
if let Some(reason) = sync::mirror_tree_unsynced(&local, &baseline, &rules) {
|
|
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()
|
|
))
|
|
}
|
|
}
|
|
}
|
|
|
|
/// Forget what a mirror folder synced before, so its next pass compares the
|
|
/// two sides as on the first sync: what one side has alone is copied to the
|
|
/// other, and a file that differs is kept twice as a conflict. Nothing is
|
|
/// deleted on either side.
|
|
///
|
|
/// This is the way out for a folder whose passes refuse to run because its
|
|
/// whole local copy, or its whole copy on Drive, is gone (#46). The baseline
|
|
/// is what makes that read as "delete everything", and it is what this
|
|
/// drops. A queued switch to on-demand is dropped with it: the push pass
|
|
/// that switch runs reads an unrecorded local file as new and would upload
|
|
/// a second copy of every file Drive already has.
|
|
pub(crate) fn reset_sync_folder(&self, id: i64) -> CoreResult<String> {
|
|
let lock = self.sync_lock(id);
|
|
let Some(_guard) = lock.try_lock() else {
|
|
return Err(CoreError::conflict(
|
|
"this folder is syncing; try again when the pass ends",
|
|
));
|
|
};
|
|
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}")))?;
|
|
if folder.mode != "mirror" {
|
|
return Err(CoreError::invalid(
|
|
"an online-only folder keeps nothing on this computer to sync from scratch",
|
|
));
|
|
}
|
|
self.db
|
|
.sync_entries_clear(id)
|
|
.map_err(|e| CoreError::internal(format!("db: {e:?}")))?;
|
|
if folder.pending_mode.as_deref() == Some("ondemand") {
|
|
self.db
|
|
.sync_folder_set_pending_mode(id, None)
|
|
.map_err(|e| CoreError::internal(format!("db: {e:?}")))?;
|
|
}
|
|
let _ = self.sync_tx.send(sync::SyncMsg::Reconcile(id));
|
|
info!(id, path = %folder.local_path, "sync folder reset");
|
|
Ok(format!(
|
|
"{} will sync from scratch",
|
|
base_name(&folder.local_path)
|
|
))
|
|
}
|
|
|
|
/// 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));
|
|
}
|
|
}
|