implemented shared with me + verification

This commit is contained in:
nvrl committed 2026-07-29 21:29:07 +02:00
1 parent 551e14b644
commit a8210ce8fd
16 files changed
+1830 -224

No files matched your search

+44 -2
View File
@@ -13,7 +13,7 @@ use pdfs_core::auth;
use pdfs_core::cache::ContentCache;
use pdfs_core::config::AppDirs;
use pdfs_core::control::{
ErrorKind, RefreshScope, Request as CtlRequest, Response as CtlResponse,
ErrorKind, MountKind, RefreshScope, Request as CtlRequest, Response as CtlResponse,
RestoreItem as CtlRestoreItem, ShareEntryKind, SyncPhase, pending_summary,
};
use pdfs_core::db::Db;
@@ -28,7 +28,7 @@ struct Cli {
/// Emit machine-readable JSON instead of formatted text.
///
/// Applies to the query commands (`status`, `ls`, `pins`, `sync list`,
/// `devices list`, `transfers`, `activity`, `cache inspect`). Commands that
/// `devices list`, `locations`, `transfers`, `activity`, `cache inspect`). Commands that
/// perform an action keep their human output — a script that needs to know
/// whether one succeeded has the exit code.
#[arg(long, global = true)]
@@ -105,6 +105,8 @@ enum Command {
Logout,
/// Show account login state, and mount status if a daemon is running.
Status,
/// List every local Proton Drive location.
Locations,
/// Mount Proton Drive at the given (or default) path. Blocks until unmounted.
Mount {
/// Mountpoint; defaults to ~/ProtonDrive.
@@ -507,6 +509,7 @@ fn main() -> Result<()> {
Command::Login { username } => cmd_login(username),
Command::Logout => cmd_logout(),
Command::Status => cmd_status(),
Command::Locations => cmd_locations(),
Command::Mount { mountpoint } => cmd_mount(mountpoint),
Command::Daemon { mountpoint } => cmd_daemon(mountpoint),
Command::Pin { path } => cmd_pin(path),
@@ -1201,6 +1204,45 @@ fn cmd_status() -> Result<()> {
Ok(())
}
fn cmd_locations() -> Result<()> {
let response = control_request(CtlRequest::ListLocations)?;
if emit_json(&response)? {
return Ok(());
}
match response {
CtlResponse::Locations { items } if items.is_empty() => {
println!("No Proton Drive locations.");
}
CtlResponse::Locations { items } => {
for location in items {
let kind = match &location.kind {
MountKind::MyFiles => "My Files".to_string(),
MountKind::Device { sync_folder_id } => {
format!("Device folder {sync_folder_id}")
}
MountKind::Shared { .. } => "Shared folder".to_string(),
};
let mode = match location.pending_mode {
Some(pending) => format!("{} -> {pending}", location.mode),
None => location.mode.to_string(),
};
let mounted = if location.mounted {
"mounted"
} else {
"not mounted"
};
println!(
"{kind}: {} ({mode}, {}, {}, {mounted})",
location.local_path, location.access, location.state
);
}
}
CtlResponse::Error { message, kind } => bail!("{}", cli_error(kind, &message)),
other => bail!("unexpected response: {other:?}"),
}
Ok(())
}
fn cmd_mount(mountpoint: Option<PathBuf>) -> Result<()> {
mount_once(mountpoint)?;
Ok(())
+110
View File
@@ -14,6 +14,7 @@ use serde::{Deserialize, Serialize};
use crate::cache::Pin;
use crate::error::Result;
pub use crate::mounts::{MountAccess, MountKind, MountMode, MountSpec};
/// Cap on the *write* half of a round-trip. A crashed daemon can leave its
/// listening socket in the kernel (e.g. as a `<defunct>` zombie): `connect`
@@ -194,6 +195,11 @@ pub enum Request {
/// it invalidates, it does not fetch.
Refresh { scope: RefreshScope },
// ---- locations --------------------------------------------------------
/// List every local Proton Drive location. Replies with
/// [`Response::Locations`].
ListLocations,
// ---- devices ----------------------------------------------------------
/// List the account's registered devices. Replies with [`Response::Devices`].
ListDevices,
@@ -1127,6 +1133,8 @@ pub enum Response {
},
/// The account's devices (reply to [`Request::ListDevices`]).
Devices { items: Vec<DeviceInfo> },
/// Every local Proton Drive location (reply to [`Request::ListLocations`]).
Locations { items: Vec<MountSpec> },
/// This device's synced folders (reply to [`Request::ListSyncFolders`]).
SyncFolders { items: Vec<SyncFolderInfo> },
/// Folders offered for restore (reply to [`Request::ListRestorableFolders`]).
@@ -1574,6 +1582,7 @@ mod tests {
#[test]
fn sharing_requests_roundtrip() {
let reqs = [
Request::ListLocations,
Request::ListDevices,
Request::RenameDevice {
uid: "dev-1".into(),
@@ -1698,6 +1707,107 @@ mod tests {
}
}
#[test]
fn locations_response_roundtrips_typed_mounts() {
let response = Response::Locations {
items: vec![
MountSpec {
id: 1,
kind: MountKind::MyFiles,
local_path: "/home/me/ProtonDrive".into(),
root_uid: "vol~root".into(),
root_share_id: "share-main".into(),
mode: MountMode::OnDemand,
access: MountAccess::Rw,
state: "idle".into(),
last_sync: 0,
pending_mode: None,
mounted: true,
progress: None,
},
MountSpec {
id: 2,
kind: MountKind::Device { sync_folder_id: 7 },
local_path: "/home/me/Work".into(),
root_uid: "device-vol~work".into(),
root_share_id: "device-share".into(),
mode: MountMode::Mirror,
access: MountAccess::Rw,
state: "syncing".into(),
last_sync: 42,
pending_mode: Some(MountMode::OnDemand),
mounted: false,
progress: None,
},
MountSpec {
id: 3,
kind: MountKind::Shared {
share_root_uid: "shared-vol~root".into(),
},
local_path: "/home/me/Shared".into(),
root_uid: "shared-vol~root".into(),
root_share_id: "shared-share".into(),
mode: MountMode::Unknown,
access: MountAccess::Ro,
state: "error".into(),
last_sync: 0,
pending_mode: Some(MountMode::Unknown),
mounted: false,
progress: Some(SyncProgress {
phase: SyncPhase::Applying,
done: 2,
total: 5,
current: "report.pdf".into(),
}),
},
],
};
let wire = serde_json::to_string(&response).unwrap();
assert!(!wire.contains('\n'), "wire form must be a single line");
assert!(wire.contains(r#""mode":"ondemand""#));
assert!(wire.contains(r#""kind":"shared""#));
assert!(wire.contains(r#""access":"ro""#));
assert!(wire.contains(r#""mode":"unknown""#));
let back: Response = serde_json::from_str(&wire).unwrap();
match back {
Response::Locations { items } => {
assert_eq!(items.len(), 3);
assert!(matches!(
items[1].kind,
MountKind::Device { sync_folder_id: 7 }
));
assert_eq!(items[1].pending_mode, Some(MountMode::OnDemand));
assert!(matches!(
&items[2].kind,
MountKind::Shared { share_root_uid }
if share_root_uid == "shared-vol~root"
));
assert_eq!(items[2].access, MountAccess::Ro);
assert_eq!(items[2].mode, MountMode::Unknown);
assert_eq!(items[2].pending_mode, Some(MountMode::Unknown));
let progress = items[2].progress.as_ref().unwrap();
assert_eq!(progress.phase, SyncPhase::Applying);
assert_eq!((progress.done, progress.total), (2, 5));
}
other => panic!("unexpected response: {other:?}"),
}
}
#[test]
fn legacy_daemons_reject_list_locations_cleanly() {
#[allow(dead_code)]
#[derive(serde::Deserialize)]
enum LegacyLocationRequest {
ListSyncFolders,
}
let wire = serde_json::to_string(&Request::ListLocations).unwrap();
assert!(
serde_json::from_str::<LegacyLocationRequest>(&wire).is_err(),
"an old daemon must reject ListLocations instead of interpreting another request"
);
}
#[test]
fn path_share_requests_keep_their_existing_wire_shape() {
let cases = [
+1
View File
@@ -77,6 +77,7 @@ const REPORTED_TABLES: &[&str] = &[
"trash",
"device",
"sync_folder",
"mount",
"sync_entry",
"pending_op",
"activity",
+48 -1
View File
@@ -9,7 +9,7 @@ use super::Db;
use crate::Result;
/// Current schema version. Bump on every forward migration added below.
pub(super) const SCHEMA_VERSION: i64 = 17;
pub(super) const SCHEMA_VERSION: i64 = 18;
impl Db {
pub(super) fn migrate(&self) -> Result<()> {
@@ -92,6 +92,9 @@ impl Db {
if current < 17 {
tx.execute_batch(MIGRATION_V17)?;
}
if current < 18 {
tx.execute_batch(MIGRATION_V18)?;
}
tx.execute(
"INSERT INTO sync_state (key, value) VALUES ('schema_version', ?1)
ON CONFLICT(key) DO UPDATE SET value = excluded.value",
@@ -425,3 +428,47 @@ CREATE TABLE share_access (
CHECK (access IN ('owner', 'editor', 'viewer', 'unknown'))
);
";
/// Schema v18: unified location presentation. Device rows carry only their
/// `sync_folder` identity; reads join all device state from that authoritative
/// table. The trigger projects future folders, while the foreign key removes a
/// presentation row with its sync folder.
const MIGRATION_V18: &str = "
CREATE TABLE mount (
id INTEGER PRIMARY KEY,
kind TEXT NOT NULL
CHECK (kind IN ('myfiles', 'device', 'shared')),
sync_folder_id INTEGER REFERENCES sync_folder(id) ON DELETE CASCADE,
share_root_uid TEXT,
local_path TEXT NOT NULL DEFAULT '',
root_uid TEXT NOT NULL DEFAULT '',
root_share_id TEXT NOT NULL DEFAULT '',
mode TEXT NOT NULL DEFAULT 'unknown',
access TEXT NOT NULL DEFAULT 'rw'
CHECK (access IN ('rw', 'ro')),
CHECK (
(kind = 'myfiles' AND sync_folder_id IS NULL AND share_root_uid IS NULL)
OR
(kind = 'device' AND sync_folder_id IS NOT NULL AND share_root_uid IS NULL)
OR
(kind = 'shared' AND sync_folder_id IS NULL AND share_root_uid IS NOT NULL)
)
);
CREATE UNIQUE INDEX mount_myfiles
ON mount(kind) WHERE kind = 'myfiles';
CREATE UNIQUE INDEX mount_device
ON mount(sync_folder_id) WHERE kind = 'device';
CREATE UNIQUE INDEX mount_shared
ON mount(share_root_uid) WHERE kind = 'shared';
INSERT INTO mount (kind, sync_folder_id)
SELECT 'device', id FROM sync_folder;
CREATE TRIGGER mount_sync_folder_insert
AFTER INSERT ON sync_folder
BEGIN
INSERT OR IGNORE INTO mount (kind, sync_folder_id)
VALUES ('device', NEW.id);
END;
";
+5
View File
@@ -27,6 +27,7 @@ mod devices;
mod local;
mod maintenance;
mod migrations;
mod mounts;
mod nodes;
mod ops;
mod photos;
@@ -110,6 +111,10 @@ impl Db {
#[cfg(test)]
pub fn open_in_memory() -> Result<Self> {
let conn = Connection::open_in_memory()?;
// Keep relational behavior identical to `open`: V18 relies on the
// `mount.sync_folder_id` cascade, and SQLite disables foreign keys per
// connection unless explicitly enabled.
conn.pragma_update(None, "foreign_keys", "ON")?;
let db = Self {
conn: Mutex::new(conn),
};
+178
View File
@@ -0,0 +1,178 @@
//! Unified mount/location presentation rows.
//!
//! A device row stores only its `sync_folder` identity. Its path, remote root,
//! mode, state, pending mode, and last-sync timestamp are selected from
//! `sync_folder`, which remains the authoritative sync-engine table.
use rusqlite::params;
use super::Db;
use crate::mounts::{MountAccess, MountKind, MountMode, MountSpec};
use crate::{Error, Result};
impl Db {
/// Project the configured primary mount into the presentation table.
///
/// `root_share_id` is optional because an offline startup may only have the
/// cached root node. A previously learned share id is retained only when it
/// belongs to the same root uid.
pub fn mount_upsert_my_files(
&self,
local_path: &str,
root_uid: &str,
root_share_id: Option<&str>,
) -> Result<i64> {
let mut conn = self.conn.lock();
let tx = conn.transaction()?;
let updated = tx.execute(
"UPDATE mount
SET local_path = ?1,
root_share_id = CASE
WHEN root_uid = ?2
THEN COALESCE(NULLIF(?3, ''), root_share_id)
ELSE COALESCE(?3, '')
END,
root_uid = ?2,
mode = 'ondemand',
access = 'rw'
WHERE kind = 'myfiles'",
params![local_path, root_uid, root_share_id],
)?;
let id = if updated == 0 {
tx.execute(
"INSERT INTO mount
(kind, local_path, root_uid, root_share_id, mode, access)
VALUES ('myfiles', ?1, ?2, COALESCE(?3, ''), 'ondemand', 'rw')",
params![local_path, root_uid, root_share_id],
)?;
tx.last_insert_rowid()
} else {
tx.query_row("SELECT id FROM mount WHERE kind = 'myfiles'", [], |row| {
row.get(0)
})?
};
tx.commit()?;
Ok(id)
}
/// Store a share id learned asynchronously, but only if the projection still
/// names the root it was resolved for.
pub fn mount_repair_my_files_share_id(
&self,
root_uid: &str,
root_share_id: &str,
) -> Result<bool> {
let conn = self.conn.lock();
Ok(conn.execute(
"UPDATE mount
SET root_share_id = ?2
WHERE kind = 'myfiles'
AND root_uid = ?1",
params![root_uid, root_share_id],
)? > 0)
}
/// Every local location, with device fields joined from `sync_folder`.
pub fn mount_list(&self) -> Result<Vec<MountSpec>> {
let conn = self.conn.lock();
let mut stmt = conn.prepare(
"SELECT
m.id,
m.kind,
m.sync_folder_id,
m.share_root_uid,
CASE WHEN m.kind = 'device' THEN sf.local_path ELSE m.local_path END,
CASE WHEN m.kind = 'device' THEN sf.remote_uid ELSE m.root_uid END,
CASE WHEN m.kind = 'device' THEN sf.remote_share_id ELSE m.root_share_id END,
CASE WHEN m.kind = 'device' THEN sf.mode ELSE m.mode END,
m.access,
CASE WHEN m.kind = 'device' THEN sf.state ELSE 'idle' END,
CASE WHEN m.kind = 'device' THEN sf.last_sync ELSE 0 END,
CASE WHEN m.kind = 'device' THEN sf.pending_mode ELSE NULL END
FROM mount AS m
LEFT JOIN sync_folder AS sf ON sf.id = m.sync_folder_id
ORDER BY CASE m.kind
WHEN 'myfiles' THEN 0
WHEN 'device' THEN 1
ELSE 2
END,
m.id",
)?;
let rows = stmt.query_map([], |row| {
Ok((
row.get::<_, i64>(0)?,
row.get::<_, String>(1)?,
row.get::<_, Option<i64>>(2)?,
row.get::<_, Option<String>>(3)?,
row.get::<_, String>(4)?,
row.get::<_, String>(5)?,
row.get::<_, String>(6)?,
row.get::<_, String>(7)?,
row.get::<_, String>(8)?,
row.get::<_, String>(9)?,
row.get::<_, i64>(10)?,
row.get::<_, Option<String>>(11)?,
))
})?;
let mut mounts = Vec::new();
for row in rows {
let (
id,
kind,
sync_folder_id,
share_root_uid,
local_path,
root_uid,
root_share_id,
mode,
access,
state,
last_sync,
pending_mode,
) = row?;
let kind = match kind.as_str() {
"myfiles" => MountKind::MyFiles,
"device" => MountKind::Device {
sync_folder_id: sync_folder_id.ok_or_else(|| {
Error::Other(format!("mount row {id} has no sync_folder_id"))
})?,
},
"shared" => MountKind::Shared {
share_root_uid: share_root_uid.ok_or_else(|| {
Error::Other(format!("mount row {id} has no share_root_uid"))
})?,
},
other => {
return Err(Error::Other(format!(
"mount row {id} has unknown kind {other:?}"
)));
}
};
let access = match access.as_str() {
"rw" => MountAccess::Rw,
"ro" => MountAccess::Ro,
other => {
return Err(Error::Other(format!(
"mount row {id} has unknown access {other:?}"
)));
}
};
mounts.push(MountSpec {
id,
kind,
local_path,
root_uid,
root_share_id,
mode: MountMode::from(mode.as_str()),
access,
state,
last_sync,
pending_mode: pending_mode.as_deref().map(MountMode::from),
mounted: false,
progress: None,
});
}
Ok(mounts)
}
}
+20 -5
View File
@@ -115,14 +115,17 @@ impl Db {
Ok(n > 0)
}
/// Update a synced folder's mode (`mirror`/`ondemand`). The folder has
/// reached the mode it was asked for, so any queued request is satisfied and
/// cleared in the same write — a `pending_mode` outliving the switch it asked
/// for would have the engine try to apply it again on the next pass.
/// Update a synced folder's mode (`mirror`/`ondemand`). A queued request for
/// that mode is satisfied and cleared in the same write. An opposite pending
/// mode arrived after this transition started, so preserve it as the newer
/// last-request intent.
pub fn sync_folder_set_mode(&self, id: i64, mode: &str) -> Result<()> {
let conn = self.conn.lock();
conn.execute(
"UPDATE sync_folder SET mode = ?2, pending_mode = NULL WHERE id = ?1",
"UPDATE sync_folder
SET mode = ?2,
pending_mode = CASE WHEN pending_mode = ?2 THEN NULL ELSE pending_mode END
WHERE id = ?1",
params![id, mode],
)?;
Ok(())
@@ -139,6 +142,18 @@ impl Db {
Ok(())
}
/// Withdraw a queued mode only if it is still the intent the caller
/// inspected. A newer request must survive an older settlement attempt.
pub fn sync_folder_clear_pending_mode_if(&self, id: i64, expected: &str) -> Result<bool> {
let conn = self.conn.lock();
let changed = conn.execute(
"UPDATE sync_folder SET pending_mode = NULL
WHERE id = ?1 AND pending_mode = ?2",
params![id, expected],
)?;
Ok(changed > 0)
}
/// Update a synced folder's state and stamp `last_sync` to now.
pub fn sync_folder_set_state(&self, id: i64, state: &str, last_sync: i64) -> Result<()> {
let conn = self.conn.lock();
+254 -4
View File
@@ -1323,15 +1323,17 @@ fn migration_v17_adds_share_access_to_a_v16_fixture() {
.as_nanos()
));
{
// V17 only adds `share_access`, so removing it and restoring the version
// stamp produces the exact schema a released V16 database had.
// Remove the later tables too, so restoring the version stamp produces
// the exact schema a released V16 database had.
let db = Db::open(&path).unwrap();
db.set_state_str("fixture_value", "survives").unwrap();
}
{
let conn = rusqlite::Connection::open(&path).unwrap();
conn.execute_batch(
"DROP TABLE share_access;
"DROP TRIGGER mount_sync_folder_insert;
DROP TABLE mount;
DROP TABLE share_access;
UPDATE sync_state SET value = '16' WHERE key = 'schema_version';",
)
.unwrap();
@@ -1347,13 +1349,211 @@ fn migration_v17_adds_share_access_to_a_v16_fixture() {
assert_eq!(db.share_access(&root).unwrap(), Some(crate::Access::Viewer));
assert_eq!(
db.state_str("schema_version").unwrap().as_deref(),
Some("17")
Some("18")
);
drop(db);
let _ = std::fs::remove_file(path);
}
#[test]
fn migration_v18_projects_v17_sync_folders_and_cascades_deletes() {
let path = std::env::temp_dir().join(format!(
"pdfs-db-v17-fixture-{}-{}.db",
std::process::id(),
std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap()
.as_nanos()
));
{
// Pin the released V17 shape that V18 reads instead of deriving a
// fixture from today's head schema and dropping newer objects.
let conn = rusqlite::Connection::open(&path).unwrap();
conn.execute_batch(
"CREATE TABLE sync_state (key TEXT PRIMARY KEY, value TEXT);
INSERT INTO sync_state VALUES ('schema_version', '17');
INSERT INTO sync_state VALUES ('fixture_value', 'survives');
CREATE TABLE sync_folder (
id INTEGER PRIMARY KEY,
local_path TEXT NOT NULL UNIQUE,
remote_uid TEXT NOT NULL,
remote_share_id TEXT NOT NULL,
mode TEXT NOT NULL DEFAULT 'mirror',
state TEXT NOT NULL DEFAULT 'idle',
last_sync INTEGER NOT NULL DEFAULT 0,
pending_mode TEXT
);
CREATE TABLE sync_entry (
folder_id INTEGER NOT NULL,
rel_path TEXT NOT NULL,
remote_uid TEXT,
local_mtime INTEGER NOT NULL DEFAULT 0,
local_size INTEGER NOT NULL DEFAULT 0,
remote_hash TEXT,
remote_rev TEXT,
PRIMARY KEY (folder_id, rel_path)
);
INSERT INTO sync_folder
(id, local_path, remote_uid, remote_share_id, mode, state, last_sync)
VALUES
(41, '/home/me/Existing', 'vol~existing', 'share-existing',
'ondemand', 'syncing', 123);
CREATE TABLE share_access (
root_uid TEXT PRIMARY KEY,
access TEXT NOT NULL
CHECK (access IN ('owner', 'editor', 'viewer', 'unknown'))
);",
)
.unwrap();
}
let db = Db::open(&path).unwrap();
let existing = 41;
let locations = db.mount_list().unwrap();
assert_eq!(locations.len(), 1);
let projected = &locations[0];
assert!(matches!(
projected.kind,
crate::mounts::MountKind::Device {
sync_folder_id
} if sync_folder_id == existing
));
assert_eq!(projected.local_path, "/home/me/Existing");
assert_eq!(projected.mode, crate::mounts::MountMode::OnDemand);
assert_eq!(projected.state, "syncing");
assert_eq!(projected.last_sync, 123);
assert_eq!(
db.state_str("fixture_value").unwrap().as_deref(),
Some("survives")
);
let added = db
.sync_folder_add("/home/me/New", "vol~new", "share-new")
.unwrap();
assert_eq!(db.mount_list().unwrap().len(), 2);
assert!(db.sync_folder_remove(added).unwrap());
assert_eq!(
db.mount_list().unwrap().len(),
1,
"the V18 foreign key must remove the device projection"
);
assert_eq!(
db.state_str("schema_version").unwrap().as_deref(),
Some("18")
);
drop(db);
let _ = std::fs::remove_file(path);
}
#[test]
fn my_files_projection_retains_share_id_only_for_the_same_root() {
let db = Db::open_in_memory().unwrap();
let id = db
.mount_upsert_my_files("/home/me/ProtonDrive", "vol~root", Some("share-main"))
.unwrap();
let same_id = db
.mount_upsert_my_files("/mnt/proton", "vol~root", None)
.unwrap();
assert_eq!(same_id, id);
assert_eq!(db.mount_list().unwrap()[0].root_share_id, "share-main");
db.mount_upsert_my_files("/mnt/proton", "vol~root-new", None)
.unwrap();
let locations = db.mount_list().unwrap();
assert_eq!(locations.len(), 1);
assert!(matches!(
locations[0].kind,
crate::mounts::MountKind::MyFiles
));
assert_eq!(locations[0].local_path, "/mnt/proton");
assert_eq!(locations[0].root_uid, "vol~root-new");
assert_eq!(
locations[0].root_share_id, "",
"a share id must never be paired with a different root uid"
);
}
#[test]
fn my_files_share_id_repair_is_conditional_on_root_uid() {
let db = Db::open_in_memory().unwrap();
db.mount_upsert_my_files("/mnt/proton", "vol~root", None)
.unwrap();
assert!(
!db.mount_repair_my_files_share_id("vol~stale", "share-stale")
.unwrap()
);
assert_eq!(db.mount_list().unwrap()[0].root_share_id, "");
assert!(
db.mount_repair_my_files_share_id("vol~root", "share-current")
.unwrap()
);
assert_eq!(db.mount_list().unwrap()[0].root_share_id, "share-current");
}
#[test]
fn mount_constraints_reject_malformed_kinds_and_missing_sync_folders() {
let db = Db::open_in_memory().unwrap();
let conn = db.conn.lock();
assert!(
conn.execute(
"INSERT INTO mount (kind, local_path, root_uid)
VALUES ('device', '/tmp/device', 'vol~device')",
[],
)
.is_err(),
"a device projection without sync_folder_id must fail its CHECK"
);
assert!(
conn.execute(
"INSERT INTO mount (kind, sync_folder_id)
VALUES ('device', 999999)",
[],
)
.is_err(),
"a device projection cannot reference a missing sync folder"
);
assert!(
conn.execute(
"INSERT INTO mount (kind, share_root_uid, local_path, root_uid)
VALUES ('myfiles', 'vol~shared', '/tmp/root', 'vol~root')",
[],
)
.is_err(),
"kind-specific columns must not be mixed"
);
}
#[test]
fn deleting_a_device_projection_does_not_delete_its_sync_folder() {
let db = Db::open_in_memory().unwrap();
let folder_id = db
.sync_folder_add("/home/me/Keep", "vol~keep", "share-keep")
.unwrap();
{
let conn = db.conn.lock();
assert_eq!(
conn.execute(
"DELETE FROM mount WHERE kind = 'device' AND sync_folder_id = ?1",
[folder_id],
)
.unwrap(),
1
);
}
assert!(db.sync_folder_get(folder_id).unwrap().is_some());
assert!(
db.mount_list().unwrap().is_empty(),
"the foreign key direction must not recreate or delete sync authority"
);
}
#[test]
fn refuses_a_schema_newer_than_the_running_build() {
let path = std::env::temp_dir().join(format!(
@@ -1953,6 +2153,56 @@ fn pending_mode_is_queued_until_the_mode_is_reached() {
assert_eq!(folder.pending_mode, None);
}
#[test]
fn mode_commit_clears_only_the_intent_it_satisfies() {
let db = Db::open_in_memory().unwrap();
let id = db
.sync_folder_add("/tmp/pdfs-mode-intent", "vol~folder", "share")
.unwrap();
db.sync_folder_set_pending_mode(id, Some("ondemand"))
.unwrap();
db.sync_folder_set_mode(id, "ondemand").unwrap();
assert_eq!(db.sync_folder_get(id).unwrap().unwrap().pending_mode, None);
db.sync_folder_set_pending_mode(id, Some("mirror")).unwrap();
db.sync_folder_set_mode(id, "ondemand").unwrap();
assert_eq!(
db.sync_folder_get(id)
.unwrap()
.unwrap()
.pending_mode
.as_deref(),
Some("mirror")
);
}
#[test]
fn clearing_a_stale_pending_mode_preserves_the_newer_intent() {
let db = Db::open_in_memory().unwrap();
let id = db
.sync_folder_add("/tmp/pdfs-pending", "vol~folder", "share")
.unwrap();
db.sync_folder_set_pending_mode(id, Some("ondemand"))
.unwrap();
db.sync_folder_set_pending_mode(id, Some("mirror")).unwrap();
assert!(
!db.sync_folder_clear_pending_mode_if(id, "ondemand")
.unwrap()
);
assert_eq!(
db.sync_folder_get(id)
.unwrap()
.unwrap()
.pending_mode
.as_deref(),
Some("mirror")
);
assert!(db.sync_folder_clear_pending_mode_if(id, "mirror").unwrap());
assert_eq!(db.sync_folder_get(id).unwrap().unwrap().pending_mode, None);
}
#[test]
fn mirror_uid_lookup_covers_the_root_and_synced_descendants_only() {
let db = Db::open_in_memory().unwrap();
+1
View File
@@ -10,6 +10,7 @@ pub mod control;
pub mod db;
pub mod error;
pub mod localindex;
pub mod mounts;
pub mod profile;
pub mod search;
pub mod service;
+146
View File
@@ -0,0 +1,146 @@
//! Unified presentation model for every local Proton Drive location.
//!
//! Device sync state remains owned by `sync_folder`; these types are the stable
//! shape exposed to front-ends after that state is joined with the `mount`
//! presentation table.
use std::fmt;
use serde::{Deserialize, Serialize};
use crate::control::SyncProgress;
/// How a location makes remote content available locally.
#[derive(Serialize, Deserialize, Debug, Clone, Copy, PartialEq, Eq, Default)]
#[serde(rename_all = "lowercase")]
pub enum MountMode {
/// A full local copy reconciled in both directions.
Mirror,
/// A FUSE session that fetches content as it is accessed.
OnDemand,
/// A mode written by a newer client.
#[default]
#[serde(other)]
Unknown,
}
impl MountMode {
pub const fn as_str(self) -> &'static str {
match self {
Self::Mirror => "mirror",
Self::OnDemand => "ondemand",
Self::Unknown => "unknown",
}
}
}
impl From<&str> for MountMode {
fn from(value: &str) -> Self {
match value {
"mirror" => Self::Mirror,
"ondemand" => Self::OnDemand,
_ => Self::Unknown,
}
}
}
impl fmt::Display for MountMode {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.write_str(self.as_str())
}
}
/// Whether the location itself was declared writable.
#[derive(Serialize, Deserialize, Debug, Clone, Copy, PartialEq, Eq)]
#[serde(rename_all = "lowercase")]
pub enum MountAccess {
Rw,
Ro,
}
impl MountAccess {
pub const fn as_str(self) -> &'static str {
match self {
Self::Rw => "rw",
Self::Ro => "ro",
}
}
}
impl fmt::Display for MountAccess {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.write_str(self.as_str())
}
}
/// What remote root a local location presents.
#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq)]
#[serde(tag = "kind", rename_all = "lowercase")]
pub enum MountKind {
/// The account's primary My Files mount.
MyFiles,
/// A folder owned by this machine's Proton Drive device.
Device { sync_folder_id: i64 },
/// A future standalone mount rooted at a node shared with this account.
Shared { share_root_uid: String },
}
/// One local Proton Drive location as exposed over the control protocol.
#[derive(Serialize, Deserialize, Debug, Clone)]
pub struct MountSpec {
/// Presentation-row id. Device actions use the `sync_folder_id` carried by
/// [`MountKind::Device`], not this id.
pub id: i64,
pub kind: MountKind,
pub local_path: String,
pub root_uid: String,
pub root_share_id: String,
pub mode: MountMode,
pub access: MountAccess,
/// `idle` | `syncing` | `error` | `conflict` for device locations.
pub state: String,
/// Last successful device sync, epoch seconds; `0` when not applicable.
pub last_sync: i64,
/// A device mode switch accepted but not yet applied.
#[serde(default)]
pub pending_mode: Option<MountMode>,
/// Whether a FUSE session currently owns `local_path`.
#[serde(default)]
pub mounted: bool,
/// Live device sync progress, absent when no pass is running.
#[serde(default)]
pub progress: Option<SyncProgress>,
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn mount_mode_keeps_legacy_wire_strings() {
assert_eq!(
serde_json::to_string(&MountMode::Mirror).unwrap(),
r#""mirror""#
);
assert_eq!(
serde_json::to_string(&MountMode::OnDemand).unwrap(),
r#""ondemand""#
);
assert_eq!(
serde_json::from_str::<MountMode>(r#""mirror""#).unwrap(),
MountMode::Mirror
);
assert_eq!(
serde_json::from_str::<MountMode>(r#""ondemand""#).unwrap(),
MountMode::OnDemand
);
}
#[test]
fn mount_mode_tolerates_unknown_values() {
assert_eq!(
serde_json::from_str::<MountMode>(r#""from-the-future""#).unwrap(),
MountMode::Unknown
);
}
}
+52 -3
View File
@@ -17,6 +17,8 @@
use serde::{Deserialize, Serialize};
use crate::mounts::MountMode;
/// File name of the profile inside the device root folder.
pub const PROFILE_FILE_NAME: &str = "profile.json";
@@ -39,8 +41,9 @@ pub struct ProfileFolder {
/// Absolute local path it was synced to. A *suggestion* on restore: another
/// machine may have a different username, home, or mount layout.
pub local_path: String,
/// `mirror` or `ondemand`.
pub mode: String,
/// `mirror` or `ondemand`. Unknown values written by a newer client do not
/// invalidate the rest of the profile.
pub mode: MountMode,
}
/// One pinned node (P5 pins), as `pins` stores it.
@@ -114,7 +117,7 @@ mod tests {
folders: vec![ProfileFolder {
remote_uid: "vol~link".to_string(),
local_path: "/home/nils/Documents".to_string(),
mode: "mirror".to_string(),
mode: MountMode::Mirror,
}],
pins: vec![ProfilePin {
uid: "vol~pin".to_string(),
@@ -133,6 +136,52 @@ mod tests {
assert_eq!(Profile::parse(&p.to_bytes().unwrap()).unwrap(), p);
}
#[test]
fn profile_keeps_legacy_mode_bytes_and_parses_old_documents() {
let bytes = sample().to_bytes().unwrap();
let json = String::from_utf8(bytes).unwrap();
assert!(json.contains(r#""mode": "mirror""#));
let old = br#"{
"version": 1,
"device_uid": "dev-1",
"hostname": "old-client",
"saved_at": 1,
"folders": [
{
"remote_uid": "vol~mirror",
"local_path": "/home/me/Mirror",
"mode": "mirror"
},
{
"remote_uid": "vol~ondemand",
"local_path": "/home/me/Cloud",
"mode": "ondemand"
}
]
}"#;
let profile = Profile::parse(old).unwrap();
assert_eq!(profile.folders[0].mode, MountMode::Mirror);
assert_eq!(profile.folders[1].mode, MountMode::OnDemand);
}
#[test]
fn unknown_folder_mode_does_not_reject_profile() {
let json = br#"{
"version": 1,
"device_uid": "dev-1",
"hostname": "future-client",
"saved_at": 1,
"folders": [{
"remote_uid": "vol~future",
"local_path": "/home/me/Future",
"mode": "streaming"
}]
}"#;
let profile = Profile::parse(json).unwrap();
assert_eq!(profile.folders[0].mode, MountMode::Unknown);
}
/// A field added by a later client must not break this one's restore: the
/// common subset is still usable, which is the whole reason `version` is
/// reserved for breaking changes only.
+4
View File
@@ -561,6 +561,10 @@ fn handle_control_conn(core: &Core, username: &str, mountpoint: &Path, stream: U
Ok(items) => CtlResponse::Devices { items },
Err(e) => CtlResponse::error(e),
},
Ok(CtlRequest::ListLocations) => match core.list_locations() {
Ok(items) => CtlResponse::Locations { items },
Err(e) => CtlResponse::error(e),
},
Ok(CtlRequest::RenameDevice { uid, name }) => match core.rename_device(&uid, &name) {
Ok(()) => CtlResponse::Ok {
message: format!("renamed device to {name}"),
+420 -122
View File
@@ -11,12 +11,13 @@ use std::collections::HashMap;
use std::path::{Path, PathBuf};
use std::sync::Arc;
use fuser::{Config, MountOption, Session};
use parking_lot::Mutex;
use pdfs_core::control::{
ActivityKind, DeviceInfo, JobItem, SyncFolderInfo, SyncPhase, SyncProgress,
ActivityKind, DeviceInfo, JobItem, MountKind, MountSpec, SyncFolderInfo, SyncPhase,
SyncProgress,
};
use pdfs_core::db::{StoredDevice, StoredSyncFolder};
use pdfs_core::mounts::MountMode;
use pdfs_core::{CoreError, CoreResult};
use proton_drive_rs::proton_sdk::ids::{DeviceUid, NodeUid};
use proton_drive_rs::{DeviceType, Node};
@@ -25,11 +26,69 @@ use tracing::{info, warn};
use super::sync::{self, base_name};
use super::{
BackgroundSession, Core, ProtonFs, State, SwitchBlocked, clear_stale_mount, device_type_str,
dir_is_empty, evict_dir_contents, fuse_connection_id, is_stale_mount, now_secs, parse_uid,
sync_folder_info, this_hostname, umount_session_unblocked,
Core, SecondaryInsertRejection, SecondaryMount, State, 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
@@ -57,6 +116,22 @@ pub(crate) fn set_adopted_device_uid(uid: Option<&str>) -> CoreResult<()> {
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();
}
}
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.
@@ -439,6 +514,8 @@ impl Core {
/// 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)
@@ -449,19 +526,16 @@ impl Core {
// folder the daemon no longer tracks, and nothing would ever unmount it.
// Tear it down first — including before trashing the remote tree it
// serves, which would otherwise leave it answering for deleted nodes.
if let Some((session, conn)) = self.mounts.lock().remove(&id) {
if let Err(e) = umount_session_unblocked(session, conn) {
warn!(id, error = %e, "unmount on-demand folder failed");
} else {
info!(id, path = %folder.local_path, "unmounted on-demand folder");
}
}
if delete_remote
&& let Some(uid) = parse_uid(&folder.remote_uid)
&& let Err(e) = self.rt.block_on(self.client.trash_nodes(&[uid]))
if let Some(mount) = self.mounts.lock().remove(&id)
&& let Err(error) = commit_after_teardown(mount.teardown(), || ())
{
warn!(id, error = %e, "delete remote device folder failed");
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)
@@ -469,8 +543,15 @@ impl Core {
{
return Err(CoreError::invalid(format!("no synced folder with id {id}")));
}
drop(_guard);
// 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.client.trash_nodes(&[uid]))
{
warn!(id, error = %e, "delete remote device folder failed");
}
Ok(())
}
@@ -507,13 +588,9 @@ impl Core {
// session is the only one that knows these inodes, so it must be the one
// notified about them. Filled in by `spawn_ondemand_mount`.
fork.notifier = Arc::new(OnceLock::new());
fork.session_live = Arc::new(std::sync::atomic::AtomicBool::new(false));
// Size upgrades are keyed by inode, which is per-fork too.
fork.size_upgrades = Arc::new(Mutex::new(HashMap::new()));
// The daemon's single drain thread belongs to the primary mount, but it
// drains ops for nodes that live *here*. Publishing this inode space is
// what lets it find them; without it a queued create's node keeps its
// `local~` uid after the upload lands and reads it as empty (B74).
fork.register_state();
fork
}
@@ -529,49 +606,84 @@ impl Core {
/// 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> {
if mode != "mirror" && mode != "ondemand" {
return Err(CoreError::invalid(format!(
"unknown mode {mode:?} (want mirror|ondemand)"
)));
}
// 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}")))?;
if folder.mode == mode {
// Already there. A queued switch the other way is the user changing
// their mind back before it landed, so withdraw it.
if folder.pending_mode.is_some() {
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(format!("staying {mode}"));
let current_mode = ApplicableMountMode::parse(&folder.mode)
.map_err(|error| CoreError::invalid(format!("unsupported current mode: {error}")))?;
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(&current.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}"));
}
}
return Ok(format!("already {mode}"));
}
let mode = target_mode;
let mode_str = target_mode.as_str();
match self.apply_sync_folder_mode(id, mode) {
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))
.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 {
"ondemand" => format!(
ApplicableMountMode::OnDemand => format!(
"{} will go on-demand once its local changes are uploaded",
base_name(&folder.local_path)
),
_ => format!(
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)),
}
}
@@ -585,27 +697,45 @@ impl Core {
let Ok(Some(folder)) = self.db.sync_folder_get(id) else {
return;
};
let Some(mode) = folder.pending_mode.as_deref() else {
let Some(expected_mode) = folder.pending_mode.as_deref() else {
return;
};
if folder.mode == mode {
let _ = self.db.sync_folder_set_pending_mode(id, None);
return;
}
match self.apply_sync_folder_mode(id, mode) {
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, "applied queued mode switch");
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, "queued mode switch still waiting");
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, error = %e, "queued mode switch failed; withdrawing");
let _ = self.db.sync_folder_set_pending_mode(id, None);
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)),
@@ -631,8 +761,10 @@ impl Core {
pub(crate) fn apply_sync_folder_mode(
&self,
id: i64,
mode: &str,
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`
@@ -650,13 +782,18 @@ impl Core {
.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 folder.mode == mode {
return Ok(format!("already {mode}"));
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 {
"ondemand" => {
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
@@ -703,25 +840,28 @@ impl Core {
if let Err(e) = evict_dir_contents(&local) {
warn!(id, path = %local.display(), error = %e, "evict local dir contents failed");
}
let session = self.spawn_ondemand_mount(&local, root).map_err(|e| {
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.mounts
.lock()
.insert(id, (session, fuse_connection_id(&local)));
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.
if let Some((session, conn)) = self.mounts.lock().remove(&id)
&& let Err(e) = umount_session_unblocked(session, conn)
if let Some(mount) = self.mounts.lock().remove(&id)
&& let Err(error) = commit_after_teardown(mount.teardown(), || ())
{
warn!(id, error = %e, "unmount on-demand folder failed");
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.
@@ -749,23 +889,52 @@ impl Core {
&self,
local: &Path,
root: Node,
) -> CoreResult<BackgroundSession> {
clear_stale_mount(local);
let mut config = Config::default();
config.mount_options = vec![
MountOption::FSName("protondrive".to_string()),
MountOption::Subtype("protondrive".to_string()),
MountOption::DefaultPermissions,
];
) -> CoreResult<SecondaryMount> {
let core = self.fork_state();
let fs = ProtonFs::new(core.clone(), root);
let session = Session::new(fs, local, &config)
.and_then(|s| s.spawn())
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())))?;
// Hand this fork its own session's notification channel, so a background
// size upgrade can drop the kernel's cached attrs for these inodes.
let _ = core.notifier.set(session.notifier());
Ok(session)
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, &current))
{
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
@@ -786,14 +955,57 @@ impl Core {
if folder.mode != "ondemand" {
continue;
}
let local = PathBuf::from(&folder.local_path);
// A daemon killed before it could unmount (systemd's stop timeout,
// say) leaves this path attached to a dead FUSE connection: it still
// exists, but every stat on it fails with ENOTCONN — so `is_dir()`
// reads it as *missing* and the folder gets parked in `error` on
// every subsequent start, with no way back but a manual
// `fusermount -u`. Detach the corpse first; the directory
// underneath then restores normally.
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.client.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, &current) => 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(&current.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);
@@ -805,11 +1017,6 @@ impl Core {
.sync_folder_set_state(folder.id, "error", now_secs());
continue;
}
// An `ondemand` folder's local dir is empty by construction — the switch
// evicts it. Finding files there means the row is lying (a switch that
// died between persisting the mode and evicting), and mounting over them
// would hide real local data behind the remote tree. Leave the files
// alone and let the user resolve it.
match dir_is_empty(&local) {
Ok(true) => {}
Ok(false) => {
@@ -831,40 +1038,16 @@ impl Core {
continue;
}
}
let Some(root_uid) = parse_uid(&folder.remote_uid) else {
warn!(id = folder.id, "restore on-demand: bad remote uid");
continue;
};
let root = match self
.rt
.block_on(self.client.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");
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: fetch remote failed");
let _ = self
.db
.sync_folder_set_state(folder.id, "error", now_secs());
continue;
}
};
match self.spawn_ondemand_mount(&local, root) {
Ok(session) => {
self.mounts
.lock()
.insert(folder.id, (session, fuse_connection_id(&local)));
let _ = self.db.sync_folder_set_state(folder.id, "idle", now_secs());
info!(id = folder.id, path = %local.display(), "remounted on-demand folder");
}
Err(e) => {
warn!(id = folder.id, error = %e, "restore on-demand: mount failed");
let _ = self
@@ -875,3 +1058,118 @@ impl Core {
}
}
}
#[cfg(test)]
mod tests {
use pdfs_core::db::StoredSyncFolder;
use super::{
ApplicableMountMode, commit_after_teardown, pending_intent_matches,
restore_snapshot_matches, select_mode_transition,
};
#[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));
}
}
+241 -24
View File
@@ -98,7 +98,9 @@ mod workers;
use background::{run_event_sync, run_local_index};
pub(crate) use mount::is_stale_mount;
pub use mount::{MountOptions, MountOutcome, mount};
use mount::{SecondaryMount, clear_stale_mount, fuse_connection_id, umount_session_unblocked};
use mount::{
SecondaryInsertRejection, SecondaryMount, clear_stale_mount, fuse_connection_id, spawn_session,
};
use reads::{ReaderSlot, STREAM_BYPASS_MIN, StreamRing};
use state::{Entry, Intervals, PendingRevision, State, WriteHandle};
use tracing::{debug, error, info, warn};
@@ -400,6 +402,10 @@ struct Core {
/// would name inodes that session has never heard of. [`Core::fork_state`]
/// gives each fork an empty cell of its own.
notifier: Arc<OnceLock<Notifier>>,
/// True only after this inode space's FUSE session spawned successfully.
/// Registration happens earlier for the primary drain, so state residency
/// alone cannot answer whether the location is mounted.
session_live: Arc<AtomicBool>,
/// In-flight upload/download progress, served to `GetQueueStatus`. Shared
/// across the FUSE session and the control-socket task.
transfers: Arc<TransferRegistry>,
@@ -417,10 +423,10 @@ struct Core {
/// id (devices.md Phase 3). Each is a `ProtonFs` rooted at the folder's remote
/// node, mounted over its local path, sharing this Core's client/cache/db but
/// with its own inode space (`fork_state`). Held so we can unmount on toggle
/// back to `mirror` and on daemon shutdown. The `u32` is the FUSE connection
/// id (see [`fuse_connection_id`]), captured at mount time so teardown can
/// abort a mid-transfer connection instead of blocking on it.
mounts: Arc<Mutex<HashMap<i64, SecondaryMount>>>,
/// back to `mirror` and on daemon shutdown. Each entry retains the FUSE
/// connection id and the fork's exact liveness flag so teardown first makes
/// that inode space unroutable.
mounts: Arc<Mutex<mount::SecondaryMountRegistry<SecondaryMount>>>,
/// Per-sync-folder locks, held for a whole reconcile pass and for a whole
/// mode switch. A `mirror→ondemand` flip evicts the local tree and mounts
/// FUSE over it, so it must never overlap a pass that is walking and
@@ -523,15 +529,23 @@ fn take_self_change(
/// the kernel to drop inode 42 is only correct on the session that minted it.
/// Background work that invalidates as well as mutates needs the matching pair,
/// which is what [`Core::for_each_mount`] hands it.
type LiveMount = (Arc<Mutex<State>>, Arc<OnceLock<Notifier>>);
type LiveMount = (
PathBuf,
Arc<Mutex<State>>,
Arc<OnceLock<Notifier>>,
Arc<AtomicBool>,
);
struct MountedState {
/// Absolute local root of this inode space.
mountpoint: PathBuf,
/// `Weak`, so an unmounted fork's state is dropped rather than pinned here;
/// a dead entry is reaped on the next registry walk.
state: std::sync::Weak<Mutex<State>>,
/// This mount's kernel notification channel — the same cell as
/// [`Core::notifier`], still empty until its session is spawned.
notifier: Arc<OnceLock<Notifier>>,
session_live: Arc<AtomicBool>,
}
/// Every mounted inode space in the daemon, shared by the primary `Core` and
@@ -546,14 +560,28 @@ struct StateRegistry(Mutex<Vec<MountedState>>);
impl StateRegistry {
/// Publish an inode space and the channel to the session that owns it.
///
/// Registering the same `state` twice would make every walk visit it twice;
/// callers register exactly once, at mount time.
fn register(&self, state: &Arc<Mutex<State>>, notifier: Arc<OnceLock<Notifier>>) {
/// A state/path pair is registered exactly once by its owner. Session
/// construction only flips `session_live`; it never republishes the state.
fn register(
&self,
mountpoint: &Path,
state: &Arc<Mutex<State>>,
notifier: Arc<OnceLock<Notifier>>,
session_live: Arc<AtomicBool>,
) {
let mut states = self.0.lock();
states.retain(|m| m.state.strong_count() > 0);
debug_assert!(
!states
.iter()
.any(|mounted| mounted.state.ptr_eq(&Arc::downgrade(state))),
"an inode state must be registered exactly once"
);
states.push(MountedState {
mountpoint: mountpoint.to_path_buf(),
state: Arc::downgrade(state),
notifier,
session_live,
});
}
@@ -565,10 +593,39 @@ impl StateRegistry {
states.retain(|m| m.state.strong_count() > 0);
states
.iter()
.filter_map(|m| Some((m.state.upgrade()?, m.notifier.clone())))
.filter_map(|m| {
Some((
m.mountpoint.clone(),
m.state.upgrade()?,
m.notifier.clone(),
m.session_live.clone(),
))
})
.collect()
}
/// The live session whose local root most specifically covers `path`.
///
/// On-demand roots can be nested below another location. Selecting by the
/// longest component prefix ensures the nested session wins rather than the
/// broader primary mount. Callers currently need the selected mount only;
/// a future path router can derive a relative suffix with
/// `path.strip_prefix(mountpoint)` without storing a second path value here.
fn covering(&self, path: &Path) -> Option<LiveMount> {
self.live()
.into_iter()
.filter(|(mountpoint, _, _, live)| {
live.load(Ordering::Acquire) && path.starts_with(mountpoint)
})
.max_by_key(|(mountpoint, _, _, _)| mountpoint.components().count())
}
fn is_mounted_at(&self, path: &Path) -> bool {
self.covering(path).is_some_and(|(mountpoint, _, _, live)| {
mountpoint == path && live.load(Ordering::Acquire)
})
}
/// Whether any live inode space owns and currently exposes `uid`.
///
/// Retained open inodes and revoked incoming shares remain interned after
@@ -578,7 +635,8 @@ impl StateRegistry {
fn owns_visible_uid(&self, uid: &NodeUid) -> bool {
self.live()
.into_iter()
.any(|(state, _)| state.lock().owns_visible_uid(uid))
.filter(|(_, _, _, live)| live.load(Ordering::Acquire))
.any(|(_, state, _, _)| state.lock().owns_visible_uid(uid))
}
}
@@ -589,6 +647,9 @@ impl StateRegistry {
enum SwitchBlocked {
/// The folder is mid-pass, or not yet safe to switch. Try again after a pass.
NotNow,
/// A queued target was canceled or replaced before the folder lock became
/// available. The newer intent owns the next transition.
Superseded,
/// The switch was attempted and broke.
Failed(String),
}
@@ -596,8 +657,13 @@ enum SwitchBlocked {
impl Core {
/// Register an inode space so background work can reach it. Called once for
/// the primary mount and once per on-demand fork.
fn register_state(&self) {
self.states.register(&self.state, self.notifier.clone());
fn register_state(&self, mountpoint: &Path) {
self.states.register(
mountpoint,
&self.state,
self.notifier.clone(),
self.session_live.clone(),
);
}
/// Run `apply` against every live inode space — this daemon's primary mount
@@ -610,7 +676,7 @@ impl Core {
/// to all of them is correct because a uid is unique across mounts — at most
/// one state has an entry for it, and the rest are no-ops.
fn for_each_state(&self, mut apply: impl FnMut(&mut State)) {
for (state, notifier) in self.states.live() {
for (_, state, notifier, _) in self.states.live() {
let changed = {
let mut state = state.lock();
apply(&mut state);
@@ -627,7 +693,7 @@ impl Core {
/// `None` for a mount whose session has not been spawned yet — there is
/// nothing to invalidate in a kernel that has never been told about it.
fn for_each_mount(&self, mut apply: impl FnMut(&mut State, Option<&Notifier>)) {
for (state, notifier) in self.states.live() {
for (_, state, notifier, _) in self.states.live() {
apply(&mut state.lock(), notifier.get());
}
}
@@ -660,7 +726,7 @@ impl Core {
live_access.push(access);
}
}
for (state, _) in self.states.live() {
for (_, state, _, _) in self.states.live() {
if Arc::ptr_eq(&state, &self.state) {
continue;
}
@@ -1152,6 +1218,7 @@ impl Core {
// exactly this.
self.wake_drain();
info!("back online");
self.repair_primary_share_id(&root.uid);
return;
}
Err(e) => {
@@ -4627,6 +4694,10 @@ mod tests {
use pdfs_core::cache::{Baseline, StagedWrite};
use pdfs_core::db::{OP_REVISION, OP_TRASH, PendingOp};
fn session_flag(live: bool) -> std::sync::Arc<std::sync::atomic::AtomicBool> {
std::sync::Arc::new(std::sync::atomic::AtomicBool::new(live))
}
/// The registry every mount publishes itself into is what lets the daemon's
/// single drain thread reach a node living in an on-demand fork's inode
/// space. A fork missing from it reads as an empty file for the life of the
@@ -4644,8 +4715,17 @@ mod tests {
let primary = std::sync::Arc::new(parking_lot::Mutex::new(primary));
let fork_a = std::sync::Arc::new(parking_lot::Mutex::new(fork_a));
let fork_b = std::sync::Arc::new(parking_lot::Mutex::new(fork_b));
for state in [&primary, &fork_a, &fork_b] {
registry.register(state, std::sync::Arc::new(std::sync::OnceLock::new()));
for (mountpoint, state) in [
("/mnt/primary", &primary),
("/mnt/fork-a", &fork_a),
("/mnt/fork-b", &fork_b),
] {
registry.register(
std::path::Path::new(mountpoint),
state,
std::sync::Arc::new(std::sync::OnceLock::new()),
session_flag(false),
);
}
assert_eq!(registry.live().len(), 3, "every registered mount is walked");
@@ -4674,6 +4754,113 @@ mod tests {
);
}
#[test]
fn state_registry_covering_uses_longest_prefix_and_ignores_dead_forks() {
let registry = StateRegistry::default();
let (primary, _primary_dir) = rooted_state("primary-volume", "primary");
let (nested, _nested_dir) = rooted_state("nested-volume", "nested");
let (sibling, _sibling_dir) = rooted_state("sibling-volume", "sibling");
let primary_uid = primary.entries[&super::ROOT_INO].uid.clone();
let primary = std::sync::Arc::new(parking_lot::Mutex::new(primary));
let nested = std::sync::Arc::new(parking_lot::Mutex::new(nested));
let sibling = std::sync::Arc::new(parking_lot::Mutex::new(sibling));
let primary_path = std::path::Path::new("/home/me/ProtonDrive");
let nested_path = std::path::Path::new("/home/me/ProtonDrive/Device");
let sibling_path = std::path::Path::new("/home/me/Archive");
let primary_live = session_flag(false);
let nested_live = session_flag(false);
let sibling_live = session_flag(true);
registry.register(
primary_path,
&primary,
std::sync::Arc::new(std::sync::OnceLock::new()),
primary_live.clone(),
);
registry.register(
nested_path,
&nested,
std::sync::Arc::new(std::sync::OnceLock::new()),
nested_live.clone(),
);
registry.register(
sibling_path,
&sibling,
std::sync::Arc::new(std::sync::OnceLock::new()),
sibling_live,
);
assert_eq!(registry.live().len(), 3);
assert!(!registry.is_mounted_at(primary_path));
assert!(!registry.is_mounted_at(nested_path));
assert!(
!registry.owns_visible_uid(&primary_uid),
"an unspawned session must not authorize resident uids"
);
assert!(
registry
.covering(&nested_path.join("folder/file.txt"))
.is_none(),
"a registered but unspawned session must not win path covering"
);
primary_live.store(true, std::sync::atomic::Ordering::Release);
nested_live.store(true, std::sync::atomic::Ordering::Release);
assert!(registry.owns_visible_uid(&primary_uid));
let (covering, covering_state, _, _) = registry
.covering(&nested_path.join("folder/file.txt"))
.unwrap();
assert_eq!(covering, nested_path);
assert!(std::sync::Arc::ptr_eq(&covering_state, &nested));
drop(covering_state);
let (covering, _, _, _) = registry.covering(nested_path).unwrap();
assert_eq!(covering, nested_path, "an exact mountpoint covers itself");
let (covering, _, _, _) = registry
.covering(std::path::Path::new("/home/me/ProtonDrive/Device/folder/"))
.unwrap();
assert_eq!(
covering, nested_path,
"a trailing separator does not change component-prefix selection"
);
let (covering, _, _, _) = registry
.covering(std::path::Path::new(
"/home/me/ProtonDrive/DeviceBackup/file",
))
.unwrap();
assert_eq!(
covering, primary_path,
"/Device must not string-prefix-match /DeviceBackup"
);
let (covering, covering_state, _, _) =
registry.covering(&sibling_path.join("file")).unwrap();
assert_eq!(covering, sibling_path);
assert!(std::sync::Arc::ptr_eq(&covering_state, &sibling));
assert!(
registry
.covering(std::path::Path::new("/home/me/ProtonDriveBackup/file"))
.is_none(),
"/ProtonDrive must not string-prefix-match /ProtonDriveBackup"
);
assert!(
registry
.covering(std::path::Path::new("/tmp/outside"))
.is_none(),
"an outside path has no covering mount"
);
assert!(registry.is_mounted_at(primary_path));
assert!(registry.is_mounted_at(nested_path));
assert!(
!registry.is_mounted_at(&nested_path.join("folder")),
"a path covered by a session is not itself a mountpoint"
);
// This is also the failed-fork case: the caller registered the fork,
// session construction failed, and its last strong `Core`/state handle
// was dropped. It must not make ListLocations report mounted.
drop(nested);
assert!(!registry.is_mounted_at(nested_path));
assert_eq!(registry.live().len(), 2);
}
#[test]
fn state_registry_resolves_uids_from_any_resident_mount() {
let registry = StateRegistry::default();
@@ -4687,8 +4874,18 @@ mod tests {
fork.children.insert(super::ROOT_INO, vec![fork_ino]);
let primary = std::sync::Arc::new(parking_lot::Mutex::new(primary));
let fork = std::sync::Arc::new(parking_lot::Mutex::new(fork));
registry.register(&primary, std::sync::Arc::new(std::sync::OnceLock::new()));
registry.register(&fork, std::sync::Arc::new(std::sync::OnceLock::new()));
registry.register(
std::path::Path::new("/mnt/primary"),
&primary,
std::sync::Arc::new(std::sync::OnceLock::new()),
session_flag(true),
);
registry.register(
std::path::Path::new("/mnt/device"),
&fork,
std::sync::Arc::new(std::sync::OnceLock::new()),
session_flag(true),
);
assert!(registry.owns_visible_uid(&primary_uid));
assert!(registry.owns_visible_uid(&fork_uid));
@@ -4711,7 +4908,12 @@ mod tests {
let child_ino = state.intern(super::ROOT_INO, child_node);
state.children.insert(super::ROOT_INO, vec![child_ino]);
let state = std::sync::Arc::new(parking_lot::Mutex::new(state));
registry.register(&state, std::sync::Arc::new(std::sync::OnceLock::new()));
registry.register(
std::path::Path::new("/mnt/device"),
&state,
std::sync::Arc::new(std::sync::OnceLock::new()),
session_flag(true),
);
assert!(registry.owns_visible_uid(&child_uid));
@@ -4769,7 +4971,12 @@ mod tests {
);
let state = std::sync::Arc::new(parking_lot::Mutex::new(state));
registry.register(&state, std::sync::Arc::new(std::sync::OnceLock::new()));
registry.register(
std::path::Path::new("/mnt/device"),
&state,
std::sync::Arc::new(std::sync::OnceLock::new()),
session_flag(true),
);
assert!(
registry.owns_visible_uid(&child_uid),
"the absent listing remains unknown until refresh, so a valid resident uid is allowed"
@@ -5015,7 +5222,12 @@ mod tests {
state.children.insert(revoked_ino, vec![child_ino]);
let state = std::sync::Arc::new(parking_lot::Mutex::new(state));
registry.register(&state, std::sync::Arc::new(std::sync::OnceLock::new()));
registry.register(
std::path::Path::new("/mnt/device"),
&state,
std::sync::Arc::new(std::sync::OnceLock::new()),
session_flag(true),
);
assert!(registry.owns_visible_uid(&open_uid));
assert!(registry.owns_visible_uid(&child_uid));
@@ -5064,7 +5276,12 @@ mod tests {
state.children.insert(cycle_b, vec![cycle_a]);
let state = std::sync::Arc::new(parking_lot::Mutex::new(state));
registry.register(&state, std::sync::Arc::new(std::sync::OnceLock::new()));
registry.register(
std::path::Path::new("/mnt/device"),
&state,
std::sync::Arc::new(std::sync::OnceLock::new()),
session_flag(true),
);
assert!(
!registry.owns_visible_uid(&foreign_uid),
+279 -56
View File
@@ -28,10 +28,101 @@ pub(crate) fn is_stale_mount(path: &Path) -> bool {
)
}
/// A secondary (on-demand sync folder) FUSE session, paired with its FUSE
/// connection id so teardown can abort a mid-transfer connection rather than
/// block on it. See [`Core::mounts`] and [`umount_session_unblocked`].
pub(super) type SecondaryMount = (BackgroundSession, Option<u32>);
/// A secondary (on-demand sync folder) FUSE session and the exact liveness flag
/// published with its forked inode state.
///
/// Keeping these together makes every teardown mark the fork unroutable before
/// unmounting, including paths that can block on an in-flight FUSE request.
pub(super) struct SecondaryMount {
session: BackgroundSession,
conn: Option<u32>,
session_live: Arc<AtomicBool>,
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub(super) enum SecondaryInsertRejection {
Closed,
Duplicate,
}
/// Secondary sessions and the daemon-wide gate controlling whether newly
/// spawned sessions may become visible. The gate and map share one lock so
/// shutdown cannot race an insertion between checking and publishing.
pub(super) struct SecondaryMountRegistry<T> {
accepting: bool,
entries: HashMap<i64, T>,
}
impl<T> Default for SecondaryMountRegistry<T> {
fn default() -> Self {
Self {
accepting: false,
entries: HashMap::new(),
}
}
}
impl<T> SecondaryMountRegistry<T> {
pub(super) fn open(&mut self) {
self.accepting = true;
}
pub(super) fn is_accepting(&self) -> bool {
self.accepting
}
pub(super) fn close(&mut self) {
self.accepting = false;
}
pub(super) fn insert(
&mut self,
id: i64,
mount: T,
) -> Result<(), (SecondaryInsertRejection, T)> {
if !self.accepting {
return Err((SecondaryInsertRejection::Closed, mount));
}
if self.entries.contains_key(&id) {
return Err((SecondaryInsertRejection::Duplicate, mount));
}
self.entries.insert(id, mount);
Ok(())
}
pub(super) fn remove(&mut self, id: &i64) -> Option<T> {
self.entries.remove(id)
}
pub(super) fn contains_key(&self, id: &i64) -> bool {
self.entries.contains_key(id)
}
pub(super) fn close_and_drain(&mut self) -> Vec<(i64, T)> {
self.close();
self.entries.drain().collect()
}
}
impl SecondaryMount {
pub(super) fn new(
session: BackgroundSession,
conn: Option<u32>,
session_live: Arc<AtomicBool>,
) -> Self {
Self {
session,
conn,
session_live,
}
}
pub(super) fn teardown(self) -> std::io::Result<()> {
teardown_session(&self.session_live, || {
umount_session_unblocked(self.session, self.conn)
})
}
}
/// The kernel's id for the FUSE connection backing the mount at `mountpoint` —
/// the directory name under `/sys/fs/fuse/connections`, which is the minor
@@ -152,6 +243,71 @@ pub struct MountOptions {
pub sweep_mode: SweepMode,
}
/// Spawn one FUSE session rooted at an arbitrary remote node.
///
/// Primary My Files and secondary on-demand device locations use this same
/// construction path so mount options, stale-mount recovery, registry
/// liveness, and notifier ownership cannot drift apart. Each caller owns its
/// state/path registration and performs it exactly once before calling here.
pub(super) fn spawn_session(
core: &Core,
mountpoint: &Path,
root: Node,
) -> std::io::Result<BackgroundSession> {
let mut config = Config::default();
config.mount_options = vec![
MountOption::FSName("protondrive".to_string()),
MountOption::Subtype("protondrive".to_string()),
MountOption::DefaultPermissions,
];
clear_stale_mount(mountpoint);
info!(mountpoint = %mountpoint.display(), "mounting Proton Drive location");
let fs = ProtonFs::new(core.clone(), root);
if core.primary {
// Only My Files rehydrates the daemon-wide node cache. A secondary
// inode space is rooted at one device subtree and must not import every
// unrelated persisted node.
core.hydrate();
}
let session = Session::new(fs, mountpoint, &config)?.spawn()?;
let _ = core.notifier.set(session.notifier());
core.session_live.store(true, Ordering::Release);
Ok(session)
}
/// Make a session unroutable before beginning any teardown that may block.
fn teardown_session<T>(session_live: &AtomicBool, teardown: impl FnOnce() -> T) -> T {
session_live.store(false, Ordering::Release);
teardown()
}
impl Core {
/// Resolve the primary share id away from startup and publish it only if the
/// projected root has not changed while the request was in flight.
pub(crate) fn repair_primary_share_id(&self, root_uid: &NodeUid) {
match self.rt.block_on(self.client.context_share_id(root_uid)) {
Ok(share_id) => {
match self
.db
.mount_repair_my_files_share_id(&root_uid.to_string(), &share_id.to_string())
{
Ok(true) => debug!(%root_uid, "repaired My Files share id"),
Ok(false) => {
debug!(%root_uid, "discarded stale My Files share-id result");
}
Err(error) => {
warn!(%root_uid, error = %error, "persist My Files share id failed");
}
}
}
Err(error) => {
debug!(%root_uid, error = %error, "resolve My Files share id failed");
}
}
}
}
/// Mount the filesystem at `mountpoint` and block until it is unmounted or the
/// daemon is asked to stop.
///
@@ -179,6 +335,8 @@ pub fn mount(
} = options;
let (root, online) = fetch_or_recall_root(&client, &rt, &db)?;
let scope = root.tree_event_scope_id();
db.mount_upsert_my_files(&mountpoint.to_string_lossy(), &root.uid.to_string(), None)
.map_err(|error| std::io::Error::other(format!("project My Files mount: {error}")))?;
let share_access = db
.all_share_access()
.map_err(|error| std::io::Error::other(format!("load shared access: {error}")))?;
@@ -228,17 +386,18 @@ pub fn mount(
no_thumbnail: Arc::new(Mutex::new(HashMap::new())),
size_upgrades: Arc::new(Mutex::new(HashMap::new())),
notifier: Arc::new(OnceLock::new()),
session_live: Arc::new(AtomicBool::new(false)),
transfers: TransferRegistry::new(),
indexing: Arc::new(AtomicBool::new(false)),
sync_progress: Arc::new(Mutex::new(HashMap::new())),
sync_tx,
mounts: Arc::new(Mutex::new(HashMap::new())),
mounts: Arc::new(Mutex::new(SecondaryMountRegistry::default())),
sync_locks: Arc::new(Mutex::new(HashMap::new())),
states: Arc::new(StateRegistry::default()),
};
// Before anything can queue work against it: the drain thread below reaches
// every mount's inode space through this registry, not through `core.state`.
core.register_state();
core.register_state(mountpoint);
// Writes queued by a previous run (or left behind by a crash) are still owed
// an upload, and reads must be served from their staged blobs until they land.
@@ -282,16 +441,6 @@ pub fn mount(
.spawn(move || core.run_online_probe())?;
}
// Re-establish on-demand mounts left over from a previous run (devices.md
// Phase 4). On its own thread: each remount fetches a remote node, and we
// must not block the main mount below on the network.
{
let core = core.clone();
std::thread::Builder::new()
.name("pdfs-restore-ondemand".into())
.spawn(move || core.restore_ondemand_mounts())?;
}
// Keep the launcher prompt's "This computer" index fresh. Its own thread:
// the walk is I/O-heavy and must never sit in front of a FUSE callback.
{
@@ -304,18 +453,6 @@ pub fn mount(
.spawn(move || run_local_index(db, indexing, transfers, mountpoint))?;
}
let mut config = Config::default();
config.mount_options = vec![
MountOption::FSName("protondrive".to_string()),
MountOption::Subtype("protondrive".to_string()),
MountOption::DefaultPermissions,
];
// A crashed previous run can leave the kernel mount in place, which makes
// the fresh mount below fail with EBUSY. Lazily detach any leftover first.
clear_stale_mount(mountpoint);
info!(mountpoint = %mountpoint.display(), "mounting Proton Drive");
// Bind the control socket before the FUSE session takes over the thread. A
// stale socket file from a previous run would block the bind, so clear it.
let _ = std::fs::remove_file(control_socket);
@@ -332,32 +469,70 @@ pub fn mount(
"control socket permissions: {e}"
)));
}
{
let core = core.clone();
let username = username.clone();
let mountpoint = mountpoint.to_path_buf();
std::thread::Builder::new()
.name("pdfs-control".into())
.spawn(move || run_control_socket(core, username, mountpoint, listener))?;
}
let fs = ProtonFs::new(core.clone(), root);
// Warm the in-memory maps from the DB so a cold start serves previously
// seen metadata without re-hitting the API (plan.md P1).
core.hydrate();
// Build the session explicitly (not `mount2`) so we can grab a `Notifier`
// for the event task. `spawn` runs the session loop on its own background
// thread; we then wait here for either a stop signal or the mount ending.
let bg = Session::new(fs, mountpoint, &config)?.spawn()?;
// Do not accept control requests until `spawn_session` has transitioned the
// primary registration to live. The listener may queue a connection during
// this short window, but its handler cannot observe `mounted = false`.
let bg = match spawn_session(&core, mountpoint, root) {
Ok(session) => session,
Err(error) => {
let _ = std::fs::remove_file(control_socket);
return Err(error);
}
};
core.mounts.lock().open();
// The connection id, captured now while the mount is live, so a stop signal
// mid-transfer can abort it rather than block `join` (see `abort_fuse_connection`).
let main_conn = fuse_connection_id(mountpoint);
let notifier = bg.notifier();
// Same channel, kept on the `Core` so background work (a size upgrade, say)
// can invalidate kernel-cached metadata without threading a handle through.
let _ = core.notifier.set(notifier.clone());
{
let control_core = core.clone();
let username = username.clone();
let mountpoint = mountpoint.to_path_buf();
if let Err(error) = std::thread::Builder::new()
.name("pdfs-control".into())
.spawn(move || run_control_socket(control_core, username, mountpoint, listener))
{
let secondaries = core.mounts.lock().close_and_drain();
for (id, mount) in secondaries {
if let Err(unmount_error) = mount.teardown() {
warn!(id, error = %unmount_error, "secondary teardown after control startup failure failed");
}
}
if let Err(unmount_error) = teardown_session(&core.session_live, || {
umount_session_unblocked(bg, main_conn)
}) {
warn!(error = %unmount_error, "unmount after control startup failure failed");
}
let _ = std::fs::remove_file(control_socket);
return Err(error);
}
}
// Re-establish on-demand mounts only after the primary session is live.
// Keep the worker so shutdown can close publication, wait for any in-flight
// fetch to observe that closure, and then drain a stable registry.
let restore_worker = {
let restore_core = core.clone();
match std::thread::Builder::new()
.name("pdfs-restore-ondemand".into())
.spawn(move || restore_core.restore_ondemand_mounts())
{
Ok(worker) => Some(worker),
Err(error) => {
warn!(%error, "start on-demand restore worker failed");
None
}
}
};
rt.spawn(run_event_sync(client, scope, core.clone()));
if online {
let repair_core = core.clone();
let root_uid = core.primary_root_uid.clone();
if let Err(error) = std::thread::Builder::new()
.name("pdfs-share-id".into())
.spawn(move || repair_core.repair_primary_share_id(&root_uid))
{
warn!(%error, "start My Files share-id repair failed");
}
}
// Stop signals (SIGTERM from `systemctl --user stop`, SIGINT from Ctrl-C)
// are delivered onto the async runtime; bridge them onto a sync channel so
@@ -396,7 +571,10 @@ pub fn mount(
match sig_rx.recv_timeout(Duration::from_millis(500)) {
Ok(()) => {
info!("stop requested; unmounting");
if let Err(e) = umount_session_unblocked(bg, main_conn) {
core.mounts.lock().close();
if let Err(e) = teardown_session(&core.session_live, || {
umount_session_unblocked(bg, main_conn)
}) {
warn!(error = %e, "umount_and_join failed");
}
break MountOutcome::Shutdown;
@@ -404,7 +582,8 @@ pub fn mount(
Err(std::sync::mpsc::RecvTimeoutError::Timeout) => {
if bg.guard.is_finished() {
info!("mount ended externally");
if let Err(e) = bg.join() {
core.mounts.lock().close();
if let Err(e) = teardown_session(&core.session_live, || bg.join()) {
warn!(error = %e, "session join failed");
}
break MountOutcome::Unmounted;
@@ -412,7 +591,8 @@ pub fn mount(
}
Err(std::sync::mpsc::RecvTimeoutError::Disconnected) => {
// Signal task gone (failed to install); fall back to join.
let _ = bg.join();
core.mounts.lock().close();
let _ = teardown_session(&core.session_live, || bg.join());
break MountOutcome::Unmounted;
}
}
@@ -420,9 +600,14 @@ pub fn mount(
// Unmount every on-demand sync folder too, or the kernel mounts linger as
// stale and the next start fails with EBUSY (devices.md Phase 3).
let secondaries: Vec<_> = core.mounts.lock().drain().collect();
for (id, (session, conn)) in secondaries {
if let Err(e) = umount_session_unblocked(session, conn) {
if let Some(worker) = restore_worker
&& worker.join().is_err()
{
warn!("on-demand restore worker panicked");
}
let secondaries = core.mounts.lock().close_and_drain();
for (id, mount) in secondaries {
if let Err(e) = mount.teardown() {
warn!(id, error = %e, "unmount on-demand folder failed");
}
}
@@ -430,3 +615,41 @@ pub fn mount(
let _ = std::fs::remove_file(control_socket);
Ok(outcome)
}
#[cfg(test)]
mod tests {
use std::sync::atomic::{AtomicBool, Ordering};
use super::{SecondaryInsertRejection, SecondaryMountRegistry, teardown_session};
#[test]
fn session_is_unroutable_before_teardown_starts() {
let live = AtomicBool::new(true);
teardown_session(&live, || {
assert!(
!live.load(Ordering::Acquire),
"teardown must never run while the registry still reports mounted"
);
});
}
#[test]
fn secondary_registry_rejects_closed_and_duplicate_insertions() {
let mut mounts = SecondaryMountRegistry::default();
assert_eq!(
mounts.insert(7, "closed"),
Err((SecondaryInsertRejection::Closed, "closed"))
);
mounts.open();
assert_eq!(mounts.insert(7, "first"), Ok(()));
assert_eq!(
mounts.insert(7, "duplicate"),
Err((SecondaryInsertRejection::Duplicate, "duplicate"))
);
assert!(mounts.contains_key(&7));
assert_eq!(mounts.close_and_drain(), vec![(7, "first")]);
assert!(!mounts.is_accepting());
}
}
+27 -7
View File
@@ -12,6 +12,7 @@ use std::sync::OnceLock;
use std::sync::atomic::{AtomicBool, Ordering};
use pdfs_core::control::RestorableFolder;
use pdfs_core::mounts::MountMode;
use pdfs_core::profile::{PROFILE_FILE_NAME, PROFILE_VERSION, Profile, ProfileFolder, ProfilePin};
use pdfs_core::{CoreError, CoreResult};
use proton_drive_rs::proton_sdk::ids::NodeUid;
@@ -133,7 +134,7 @@ impl Core {
local_path: f.local_path,
// A queued switch is where the user asked the folder to be, so
// that is what the next machine should set up.
mode: f.pending_mode.unwrap_or(f.mode),
mode: MountMode::from(f.pending_mode.unwrap_or(f.mode).as_str()),
})
.collect();
let pins = self
@@ -267,7 +268,7 @@ impl Core {
name: n.name,
local_path: proposed.to_string_lossy().to_string(),
mode: saved
.map(|f| f.mode.clone())
.map(|f| f.mode.to_string())
.unwrap_or_else(|| "mirror".into()),
}
})
@@ -319,10 +320,10 @@ impl Core {
skipped.push(format!("{}: {e}", item.local_path));
continue;
}
let mode = match item.mode.as_str() {
"mirror" | "ondemand" => item.mode.as_str(),
other => {
skipped.push(format!("{}: unknown mode {other}", item.local_path));
let mode = match restore_mode(&item.mode) {
Some(mode) => mode,
None => {
skipped.push(format!("{}: unknown mode {}", item.local_path, item.mode));
continue;
}
};
@@ -333,7 +334,7 @@ impl Core {
// `ondemand` is applied through the normal request path, which
// mounts the FUSE session and handles the folder being busy — the
// restore does not get its own copy of that logic.
if mode == "ondemand" {
if mode == MountMode::OnDemand {
let _ = self.request_sync_folder_mode(id, "ondemand");
}
let _ = self.sync_tx.send(sync::SyncMsg::Reconcile(id));
@@ -350,6 +351,17 @@ impl Core {
}
}
/// Accept only modes this build can safely restore. `MountMode` itself is
/// tolerant so a future value does not reject the whole profile; the folder
/// carrying that value is skipped here with the message assembled above.
fn restore_mode(value: &str) -> Option<MountMode> {
match MountMode::from(value) {
MountMode::Mirror => Some(MountMode::Mirror),
MountMode::OnDemand => Some(MountMode::OnDemand),
MountMode::Unknown => None,
}
}
/// Whether a path recorded on another machine can be reused verbatim here.
///
/// Requires the *parent* to exist: `~/Documents` on a machine with the same
@@ -382,6 +394,14 @@ mod tests {
assert!(!path_is_usable_here(Path::new("relative/path")));
}
#[test]
fn unknown_profile_mode_is_skipped_without_rejecting_known_modes() {
assert_eq!(restore_mode("mirror"), Some(MountMode::Mirror));
assert_eq!(restore_mode("ondemand"), Some(MountMode::OnDemand));
assert_eq!(restore_mode("streaming"), None);
assert_eq!(restore_mode("unknown"), None);
}
/// A path whose parent exists but which does not itself is exactly the
/// restore case — the directory is about to be created.
#[test]