diff --git a/crates/pdfs-cli/src/main.rs b/crates/pdfs-cli/src/main.rs index ebbd61d..8f1c3f8 100644 --- a/crates/pdfs-cli/src/main.rs +++ b/crates/pdfs-cli/src/main.rs @@ -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) -> Result<()> { mount_once(mountpoint)?; Ok(()) diff --git a/crates/pdfs-core/src/control.rs b/crates/pdfs-core/src/control.rs index 254aa97..167e938 100644 --- a/crates/pdfs-core/src/control.rs +++ b/crates/pdfs-core/src/control.rs @@ -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 `` 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 }, + /// Every local Proton Drive location (reply to [`Request::ListLocations`]). + Locations { items: Vec }, /// This device's synced folders (reply to [`Request::ListSyncFolders`]). SyncFolders { items: Vec }, /// 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::(&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 = [ diff --git a/crates/pdfs-core/src/db/maintenance.rs b/crates/pdfs-core/src/db/maintenance.rs index cca4233..2c09e77 100644 --- a/crates/pdfs-core/src/db/maintenance.rs +++ b/crates/pdfs-core/src/db/maintenance.rs @@ -77,6 +77,7 @@ const REPORTED_TABLES: &[&str] = &[ "trash", "device", "sync_folder", + "mount", "sync_entry", "pending_op", "activity", diff --git a/crates/pdfs-core/src/db/migrations.rs b/crates/pdfs-core/src/db/migrations.rs index e79853f..e7241ac 100644 --- a/crates/pdfs-core/src/db/migrations.rs +++ b/crates/pdfs-core/src/db/migrations.rs @@ -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; +"; diff --git a/crates/pdfs-core/src/db/mod.rs b/crates/pdfs-core/src/db/mod.rs index d35a566..ed50884 100644 --- a/crates/pdfs-core/src/db/mod.rs +++ b/crates/pdfs-core/src/db/mod.rs @@ -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 { 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), }; diff --git a/crates/pdfs-core/src/db/mounts.rs b/crates/pdfs-core/src/db/mounts.rs new file mode 100644 index 0000000..6031c37 --- /dev/null +++ b/crates/pdfs-core/src/db/mounts.rs @@ -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 { + 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 { + 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> { + 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>(2)?, + row.get::<_, Option>(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>(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) + } +} diff --git a/crates/pdfs-core/src/db/sync.rs b/crates/pdfs-core/src/db/sync.rs index b813fe5..1aff0e1 100644 --- a/crates/pdfs-core/src/db/sync.rs +++ b/crates/pdfs-core/src/db/sync.rs @@ -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 { + 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(); diff --git a/crates/pdfs-core/src/db/tests.rs b/crates/pdfs-core/src/db/tests.rs index bafea25..050a89d 100644 --- a/crates/pdfs-core/src/db/tests.rs +++ b/crates/pdfs-core/src/db/tests.rs @@ -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(); diff --git a/crates/pdfs-core/src/lib.rs b/crates/pdfs-core/src/lib.rs index 8d4babb..5aaf5bc 100644 --- a/crates/pdfs-core/src/lib.rs +++ b/crates/pdfs-core/src/lib.rs @@ -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; diff --git a/crates/pdfs-core/src/mounts.rs b/crates/pdfs-core/src/mounts.rs new file mode 100644 index 0000000..615ff00 --- /dev/null +++ b/crates/pdfs-core/src/mounts.rs @@ -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, + /// 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, +} + +#[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::(r#""mirror""#).unwrap(), + MountMode::Mirror + ); + assert_eq!( + serde_json::from_str::(r#""ondemand""#).unwrap(), + MountMode::OnDemand + ); + } + + #[test] + fn mount_mode_tolerates_unknown_values() { + assert_eq!( + serde_json::from_str::(r#""from-the-future""#).unwrap(), + MountMode::Unknown + ); + } +} diff --git a/crates/pdfs-core/src/profile.rs b/crates/pdfs-core/src/profile.rs index 4351bb7..5f12b5f 100644 --- a/crates/pdfs-core/src/profile.rs +++ b/crates/pdfs-core/src/profile.rs @@ -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. diff --git a/crates/pdfs-fuse/src/control.rs b/crates/pdfs-fuse/src/control.rs index 212dd31..1de9170 100644 --- a/crates/pdfs-fuse/src/control.rs +++ b/crates/pdfs-fuse/src/control.rs @@ -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}"), diff --git a/crates/pdfs-fuse/src/devices.rs b/crates/pdfs-fuse/src/devices.rs index 4193f5d..3ca24f3 100644 --- a/crates/pdfs-fuse/src/devices.rs +++ b/crates/pdfs-fuse/src/devices.rs @@ -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 { + 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, 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( + teardown: Result<(), E>, + commit: impl FnOnce() -> T, +) -> Result { + 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> { + 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 { - 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(¤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}")); + } } - 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 { + 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 { - 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 { 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, ¤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 @@ -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, ¤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); @@ -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)); + } +} diff --git a/crates/pdfs-fuse/src/lib.rs b/crates/pdfs-fuse/src/lib.rs index 2a2f18f..6aebc3f 100644 --- a/crates/pdfs-fuse/src/lib.rs +++ b/crates/pdfs-fuse/src/lib.rs @@ -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>, + /// 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, /// In-flight upload/download progress, served to `GetQueueStatus`. Shared /// across the FUSE session and the control-socket task. transfers: Arc, @@ -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>>, + /// 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>>, /// 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>, Arc>); +type LiveMount = ( + PathBuf, + Arc>, + Arc>, + Arc, +); 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>, /// This mount's kernel notification channel — the same cell as /// [`Core::notifier`], still empty until its session is spawned. notifier: Arc>, + session_live: Arc, } /// Every mounted inode space in the daemon, shared by the primary `Core` and @@ -546,14 +560,28 @@ struct StateRegistry(Mutex>); 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>, notifier: Arc>) { + /// 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>, + notifier: Arc>, + session_live: Arc, + ) { 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 { + 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::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), diff --git a/crates/pdfs-fuse/src/mount.rs b/crates/pdfs-fuse/src/mount.rs index 9067f73..a97451e 100644 --- a/crates/pdfs-fuse/src/mount.rs +++ b/crates/pdfs-fuse/src/mount.rs @@ -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); +/// 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, + session_live: Arc, +} + +#[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 { + accepting: bool, + entries: HashMap, +} + +impl Default for SecondaryMountRegistry { + fn default() -> Self { + Self { + accepting: false, + entries: HashMap::new(), + } + } +} + +impl SecondaryMountRegistry { + 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 { + 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, + session_live: Arc, + ) -> 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 { + 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(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()); + } +} diff --git a/crates/pdfs-fuse/src/profile.rs b/crates/pdfs-fuse/src/profile.rs index 4ffd9fb..4836b5b 100644 --- a/crates/pdfs-fuse/src/profile.rs +++ b/crates/pdfs-fuse/src/profile.rs @@ -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 { + 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]