From 68e07ed426610731d83ae743549b63a5bcf5a747 Mon Sep 17 00:00:00 2001 From: Nils Pukropp Date: Fri, 17 Jul 2026 10:23:24 +0200 Subject: [PATCH] implemented offline scaffold --- Cargo.lock | 34 + crates/pdfs-cli/src/main.rs | 60 +- crates/pdfs-core/src/cache.rs | 133 ++++ crates/pdfs-core/src/control.rs | 82 ++- crates/pdfs-core/src/db.rs | 359 +++++++++- crates/pdfs-fuse/src/lib.rs | 1163 +++++++++++++++++++++++++++---- crates/pdfs-fuse/src/sync.rs | 563 +++++++++++++-- crates/pdfs-gui/src/app.rs | 280 +++++++- crates/pdfs-gui/src/main.rs | 13 +- 9 files changed, 2412 insertions(+), 275 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 1bf3bee..36d1ed8 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -270,6 +270,18 @@ dependencies = [ "pin-project-lite", ] +[[package]] +name = "async-compression" +version = "0.4.42" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e79b3f8a79cccc2898f31920fc69f304859b3bd567490f75ebf51ae1c792a9ac" +dependencies = [ + "compression-codecs", + "compression-core", + "pin-project-lite", + "tokio", +] + [[package]] name = "async-io" version = "2.6.0" @@ -923,6 +935,23 @@ dependencies = [ "memchr", ] +[[package]] +name = "compression-codecs" +version = "0.4.38" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ce2548391e9c1929c21bf6aa2680af86fe4c1b33e6cea9ac1cfeec0bd11218cf" +dependencies = [ + "compression-core", + "flate2", + "memchr", +] + +[[package]] +name = "compression-core" +version = "0.4.32" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cc14f565cf027a105f7a44ccf9e5b424348421a1d8952a8fc9d499d313107789" + [[package]] name = "concurrent-queue" version = "2.5.0" @@ -5043,12 +5072,17 @@ version = "0.6.11" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "4cfcf7e2740e6fc6d4d688b4ef00650406bb94adf4731e43c096c3a19fe40840" dependencies = [ + "async-compression", "bitflags 2.13.0", "bytes", + "futures-core", "futures-util", "http", "http-body", + "http-body-util", "pin-project-lite", + "tokio", + "tokio-util", "tower", "tower-layer", "tower-service", diff --git a/crates/pdfs-cli/src/main.rs b/crates/pdfs-cli/src/main.rs index 5ea80aa..e0b074d 100644 --- a/crates/pdfs-cli/src/main.rs +++ b/crates/pdfs-cli/src/main.rs @@ -11,7 +11,7 @@ use pdfs_core::auth; use pdfs_core::cache::ContentCache; use pdfs_core::config::AppDirs; use pdfs_core::control::{ - Request as CtlRequest, Response as CtlResponse, ShareEntryKind, SyncPhase, + RefreshScope, Request as CtlRequest, Response as CtlResponse, ShareEntryKind, SyncPhase, }; use pdfs_core::db::Db; @@ -146,6 +146,15 @@ enum Command { /// Permanently delete everything in the trash. This cannot be undone. EmptyTrash, + /// Drop a cached listing so the next read re-fetches it from the server. + /// Use it when another client changed something and the daemon hasn't + /// noticed yet. + Refresh { + /// What to refresh: a folder path (inside the mountpoint or relative to + /// it), `trash`, or `photos`. The mount root if omitted. + target: Option, + }, + /// Manage the account's registered devices. Devices { #[command(subcommand)] @@ -376,6 +385,7 @@ fn main() -> Result<()> { Command::Restore { uids } => cmd_restore(uids), Command::DeleteForever { uids } => cmd_delete_forever(uids), Command::EmptyTrash => cmd_empty_trash(), + Command::Refresh { target } => cmd_refresh(target), Command::Devices { action } => cmd_devices(action), Command::Sync { action } => cmd_sync(action), Command::Share { @@ -451,14 +461,23 @@ fn cmd_sync(action: SyncCmd) -> Result<()> { } else { f.last_sync.to_string() }; + // A queued switch is reported as the mode the folder is moving + // to, so `sync list` never reads as if the request was dropped. + let mode = match &f.pending_mode { + Some(pending) => format!("{} → {pending}", f.mode), + None => f.mode.clone(), + }; println!( - "[{}] {} ({}, {}, synced: {sync})", - f.id, f.local_path, f.mode, f.state + "[{}] {} ({mode}, {}, synced: {sync})", + f.id, f.local_path, f.state ); // A pass in flight says what it is doing, indented under it. if let Some(p) = &f.progress { match p.phase { - SyncPhase::Scanning => println!(" scanning…"), + SyncPhase::Scanning if p.total == 0 => println!(" scanning…"), + SyncPhase::Scanning => { + println!(" scanning: {} of {}", p.done, p.total.max(p.done)) + } SyncPhase::Applying => println!( " {} of {} {}", p.done + 1, @@ -809,9 +828,19 @@ fn cmd_status() -> Result<()> { // If a mount daemon is running, report live status from its control socket. match control_request(CtlRequest::Status) { Ok(CtlResponse::Status { - mountpoint, pinned, .. + mountpoint, + pinned, + online, + pending_uploads, + .. }) => { - println!("Mounted at {mountpoint} ({pinned} pinned)"); + let state = if online { "" } else { ", offline" }; + let queued = match pending_uploads { + 0 => String::new(), + 1 => ", 1 upload queued".to_string(), + n => format!(", {n} uploads queued"), + }; + println!("Mounted at {mountpoint} ({pinned} pinned{state}{queued})"); } Ok(other) => println!("Mount: unexpected response {other:?}"), Err(_) => println!("Mount: not running."), @@ -1166,6 +1195,25 @@ fn cmd_empty_trash() -> Result<()> { Ok(()) } +/// Drop a cached listing. `trash` and `photos` name those two listings; anything +/// else is read as a folder path, so a folder actually called "trash" is still +/// reachable as `./trash`. +fn cmd_refresh(target: Option) -> Result<()> { + let scope = match target.as_deref() { + Some("trash") => RefreshScope::Trash, + Some("photos") => RefreshScope::Photos, + path => RefreshScope::Dir { + path: path.unwrap_or("").to_string(), + }, + }; + match control_request(CtlRequest::Refresh { scope })? { + CtlResponse::Ok { message } => println!("{message}"), + CtlResponse::Error { message } => bail!("{message}"), + other => bail!("unexpected response: {other:?}"), + } + Ok(()) +} + /// Ask for a typed `yes` before an irreversible destroy. Errors (aborting the /// command) on anything else, including a non-interactive stdin: a piped or /// scripted invocation must not silently wipe the trash. diff --git a/crates/pdfs-core/src/cache.rs b/crates/pdfs-core/src/cache.rs index 9357a00..d57bc64 100644 --- a/crates/pdfs-core/src/cache.rs +++ b/crates/pdfs-core/src/cache.rs @@ -82,6 +82,34 @@ struct PinFile { pins: BTreeMap, } +/// What a staged file actually contains, written beside it as `.json` +/// (offline.md Phase 2). +/// +/// A staged file is **not** necessarily valid whole-file content, and that is +/// the whole reason this exists. A partial overwrite commits by filling the +/// untouched regions from the remote base — which is exactly what fails when the +/// network is down, so what lands in staging is the authored bytes with *zeros* +/// in the gaps. Uploading that as-is would silently corrupt the file. +/// `authored` says which ranges are real; `complete` says whether the file can +/// be uploaded as it stands. +#[derive(Serialize, Deserialize, Debug, Clone)] +pub struct StagedWrite { + /// Node the write targeted, in `volume~link` display form. + pub uid: String, + /// Length of the intended new content. + pub len: u64, + /// Size and mtime of the base revision the write was made against, so a + /// retry can tell whether the remote moved on in the meantime. + pub base_size: u64, + pub base_mtime: i64, + /// Locally authored `[start, end)` ranges. Everything else in the file is a + /// zero-filled hole, not content. + pub authored: Vec<(u64, u64)>, + /// True when `authored` covers the whole file, i.e. the staged bytes are the + /// complete new content and can be uploaded directly. + pub complete: bool, +} + /// Content cache rooted at a directory, with a sibling pin-registry file. pub struct ContentCache { /// Directory holding `` blobs and `.meta` tags. @@ -96,6 +124,12 @@ pub struct ContentCache { /// Subdirectory for write-handle scratch files (disk-backed write buffers). /// Emptied on open so a crashed run leaves no orphans. scratch_dir: PathBuf, + /// Subdirectory holding the bytes of writes that have not been uploaded yet + /// (offline.md Phase 2/3) — every released write passes through here, since + /// the upload is a queued op performed later. Unlike `scratch_dir` this is + /// **never** emptied on open: these are the only copy of content the user + /// authored, and the whole point is that they outlive the daemon. + staging_dir: PathBuf, /// JSON pin registry path. pins_path: PathBuf, /// Soft cap on total blob bytes. Exceeded only transiently: a `store` @@ -133,6 +167,9 @@ impl ContentCache { let scratch_dir = content_dir.join("scratch"); let _ = std::fs::remove_dir_all(&scratch_dir); std::fs::create_dir_all(&scratch_dir)?; + // Staging, by contrast, is deliberately preserved across runs. + let staging_dir = content_dir.join("staging"); + std::fs::create_dir_all(&staging_dir)?; if let Some(parent) = pins_path.parent() { std::fs::create_dir_all(parent)?; } @@ -141,6 +178,7 @@ impl ContentCache { thumb_dir, block_dir, scratch_dir, + staging_dir, pins_path, max_bytes: AtomicU64::new(max_bytes), db, @@ -414,6 +452,53 @@ impl ContentCache { Ok((file, path)) } + /// Move a released scratch file into the staging directory with a + /// [`StagedWrite`] sidecar, and return where it landed. + /// + /// This is what makes a write survive its upload: the caller is releasing a + /// write handle and would otherwise delete the file, so until the bytes are + /// on the remote, staging holds the only copy. Every dirty handle goes + /// through here — the upload is a queued op performed later (offline.md + /// Phase 3), and a staged file is also what a human can recover from if the + /// queue never drains. + /// + /// Falls back to a copy when the rename crosses a filesystem boundary; a + /// failure here means we could not save the bytes at all, so it is reported + /// rather than swallowed. + pub fn stage_write(&self, meta: &StagedWrite, scratch: &Path) -> Result { + static N: AtomicU64 = AtomicU64::new(0); + // The uid goes in the name so a staged file can be tied back to its node + // without a database; `/` in a uid would otherwise open a path. + let safe_uid = meta.uid.replace(['/', '\\'], "_"); + let path = self.staging_dir.join(format!( + "{}-{}-{}", + safe_uid, + now_millis(), + N.fetch_add(1, Ordering::Relaxed) + )); + if let Err(e) = std::fs::rename(scratch, &path) { + // Cross-device: rename cannot move it, so copy and drop the original. + let _ = e; + std::fs::copy(scratch, &path)?; + let _ = std::fs::remove_file(scratch); + } + // Sidecar last: a staged file without one is still recoverable bytes, + // while a sidecar without a file would describe nothing. + std::fs::write( + path.with_extension("json"), + serde_json::to_vec_pretty(meta)?, + )?; + Ok(path) + } + + /// Drop a staged blob and its sidecar, once its bytes are safely on the + /// remote (or the op that owned them has been superseded). Best effort: + /// leftovers cost disk, while failing here would strand a drained op. + pub fn discard_staged(&self, path: &Path) { + let _ = std::fs::remove_file(path); + let _ = std::fs::remove_file(path.with_extension("json")); + } + /// Remove any cached blob + meta for `uid`, including its thumbnails (best /// effort; absence is fine). Thumbnails are tied to the revision, so a /// content eviction must drop them too. @@ -988,6 +1073,54 @@ mod tests { assert!(c.cached_block(&u, 1, 12, 2).is_some(), "newest survives"); } + /// A failed upload must never cost the user their bytes (offline.md Phase 2): + /// the scratch file moves to staging intact, and staging survives a reopen + /// the way scratch deliberately does not. + #[test] + fn failed_write_is_staged_and_survives_reopen() { + use std::io::Write as _; + let dir = TempDir::new(); + let db = Arc::new(Db::open_in_memory().unwrap()); + let content = dir.path().join("content"); + let c = ContentCache::open(content.clone(), dir.path().join("pins.json"), 0, db.clone()) + .unwrap(); + + let (mut f, scratch) = c.create_scratch().unwrap(); + f.write_all(b"unsent work").unwrap(); + drop(f); + + // A partial overwrite: only the first 11 bytes are real, so the staged + // file must not be mistaken for uploadable whole-file content. + let meta = StagedWrite { + uid: uid("a").to_string(), + len: 40, + base_size: 40, + base_mtime: 100, + authored: vec![(0, 11)], + complete: false, + }; + let staged = c.stage_write(&meta, &scratch).unwrap(); + assert!(!scratch.exists(), "scratch is moved, not copied"); + assert_eq!(std::fs::read(&staged).unwrap(), b"unsent work"); + + let sidecar: StagedWrite = + serde_json::from_slice(&std::fs::read(staged.with_extension("json")).unwrap()).unwrap(); + assert_eq!(sidecar.authored, vec![(0, 11)]); + assert!( + !sidecar.complete, + "gaps are zeros, not content: uploading this as-is would corrupt the file" + ); + + // Reopening wipes scratch (worthless leftovers) but must keep staging. + drop(c); + let _c2 = ContentCache::open(content, dir.path().join("pins.json"), 0, db).unwrap(); + assert_eq!( + std::fs::read(&staged).unwrap(), + b"unsent work", + "staged bytes outlive the daemon that failed to upload them" + ); + } + #[test] fn scratch_file_is_writable_and_isolated() { use std::io::{Read as _, Seek as _, SeekFrom, Write as _}; diff --git a/crates/pdfs-core/src/control.rs b/crates/pdfs-core/src/control.rs index 651a6fa..2bf930a 100644 --- a/crates/pdfs-core/src/control.rs +++ b/crates/pdfs-core/src/control.rs @@ -132,6 +132,14 @@ pub enum Request { /// Replies with [`Response::Ok`]. EmptyTrash, + /// Drop a cached listing so the *next* read of it re-enumerates from the + /// server. Replies with [`Response::Ok`]. This is what a front-end's Refresh + /// button raises: the daemon serves listings from its persisted cache, which + /// only notices another client's changes when its TTL lapses, so a user who + /// knows the cache is stale needs a way to say so. Cheap and idempotent — + /// it invalidates, it does not fetch. + Refresh { scope: RefreshScope }, + // ---- devices ---------------------------------------------------------- /// List the account's registered devices. Replies with [`Response::Devices`]. ListDevices, @@ -258,6 +266,23 @@ pub enum ShareEntryKind { ExternalInvite, } +/// Which cached listing a [`Request::Refresh`] drops. +/// +/// Only the listings the daemon caches need naming here — the sharing, devices +/// and activity listings are always fetched live, so a front-end refreshes those +/// by simply re-asking. +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq)] +pub enum RefreshScope { + /// One folder's child listing, by mountpoint-relative path (`""` = root). + /// Only the folder itself, not its subtree: refreshing what the user is + /// looking at shouldn't re-walk everything below it. + Dir { path: String }, + /// The trash listing. + Trash, + /// The photos timeline. + Photos, +} + /// A registered device in a [`Response::Devices`] listing. #[derive(Serialize, Deserialize, Debug, Clone)] pub struct DeviceInfo { @@ -269,6 +294,11 @@ pub struct DeviceInfo { pub device_type: String, /// Last sync time, epoch seconds; `None` if it never synced. pub last_sync: Option, + /// Whether this is the device *this* machine syncs to. Deleting it would + /// delete the cloud copy of the folders this machine is syncing, so a + /// front-end must not offer that as casually as removing another computer. + #[serde(default)] + pub this_device: bool, } /// One synced local folder on this machine's device (in [`Response::SyncFolders`]). @@ -282,6 +312,12 @@ pub struct SyncFolderInfo { pub remote_uid: String, /// `mirror` (full local copy, two-way synced) or `ondemand` (FUSE mount). pub mode: String, + /// A mode switch the user asked for that the daemon has queued: it applies + /// once the folder's current pass has pushed any local changes up. `None` + /// when nothing is queued. A front-end should paint the folder as already + /// heading there — the request was accepted, not rejected. + #[serde(default)] + pub pending_mode: Option, /// `idle` | `syncing` | `error` | `conflict`. pub state: String, /// Last successful sync, epoch seconds; `0` if never. @@ -296,7 +332,8 @@ pub struct SyncFolderInfo { #[derive(Serialize, Deserialize, Debug, Clone, Copy, PartialEq, Eq)] pub enum SyncPhase { /// Walking the local tree, the remote tree and the stored baseline to work - /// out what changed. No per-item counts exist yet. + /// out what changed. `done` counts the items checked so far; `total` is how + /// many the last pass saw, so it is an estimate the walk can overshoot. Scanning, /// Applying the diff: creating folders, uploading, downloading, deleting. Applying, @@ -307,12 +344,15 @@ pub enum SyncPhase { #[derive(Serialize, Deserialize, Debug, Clone)] pub struct SyncProgress { pub phase: SyncPhase, - /// Items applied so far this pass. + /// Items checked ([`SyncPhase::Scanning`]) or applied ([`SyncPhase::Applying`]) + /// so far this pass. pub done: usize, - /// Items known to need applying. This *grows* during a pass: paths are - /// classified depth by depth (a folder must exist remotely before its - /// children can be queued), so the total is what has been discovered so far, - /// not a figure fixed up front. + /// How many items `done` is counting towards. Neither phase can fix this up + /// front, so it moves: while scanning it is the size of the last pass's + /// baseline — an estimate the walk may overshoot when the folder has grown — + /// and while applying it *grows*, because paths are classified depth by depth + /// (a folder must exist remotely before its children can be queued). `0` means + /// no estimate exists (a folder that has never synced), i.e. indeterminate. pub total: usize, /// The name of an item currently being applied, or empty between items. /// Several run at once; this is just the most recently started. @@ -579,6 +619,12 @@ pub struct PhotoThumb { pub pending: bool, } +/// A daemon too old to report connectivity is assumed online — it could not +/// have mounted at all otherwise. +fn default_online() -> bool { + true +} + /// The daemon's reply to a [`Request`]. #[derive(Serialize, Deserialize, Debug)] pub enum Response { @@ -595,6 +641,16 @@ pub enum Response { budget: u64, /// The pin registry. pins: Vec, + /// False when the daemon is serving the cached tree because the API is + /// unreachable (offline.md Phase 1). Cached and pinned content still + /// reads; anything else fails until the network is back. + #[serde(default = "default_online")] + online: bool, + /// Writes accepted locally but not yet uploaded (offline.md Phase 3). + /// Non-zero means the mount is ahead of the remote — either a copy is + /// still draining, or it cannot drain because we are offline. + #[serde(default)] + pending_uploads: u64, }, /// A human-readable success message. Ok { message: String }, @@ -729,6 +785,20 @@ mod tests { uids: vec!["vol~link".into()], }, Request::EmptyTrash, + Request::Refresh { + scope: RefreshScope::Dir { path: "a/b".into() }, + }, + Request::Refresh { + scope: RefreshScope::Dir { + path: String::new(), + }, + }, + Request::Refresh { + scope: RefreshScope::Trash, + }, + Request::Refresh { + scope: RefreshScope::Photos, + }, ]; for req in reqs { let line = serde_json::to_string(&req).unwrap(); diff --git a/crates/pdfs-core/src/db.rs b/crates/pdfs-core/src/db.rs index d6a5ff8..b830007 100644 --- a/crates/pdfs-core/src/db.rs +++ b/crates/pdfs-core/src/db.rs @@ -23,12 +23,44 @@ use crate::control::{ActivityEntry, ActivityKind}; use crate::localindex::LocalEntry; /// Current schema version. Bump on every forward migration added below. -const SCHEMA_VERSION: i64 = 9; +const SCHEMA_VERSION: i64 = 11; /// How many activity rows to keep. Older rows are pruned on insert, so the feed /// stays a bounded "recent history" rather than growing without limit. const ACTIVITY_KEEP: i64 = 2000; +/// The `kind` of a [`PendingOp`] that uploads a staged file as a new revision. +/// The only kind so far: metadata mutations still go straight to the API and so +/// still need the network (offline.md Phase 3 is landing in two halves). +pub const OP_REVISION: &str = "revision"; + +/// A mutation that has been accepted locally but not yet performed against the +/// API — the durable half of the write-back queue (offline.md Phase 3). +/// +/// The daemon answers the FUSE call the moment this row and its staged blob are +/// on disk, so a `cp` into the mount runs at disk speed and the upload happens +/// behind it. That also makes an offline write succeed rather than EIO: the row +/// simply waits for the network. +#[derive(Debug, Clone)] +pub struct PendingOp { + /// Row id, `0` on a value being inserted. + pub id: i64, + /// See [`OP_REVISION`]. + pub kind: String, + /// Node this op targets. + pub uid: String, + /// Staged blob holding the bytes to upload. + pub blob_path: Option, + /// Serialized [`StagedWrite`](crate::cache::StagedWrite). + pub meta_json: Option, + /// When the op was queued (ms since epoch). + pub created_at: i64, + pub attempts: i64, + pub last_error: Option, + /// Earliest ms at which to retry, for backoff. + pub next_attempt_at: i64, +} + /// Size the WAL is truncated back to after a checkpoint. Comfortably above the /// steady-state working set (a few MB), so the truncation only claws back the /// outliers rather than fighting the normal write path for disk. @@ -107,6 +139,11 @@ pub struct StoredSyncFolder { pub remote_share_id: String, /// `mirror` (full local copy, two-way synced) or `ondemand` (FUSE mount). pub mode: String, + /// A mode the user asked for that could not be applied on the spot — the + /// folder was mid-pass, or had un-uploaded changes. The engine applies it + /// once the folder is safe to switch, so the request is queued rather than + /// rejected. `None` when the folder is where the user wants it. + pub pending_mode: Option, /// `idle` | `syncing` | `error` | `conflict`. pub state: String, pub last_sync: i64, @@ -224,6 +261,12 @@ impl Db { if current < 9 { tx.execute_batch(MIGRATION_V9)?; } + if current < 10 { + tx.execute_batch(MIGRATION_V10)?; + } + if current < 11 { + tx.execute_batch(MIGRATION_V11)?; + } tx.execute( "INSERT INTO sync_state (key, value) VALUES ('schema_version', ?1) ON CONFLICT(key) DO UPDATE SET value = excluded.value", @@ -374,6 +417,24 @@ impl Db { Ok(out) } + /// Load one persisted node back by uid. Used to recover the My Files root + /// when the API is unreachable, so the mount can still serve the cached tree + /// (offline.md Phase 1). + pub fn node_by_uid(&self, uid: &str) -> Result> { + let conn = self.conn.lock().unwrap(); + let json: Option = conn + .query_row( + "SELECT node_json FROM nodes WHERE uid = ?1 AND node_json IS NOT NULL", + params![uid], + |r| r.get(0), + ) + .optional()?; + match json { + Some(json) => Ok(Some(serde_json::from_str(&json)?)), + None => Ok(None), + } + } + /// Read the persisted incremental-sync cursor (a `DriveEventId`), if any. /// The daemon resumes from this on restart instead of reseeding to the /// server head, so changes made while unmounted are still applied (P2). @@ -400,6 +461,108 @@ impl Db { Ok(()) } + /// Read a `sync_state` value as a string. + pub fn state_str(&self, key: &str) -> Result> { + let conn = self.conn.lock().unwrap(); + let v = conn + .query_row( + "SELECT value FROM sync_state WHERE key = ?1", + params![key], + |r| r.get::<_, String>(0), + ) + .optional()?; + Ok(v) + } + + /// Write a `sync_state` string value. + pub fn set_state_str(&self, key: &str, value: &str) -> Result<()> { + let conn = self.conn.lock().unwrap(); + conn.execute( + "INSERT INTO sync_state (key, value) VALUES (?1, ?2) + ON CONFLICT(key) DO UPDATE SET value = excluded.value", + params![key, value], + )?; + Ok(()) + } + + /// Queue an upload to be performed by the drain worker, returning its row id. + /// + /// One pending op per node: a second write to the same file before the first + /// has drained replaces it, since the newer blob already contains everything + /// the older one did. The superseded blob's path is returned so the caller can + /// delete it. + pub fn enqueue_op(&self, op: &PendingOp) -> Result<(i64, Option)> { + let conn = self.conn.lock().unwrap(); + let superseded: Option = conn + .query_row( + "SELECT blob_path FROM pending_op WHERE uid = ?1", + params![op.uid], + |r| r.get(0), + ) + .optional()? + .flatten(); + conn.execute("DELETE FROM pending_op WHERE uid = ?1", params![op.uid])?; + conn.execute( + "INSERT INTO pending_op (kind, uid, blob_path, meta_json, created_at) + VALUES (?1, ?2, ?3, ?4, ?5)", + params![op.kind, op.uid, op.blob_path, op.meta_json, op.created_at], + )?; + Ok((conn.last_insert_rowid(), superseded)) + } + + /// Every queued op, oldest first. The drain worker replays them in this order + /// so a file's writes land in the order they were made. + pub fn pending_ops(&self) -> Result> { + let conn = self.conn.lock().unwrap(); + let mut stmt = conn.prepare( + "SELECT id, kind, uid, blob_path, meta_json, created_at, attempts, + last_error, next_attempt_at + FROM pending_op ORDER BY id", + )?; + let rows = stmt + .query_map([], |r| { + Ok(PendingOp { + id: r.get(0)?, + kind: r.get(1)?, + uid: r.get(2)?, + blob_path: r.get(3)?, + meta_json: r.get(4)?, + created_at: r.get(5)?, + attempts: r.get(6)?, + last_error: r.get(7)?, + next_attempt_at: r.get(8)?, + }) + })? + .collect::, _>>()?; + Ok(rows) + } + + /// How many uploads are still waiting, for `Response::Status`. + pub fn pending_op_count(&self) -> Result { + let conn = self.conn.lock().unwrap(); + Ok(conn.query_row("SELECT COUNT(*) FROM pending_op", [], |r| r.get(0))?) + } + + /// Drop a queued op, once its upload has actually landed. + pub fn delete_op(&self, id: i64) -> Result<()> { + let conn = self.conn.lock().unwrap(); + conn.execute("DELETE FROM pending_op WHERE id = ?1", params![id])?; + Ok(()) + } + + /// Record a failed attempt and when to next try. Leaves the row in place — + /// the staged bytes are still the only copy of the user's write. + pub fn record_op_failure(&self, id: i64, error: &str, next_attempt_at: i64) -> Result<()> { + let conn = self.conn.lock().unwrap(); + conn.execute( + "UPDATE pending_op + SET attempts = attempts + 1, last_error = ?2, next_attempt_at = ?3 + WHERE id = ?1", + params![id, error, next_attempt_at], + )?; + Ok(()) + } + /// Read a `sync_state` value as an integer (the freshness stamps of the /// photos timeline and the trash listing are kept there). pub fn state_i64(&self, key: &str) -> Result> { @@ -644,7 +807,7 @@ impl Db { pub fn sync_folder_list(&self) -> Result> { let conn = self.conn.lock().unwrap(); let mut stmt = conn.prepare( - "SELECT id, local_path, remote_uid, remote_share_id, mode, state, last_sync + "SELECT id, local_path, remote_uid, remote_share_id, mode, pending_mode, state, last_sync FROM sync_folder ORDER BY id", )?; let rows = stmt.query_map([], |r| { @@ -654,8 +817,9 @@ impl Db { remote_uid: r.get(2)?, remote_share_id: r.get(3)?, mode: r.get(4)?, - state: r.get(5)?, - last_sync: r.get(6)?, + pending_mode: r.get(5)?, + state: r.get(6)?, + last_sync: r.get(7)?, }) })?; let mut items = Vec::new(); @@ -669,7 +833,7 @@ impl Db { pub fn sync_folder_get(&self, id: i64) -> Result> { let conn = self.conn.lock().unwrap(); conn.query_row( - "SELECT id, local_path, remote_uid, remote_share_id, mode, state, last_sync + "SELECT id, local_path, remote_uid, remote_share_id, mode, pending_mode, state, last_sync FROM sync_folder WHERE id = ?1", params![id], |r| { @@ -679,8 +843,9 @@ impl Db { remote_uid: r.get(2)?, remote_share_id: r.get(3)?, mode: r.get(4)?, - state: r.get(5)?, - last_sync: r.get(6)?, + pending_mode: r.get(5)?, + state: r.get(6)?, + last_sync: r.get(7)?, }) }, ) @@ -698,11 +863,25 @@ impl Db { Ok(n > 0) } - /// Update a synced folder's mode (`mirror`/`ondemand`). + /// 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. pub fn sync_folder_set_mode(&self, id: i64, mode: &str) -> Result<()> { let conn = self.conn.lock().unwrap(); conn.execute( - "UPDATE sync_folder SET mode = ?2 WHERE id = ?1", + "UPDATE sync_folder SET mode = ?2, pending_mode = NULL WHERE id = ?1", + params![id, mode], + )?; + Ok(()) + } + + /// Queue (or, with `None`, withdraw) a mode the folder should move to once it + /// is safe to switch. + pub fn sync_folder_set_pending_mode(&self, id: i64, mode: Option<&str>) -> Result<()> { + let conn = self.conn.lock().unwrap(); + conn.execute( + "UPDATE sync_folder SET pending_mode = ?2 WHERE id = ?1", params![id, mode], )?; Ok(()) @@ -1481,6 +1660,41 @@ CREATE TABLE activity ( CREATE INDEX activity_time ON activity(time DESC); "; +/// Schema v10: a mode switch the user asked for is queued rather than rejected +/// when the folder is mid-pass or has un-uploaded changes, so `pending_mode` +/// records the intent until the engine can act on it. NULL is the resting state: +/// the folder is already where the user wants it. +const MIGRATION_V10: &str = " +ALTER TABLE sync_folder ADD COLUMN pending_mode TEXT; +"; + +/// Schema v11: writes no longer upload inside the FUSE handler. `release` stages +/// the bytes and records the intended upload here, and a drain worker performs it +/// (offline.md Phase 3). The row outlives the process, so a write survives both a +/// dead network and a restart. +/// +/// `blob_path` points into the content cache's `staging/` dir and `meta_json` is +/// the [`StagedWrite`] sidecar describing which of its bytes are real — an +/// incomplete blob must be gap-filled from the remote base before it can be +/// uploaded. `next_attempt_at` is a ms deadline implementing retry backoff. +/// +/// [`StagedWrite`]: crate::cache::StagedWrite +const MIGRATION_V11: &str = " +CREATE TABLE pending_op ( + id INTEGER PRIMARY KEY, + kind TEXT NOT NULL, + uid TEXT NOT NULL, + blob_path TEXT, + meta_json TEXT, + created_at INTEGER NOT NULL, + attempts INTEGER NOT NULL DEFAULT 0, + last_error TEXT, + next_attempt_at INTEGER NOT NULL DEFAULT 0 +); + +CREATE INDEX pending_op_uid ON pending_op(uid); +"; + #[cfg(test)] mod tests { use super::*; @@ -1666,6 +1880,26 @@ mod tests { ) } + /// Recovering the root by uid is what lets the daemon mount offline + /// (offline.md Phase 1): the uid is remembered in `sync_state`, the node + /// itself comes back out of `nodes`. + #[test] + fn node_by_uid_recovers_a_stored_node() { + let db = Db::open_in_memory().unwrap(); + let root = folder("root", None, "My Files"); + db.upsert_node(&root).unwrap(); + db.set_state_str("root_uid", &root.uid.to_string()).unwrap(); + + let key = db.state_str("root_uid").unwrap().unwrap(); + let got = db.node_by_uid(&key).unwrap().expect("root recovered"); + assert_eq!(got.uid, root.uid); + assert_eq!(got.name, "My Files"); + assert!(got.is_folder()); + + assert!(db.node_by_uid("vol~nope").unwrap().is_none()); + assert!(db.state_str("never_written").unwrap().is_none()); + } + #[test] fn upsert_and_load_all_roundtrip() { let db = Db::open_in_memory().unwrap(); @@ -1948,6 +2182,72 @@ mod tests { assert_eq!(version, SCHEMA_VERSION.to_string()); } + #[test] + fn a_second_write_supersedes_the_first_pending_op() { + let db = Db::open_in_memory().unwrap(); + let op = |blob: &str| PendingOp { + id: 0, + kind: OP_REVISION.to_string(), + uid: uid("a").to_string(), + blob_path: Some(blob.to_string()), + meta_json: Some("{}".to_string()), + created_at: 1, + attempts: 0, + last_error: None, + next_attempt_at: 0, + }; + + let (_, superseded) = db.enqueue_op(&op("/staging/first")).unwrap(); + assert_eq!(superseded, None, "nothing to supersede on the first write"); + + // The newer blob already contains everything the older one did, so the + // older op must go — and its blob must be reported so it can be deleted + // rather than leaked. + let (id2, superseded) = db.enqueue_op(&op("/staging/second")).unwrap(); + assert_eq!(superseded.as_deref(), Some("/staging/first")); + + let ops = db.pending_ops().unwrap(); + assert_eq!(ops.len(), 1, "one queued upload per node"); + assert_eq!(ops[0].id, id2); + assert_eq!(ops[0].blob_path.as_deref(), Some("/staging/second")); + assert_eq!(db.pending_op_count().unwrap(), 1); + } + + #[test] + fn a_failed_op_stays_queued_with_backoff() { + let db = Db::open_in_memory().unwrap(); + db.enqueue_op(&PendingOp { + id: 0, + kind: OP_REVISION.to_string(), + uid: uid("a").to_string(), + blob_path: Some("/staging/blob".to_string()), + meta_json: Some("{}".to_string()), + created_at: 1, + attempts: 0, + last_error: None, + next_attempt_at: 0, + }) + .unwrap(); + let id = db.pending_ops().unwrap()[0].id; + + db.record_op_failure(id, "network unreachable", 5_000) + .unwrap(); + + // The staged blob is the only copy of the user's bytes: a failure must + // never drop the row, only defer it. + let ops = db.pending_ops().unwrap(); + assert_eq!(ops.len(), 1); + assert_eq!(ops[0].attempts, 1); + assert_eq!(ops[0].last_error.as_deref(), Some("network unreachable")); + assert_eq!(ops[0].next_attempt_at, 5_000); + + db.record_op_failure(id, "still down", 9_000).unwrap(); + assert_eq!(db.pending_ops().unwrap()[0].attempts, 2); + + db.delete_op(id).unwrap(); + assert_eq!(db.pending_op_count().unwrap(), 0); + } + #[test] fn migrate_is_idempotent() { let db = Db::open_in_memory().unwrap(); @@ -1955,6 +2255,47 @@ mod tests { db.migrate().unwrap(); } + /// A queued mode switch is a promise the daemon has to keep across a restart, + /// so it lives in the row, and reaching the mode is what retires it. + #[test] + fn pending_mode_is_queued_until_the_mode_is_reached() { + let db = Db::open_in_memory().unwrap(); + let id = db + .sync_folder_add("/home/me/Downloads", "v~l", "s") + .unwrap(); + assert_eq!(db.sync_folder_get(id).unwrap().unwrap().pending_mode, None); + + db.sync_folder_set_pending_mode(id, Some("ondemand")) + .unwrap(); + assert_eq!( + db.sync_folder_get(id) + .unwrap() + .unwrap() + .pending_mode + .as_deref(), + Some("ondemand") + ); + // The listing carries it too — it is what the front-ends paint from. + assert_eq!( + db.sync_folder_list().unwrap()[0].pending_mode.as_deref(), + Some("ondemand") + ); + + // Landing the switch satisfies the request: a `pending_mode` outliving it + // would have the engine try to apply the same switch on every later pass. + db.sync_folder_set_mode(id, "ondemand").unwrap(); + let folder = db.sync_folder_get(id).unwrap().unwrap(); + assert_eq!(folder.mode, "ondemand"); + assert_eq!(folder.pending_mode, None); + + // And the user can withdraw a request that hasn't landed yet. + db.sync_folder_set_pending_mode(id, Some("mirror")).unwrap(); + db.sync_folder_set_pending_mode(id, None).unwrap(); + let folder = db.sync_folder_get(id).unwrap().unwrap(); + assert_eq!(folder.mode, "ondemand"); + assert_eq!(folder.pending_mode, None); + } + #[test] fn cache_index_touch_access_and_lru_order() { let db = Db::open_in_memory().unwrap(); diff --git a/crates/pdfs-fuse/src/lib.rs b/crates/pdfs-fuse/src/lib.rs index 3b9349d..1781aea 100644 --- a/crates/pdfs-fuse/src/lib.rs +++ b/crates/pdfs-fuse/src/lib.rs @@ -39,7 +39,7 @@ use std::os::unix::fs::FileExt; use std::os::unix::net::{UnixListener, UnixStream}; use std::path::{Component, Path, PathBuf}; use std::sync::atomic::{AtomicBool, Ordering}; -use std::sync::{Arc, Mutex}; +use std::sync::{Arc, Condvar, Mutex}; use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; use fuser::ReplyXattr; @@ -49,27 +49,29 @@ use fuser::{ RenameFlags, ReplyAttr, ReplyCreate, ReplyData, ReplyDirectory, ReplyEmpty, ReplyEntry, ReplyOpen, ReplyWrite, Request, Session, TimeOrNow, WriteFlags, }; -use pdfs_core::cache::{BLOCK_SIZE, ContentCache}; +use pdfs_core::cache::{BLOCK_SIZE, ContentCache, StagedWrite}; use pdfs_core::config::AppDirs; use pdfs_core::control::{ ActivityEntry, ActivityKind, BookmarkInfo, DeviceInfo, DirEntry, InvitationInfo, JobItem, - LocalHit, PhotoItem, PhotoThumb, PublicLinkInfo, Request as CtlRequest, + LocalHit, PhotoItem, PhotoThumb, PublicLinkInfo, RefreshScope, Request as CtlRequest, Response as CtlResponse, SearchHit, ShareEntry, ShareEntryKind, SharedItem, SyncFolderInfo, SyncPhase, SyncProgress, TransferDirection, }; use pdfs_core::db::{ - self, Db, StoredDevice, StoredNode, StoredPhoto, StoredSyncFolder, StoredTrash, + self, Db, OP_REVISION, PendingOp, StoredDevice, StoredNode, StoredPhoto, StoredSyncFolder, + StoredTrash, }; use pdfs_core::localindex; use proton_drive_rs::proton_sdk::error::ProtonError; use proton_drive_rs::proton_sdk::ids::{DeviceUid, DriveEventId, LinkId, NodeUid, VolumeId}; use proton_drive_rs::{ DeviceType, DriveEvent, DriveEventScopeId, MemberRole, Node, NodeKind, ProtonDriveClient, - ProtonPhotosClient, ThumbnailType, + ProtonPhotosClient, RevisionReader, ThumbnailType, }; mod sync; mod transfers; +use sync::base_name; use tracing::{debug, error, info, warn}; use transfers::{CountingReader, CountingWriter, JobGuard, OwnedCountingReader, TransferRegistry}; @@ -80,6 +82,22 @@ const TTL: Duration = Duration::from_secs(30); /// How often the background task polls the remote event cursor. const POLL_INTERVAL: Duration = Duration::from_secs(10); +/// First and longest delay between probes for the network coming back after an +/// offline mount (offline.md Phase 1). Doubles from min to max: a laptop shut in +/// a bag is the common case, so the steady state must be cheap, while a brief +/// blip should still recover in seconds. +/// Retry backoff for a queued upload, doubling per attempt between these. The +/// floor is short because the common failure is a brief network blip; the +/// ceiling keeps a persistently failing op from spinning. +const DRAIN_BACKOFF_MIN: Duration = Duration::from_secs(2); +const DRAIN_BACKOFF_MAX: Duration = Duration::from_secs(300); +/// How long the drain worker sleeps when it has nothing due. It is woken +/// directly on a new write or a reconnect, so this only bounds how late a +/// backoff can fire. +const DRAIN_IDLE_POLL: Duration = Duration::from_secs(30); + +const ONLINE_PROBE_MIN: Duration = Duration::from_secs(5); +const ONLINE_PROBE_MAX: Duration = Duration::from_secs(300); /// How long the persisted photos timeline stays good before a page request /// revalidates it. The SDK hands back the whole timeline at once, so it is stored /// in the DB and every page is sliced from there; a stale one is still served @@ -224,6 +242,19 @@ struct WriteHandle { dirty: bool, } +/// A released write whose upload has not happened yet (offline.md Phase 3). +/// +/// The bytes live in the content cache's staging dir and the intent lives in the +/// `pending_op` table; this pairs them in memory so a read can be served without +/// a database round trip. +#[derive(Clone)] +struct PendingRevision { + /// Staged blob holding the written bytes. + path: PathBuf, + /// Which of those bytes are real, and what base the gaps refer to. + meta: StagedWrite, +} + /// Mutable inode bookkeeping, guarded by a mutex because fuser drives the /// `Filesystem` trait through `&self`. struct State { @@ -414,6 +445,28 @@ async fn run_uploads( } /// Shared engine behind both the FUSE callbacks and the control socket: the +/// How many [`RevisionReader`]s stay open at once. +/// +/// A reader holds its revision's content key and block table — a few KB even for +/// a large file, so this is bounded for tidiness and staleness rather than +/// memory. Evicted least-recently-used; a dropped reader costs one re-open +/// (two API calls and a node-key unlock) the next time that file is read. +const MAX_OPEN_READERS: usize = 64; + +/// An open reader plus the node metadata it was opened against. +/// +/// The SDK pins a reader to the revision that was active at `open_revision`, so +/// a reader is only reusable while the node still reports the same +/// `(mtime, size)` — the same validity pair the content cache uses (a new +/// revision bumps mtime). On a mismatch the reader is dropped and reopened. +struct CachedReader { + reader: Arc, + mtime: i64, + size: u64, + /// For LRU eviction. + last_used: Instant, +} + /// Drive client, a Tokio handle to bridge the synchronous FUSE/socket threads /// to the async SDK, the inode bookkeeping, and the on-disk content cache. /// @@ -425,10 +478,31 @@ struct Core { rt: tokio::runtime::Handle, state: Arc>, cache: Arc, + /// Open [`RevisionReader`]s keyed by node, so the block fetches of a file + /// resolve its keys and block table once instead of once per block. + /// Validated by `(mtime, size)` exactly like the content cache, and bounded + /// by [`MAX_OPEN_READERS`]. + readers: Arc>>, /// Unified SQLite metadata cache: the persistence layer behind the in-memory /// `State` maps. Every mutation writes through here, and the maps rehydrate /// from it on mount (plan.md P1). db: Arc, + /// False while the API is unreachable and we are serving the cached tree + /// (offline.md Phase 1). Set by the probe thread; read by front-ends through + /// `Response::Status` so the UI can say so rather than leaving the user to + /// infer it from a wall of EIO. + online: Arc, + /// Writes accepted from the kernel but not yet uploaded, keyed by node + /// (offline.md Phase 3). The in-memory face of the `pending_op` table, from + /// which it is rebuilt on mount. + /// + /// Two things read it: [`Core::read_range`], because until the op drains the + /// staged blob *is* the file's content and the remote still holds the old + /// revision; and the drain worker, which performs the uploads. + pending: Arc>>, + /// Nudges the drain worker: set true and notify to have it re-examine the + /// queue instead of waiting out its backoff. + drain_wake: Arc<(Mutex, Condvar)>, /// True while a background refresh of the photos timeline (resp. the trash) is /// already running, so a burst of page requests against a stale listing kicks /// off one refresh rather than one per request. @@ -466,6 +540,17 @@ struct Core { sync_locks: Arc>>>>, } +/// Why [`Core::apply_sync_folder_mode`] did not switch a folder. The two cases are +/// answered very differently: `NotNow` is the normal state of a folder that is busy +/// or has local changes still to push, and the caller queues the request; `Failed` +/// is a real fault the user has to hear about. +enum SwitchBlocked { + /// The folder is mid-pass, or not yet safe to switch. Try again after a pass. + NotNow, + /// The switch was attempted and broke. + Failed(String), +} + impl Core { /// Rehydrate the in-memory `State` maps from the DB on mount, so a cold /// start serves previously-seen metadata (stable inodes, instant listings) @@ -537,6 +622,86 @@ impl Core { info!(nodes = st.entries.len(), "hydrated metadata cache from db"); } + /// Rebuild the in-memory pending map from the `pending_op` table on mount + /// (offline.md Phase 3). + /// + /// A queued write survives a restart — that is the point of persisting it — + /// so until the drain worker gets to it, reads of that file must still come + /// from its staged blob rather than the remote's older revision. + /// + /// A row whose blob has gone missing is dropped: there is nothing left to + /// upload, and keeping it would fail forever. + fn hydrate_pending(&self) { + let ops = match self.db.pending_ops() { + Ok(ops) => ops, + Err(e) => { + error!(error = %e, "loading pending uploads failed"); + return; + } + }; + let mut map = self.pending.lock().unwrap(); + for op in ops { + let parsed = op + .meta_json + .as_deref() + .and_then(|j| serde_json::from_str::(j).ok()) + .zip(op.blob_path.as_deref().map(PathBuf::from)) + .zip(parse_node_uid(&op.uid)); + let Some(((meta, path), uid)) = parsed else { + error!(uid = %op.uid, id = op.id, "pending op is unreadable; dropping"); + let _ = self.db.delete_op(op.id); + continue; + }; + if !path.exists() { + error!(%uid, path = %path.display(), "staged blob is gone; dropping pending op"); + let _ = self.db.delete_op(op.id); + continue; + } + map.insert(uid, PendingRevision { path, meta }); + } + if !map.is_empty() { + info!(count = map.len(), "restored pending uploads"); + } + } + + /// Poll for the API becoming reachable again after an offline mount, then + /// flip `online` and refresh the root (offline.md Phase 1). Runs on its own + /// thread and returns once we are back online: nothing sets `online` false + /// again, because a mount that has been online once keeps its live event + /// sync, which does its own retrying. + /// + /// Backs off to [`ONLINE_PROBE_MAX`] rather than hammering a fixed interval — + /// a laptop can sit offline for days, and each probe is a real API round trip. + fn run_online_probe(&self) { + let mut delay = ONLINE_PROBE_MIN; + loop { + std::thread::sleep(delay); + match self.rt.block_on(self.client.get_my_files_folder()) { + Ok(root) => { + { + let mut st = self.state.lock().unwrap(); + if let Some(e) = st.entries.get_mut(&ROOT_INO) { + e.node = root.clone(); + } + } + if let Err(e) = self.db.upsert_node(&root) { + warn!(error = %e, "refresh root after reconnect failed"); + } + self.online.store(true, Ordering::Relaxed); + // Anything written while offline is queued and waiting on + // exactly this. + self.wake_drain(); + info!("back online"); + return; + } + Err(e) => { + debug!(error = %e, ?delay, "online probe failed; still offline"); + delay = (delay * 2).min(ONLINE_PROBE_MAX); + } + } + } + } + /// Enumerate `ino`'s children from the remote and cache them. No-op if the /// directory has already been listed. Network I/O happens without the lock /// held so concurrent metadata reads aren't blocked behind a fetch. @@ -669,6 +834,86 @@ impl Core { nodes.into_iter().next().ok_or(Errno::ENOENT) } + /// An open [`RevisionReader`] for `uid`, reusing the cached one when it is + /// still valid for `(mtime, fsize)` and opening a fresh one otherwise. + /// + /// Opening resolves the file's link details, ancestor keys, node key (an S2K + /// unlock) and block table. Doing that once per file rather than once per + /// block is the whole point: a cold 100 MB read is 25 block misses, which + /// used to mean 25 full resolutions (50 API calls, 25 unlocks) and now means + /// one. + /// + /// Two threads racing on the same cold file may both open a reader; the + /// loser's is dropped. That costs one redundant open and is cheaper than + /// holding a lock across the network call. + async fn revision_reader( + &self, + uid: &NodeUid, + mtime: i64, + fsize: u64, + ) -> Result, Errno> { + // Fast path: a valid reader is already open. The guard is dropped before + // any await — network I/O must never run under it. + if let Some(hit) = { + let mut readers = self.readers.lock().unwrap(); + match readers.get_mut(uid) { + Some(entry) if entry.mtime == mtime && entry.size == fsize => { + entry.last_used = Instant::now(); + Some(entry.reader.clone()) + } + _ => None, + } + } { + return Ok(hit); + } + + let reader = Arc::new(self.client.open_revision(uid).await.map_err(|e| { + warn!(%uid, error = %e, "open_revision failed"); + Errno::EIO + })?); + + let mut readers = self.readers.lock().unwrap(); + // Someone may have opened one while we were on the network; either is + // equally fresh, so prefer theirs and let ours drop. + if let Some(entry) = readers.get_mut(uid) + && entry.mtime == mtime + && entry.size == fsize + { + entry.last_used = Instant::now(); + return Ok(entry.reader.clone()); + } + + readers.insert( + uid.clone(), + CachedReader { + reader: reader.clone(), + mtime, + size: fsize, + last_used: Instant::now(), + }, + ); + + // Bound the map: drop the least recently used until we are back at cap. + while readers.len() > MAX_OPEN_READERS { + let Some(victim) = readers + .iter() + .min_by_key(|(_, entry)| entry.last_used) + .map(|(uid, _)| uid.clone()) + else { + break; + }; + readers.remove(&victim); + } + + Ok(reader) + } + + /// Drop any open reader for `uid`, so the next read reopens against the + /// current revision. Called wherever cached content is evicted. + fn evict_reader(&self, uid: &NodeUid) { + self.readers.lock().unwrap().remove(uid); + } + /// Serve bytes `[offset, offset + len)` of `uid`'s active revision, hitting /// the on-disk caches before the network: a whole-file blob (pinned files) /// first, then the block cache — fetching only the [`BLOCK_SIZE`]-aligned @@ -681,6 +926,81 @@ impl Core { fsize: u64, offset: u64, len: u64, + ) -> Result, Errno> { + // A queued write has not reached the remote yet, so the remote's current + // revision is stale and the staged blob is the truth. Serve from it until + // the drain worker lands the upload (offline.md Phase 3). + if let Some(pending) = self.pending.lock().unwrap().get(uid).cloned() { + return self.read_pending(&pending, offset, len); + } + self.read_range_remote(uid, mtime, fsize, offset, len) + } + + /// Serve a read from a staged blob, falling back to the remote base for any + /// range the write did not author (an incomplete [`StagedWrite`] holds zeros + /// there, which must never be handed out as content). + fn read_pending( + &self, + pending: &PendingRevision, + offset: u64, + len: u64, + ) -> Result, Errno> { + let m = &pending.meta; + if offset >= m.len || len == 0 { + return Ok(Vec::new()); + } + let uid = parse_node_uid(&m.uid).ok_or(Errno::EIO)?; + let file = File::open(&pending.path).map_err(|e| { + error!(%uid, path = %pending.path.display(), error = %e, "open staged blob failed"); + Errno::EIO + })?; + let mut written = Intervals::default(); + for &(s, e) in &m.authored { + written.add(s, e); + } + // Same merge as `serve_open_read`, but resolving gaps against the remote + // rather than through `read_range` — going through `read_range` would find + // this very pending op and recurse. + let end = offset.saturating_add(len).min(m.len); + let mut out = Vec::with_capacity((end - offset) as usize); + for (s, e, authored) in written.segments(offset, end) { + if authored { + let mut buf = vec![0u8; (e - s) as usize]; + file.read_exact_at(&mut buf, s).map_err(|err| { + warn!(%uid, error = %err, "staged blob read failed"); + Errno::EIO + })?; + out.extend_from_slice(&buf); + continue; + } + let bend = e.min(m.base_size); + if s < bend { + out.extend_from_slice(&self.read_range_remote( + &uid, + m.base_mtime, + m.base_size, + s, + bend - s, + )?); + } + // Past the base: a hole the write extended over. + out.resize(out.len() + e.saturating_sub(s.max(m.base_size)) as usize, 0); + } + Ok(out) + } + + /// Read from the content cache, else the remote. The base-content path, with + /// no awareness of queued writes — callers wanting the file's *current* + /// content want [`Core::read_range`]. Gap-filling a staged blob is the one + /// caller that genuinely means "the base", since that is what its zeroed + /// ranges have to be filled from. + fn read_range_remote( + &self, + uid: &NodeUid, + mtime: i64, + fsize: u64, + offset: u64, + len: u64, ) -> Result, Errno> { if let Some(bytes) = self.cache.read_range(uid, mtime, fsize, offset, len) { return Ok(bytes); @@ -713,19 +1033,24 @@ impl Core { if !misses.is_empty() { let fetched = self.rt.block_on(async { + // Resolve the file's keys and block table once, then read every + // missing block through the shared reader. Previously each block + // called `download_range`, which redid that resolution per block. + let reader = self.revision_reader(uid, mtime, fsize).await?; + let mut set = tokio::task::JoinSet::new(); for &bidx in &misses { - let client = self.client.clone(); + let reader = reader.clone(); let uid = uid.clone(); let bstart = bidx * BLOCK_SIZE; let blen = BLOCK_SIZE.min(fsize - bstart); set.spawn(async move { - client - .download_range(&uid, bstart, blen) + reader + .read_at(bstart, blen) .await .map(|bytes| (bidx, bytes)) .map_err(|e| { - warn!(%uid, bstart, blen, error = %e, "download_range failed"); + warn!(%uid, bstart, blen, error = %e, "block read failed"); Errno::EIO }) }); @@ -804,32 +1129,23 @@ impl Core { Ok(out) } - /// Upload a write handle's scratch file as a new revision if it is dirty. - /// Untouched bytes within the base are filled from the remote first (reusing - /// the block cache), so a partial overwrite never had to pre-download the - /// whole file. On success the handle's base is advanced to the just-sealed - /// revision and `written` cleared, so later reads of untouched regions see - /// the new content. No-op for a clean (or unknown) handle. Network I/O runs - /// without the lock held. - fn commit(&self, fh: u64) -> Result<(), Errno> { - let (uid, file, path, len, base_mtime, base_size, written, ino) = { - let st = self.state.lock().unwrap(); - match st.handles.get(&fh) { - Some(h) if h.dirty => ( - h.uid.clone(), - h.file.clone(), - h.path.clone(), - h.len, - h.base_mtime, - h.base_size, - h.written.clone(), - h.ino, - ), - _ => return Ok(()), - } - }; - // Materialize the full new content in the scratch file: ensure its length, - // then fill every untouched range that overlaps the base with base bytes. + /// Fill every unauthored range of a scratch/staged file that overlaps its + /// base with the base's bytes, so the file becomes the complete new content. + /// + /// This is the step a partial overwrite cannot skip: only the authored bytes + /// were ever written to disk, and a revision upload sends the whole file. + /// The gaps come from the *remote base* (through the block cache, so a small + /// edit of a large file does not pull all of it), which is exactly why this + /// can fail with no network — see `StagedWrite`. + fn fill_gaps( + &self, + uid: &NodeUid, + file: &File, + len: u64, + base_mtime: i64, + base_size: u64, + written: &Intervals, + ) -> Result<(), Errno> { file.set_len(len).map_err(|e| { error!(%uid, error = %e, "resize scratch file failed"); Errno::EIO @@ -842,64 +1158,280 @@ impl Core { if s >= bend { continue; // wholly past the base: already zero-filled on disk } - let bytes = self.read_range(&uid, base_mtime, base_size, s, bend - s)?; + let bytes = self.read_range_remote(uid, base_mtime, base_size, s, bend - s)?; file.write_all_at(&bytes, s).map_err(|err| { error!(%uid, error = %err, "scratch gap-fill write failed"); Errno::EIO })?; } - // Stream the scratch file up as one revision (fresh handle reads from 0). - let reader = File::open(&path).map_err(|e| { - error!(%uid, error = %e, "reopen scratch file failed"); + Ok(()) + } + + /// Accept a released write handle's bytes and queue their upload + /// (offline.md Phase 3). + /// + /// This is what makes a copy into the mount run at disk speed: the caller's + /// `close` returns once the bytes are staged on local disk and the intent is + /// in `pending_op`, instead of waiting out a full upload inside the FUSE + /// handler. It is also what makes an offline write succeed rather than EIO — + /// the queued op simply waits for the network. + /// + /// The scratch file is *moved* into staging, never copied: it is the only + /// copy of what the user wrote. + fn queue_revision(&self, h: &WriteHandle) -> Result<(), Errno> { + if !h.dirty { + let _ = std::fs::remove_file(&h.path); + return Ok(()); + } + // Materialize the full content now if we can. A complete blob is + // uploadable without the network and, crucially, lets a later write to + // the same file supersede this one safely. + let filled = self + .fill_gaps( + &h.uid, + &h.file, + h.len, + h.base_mtime, + h.base_size, + &h.written, + ) + .is_ok(); + let authored: Vec<(u64, u64)> = if filled { + vec![(0, h.len)] + } else { + h.written + .segments(0, h.len) + .into_iter() + .filter(|&(_, _, authored)| authored) + .map(|(s, e, _)| (s, e)) + .collect() + }; + let meta = StagedWrite { + uid: h.uid.to_string(), + len: h.len, + base_size: h.base_size, + base_mtime: h.base_mtime, + complete: authored == [(0, h.len)], + authored, + }; + // An incomplete blob's gaps refer to the *remote* base. If an earlier + // write to this file is still queued, the remote no longer holds that + // base — the previous staged blob does — so superseding it would fill + // those gaps from the wrong revision. Rather than corrupt the file, + // refuse the write and keep the bytes recoverable (Phase 2 behaviour). + // Only reachable offline, editing in place a file whose previous edit + // has not drained and whose base is not cached. + if !meta.complete && self.pending.lock().unwrap().contains_key(&h.uid) { + self.stage_orphaned_write(h, &meta); + return Err(Errno::EIO); + } + let path = self.cache.stage_write(&meta, &h.path).map_err(|e| { + error!(uid = %h.uid, error = %e, "staging write failed"); Errno::EIO })?; - let name = { - let st = self.state.lock().unwrap(); - st.entries - .get(&ino) - .map(|e| e.node.name.clone()) - .unwrap_or_default() + let op = PendingOp { + id: 0, + kind: OP_REVISION.to_string(), + uid: h.uid.to_string(), + blob_path: Some(path.to_string_lossy().into_owned()), + meta_json: Some(serde_json::to_string(&meta).unwrap_or_default()), + created_at: now_millis(), + attempts: 0, + last_error: None, + next_attempt_at: 0, }; - let guard = self - .transfers - .begin(name, uid.to_string(), TransferDirection::Upload, len); - let reader = CountingReader::new(reader, &guard); - self.rt - .block_on(self.client.upload_new_revision_from( - &uid, - reader, - len as i64, - Vec::new(), - None, - )) - .map_err(|e| { - error!(%uid, error = %e, "upload new revision failed"); - Errno::EIO - })?; - drop(guard); + let (_id, superseded) = self.db.enqueue_op(&op).map_err(|e| { + error!(uid = %h.uid, error = %e, "queueing upload failed"); + Errno::EIO + })?; + if let Some(old) = superseded { + self.cache.discard_staged(Path::new(&old)); + } + self.pending + .lock() + .unwrap() + .insert(h.uid.clone(), PendingRevision { path, meta }); + // Reflect the write in the tree straight away: `ls` must show the new + // size and mtime even though the remote still holds the old revision. let now = now_secs(); { let mut st = self.state.lock().unwrap(); - if let Some(h) = st.handles.get_mut(&fh) { - h.dirty = false; - // The scratch file now equals the sealed revision; treat it all as - // base so further reads of untouched bytes hit the new content. - h.written = Intervals::default(); - h.base_mtime = now; - h.base_size = len; - } - st.set_size(ino, len); - st.touch_mtime(ino, now); + st.set_size(h.ino, h.len); + st.touch_mtime(h.ino, now); } - // The sealed content differs from any cached blob/blocks; refresh a pinned - // file's whole-file blob, otherwise evict so reads re-fetch fresh. - if self.cache.is_pinned(&uid) { - if let Ok(bytes) = std::fs::read(&path) { - let _ = self.cache.store(&uid, now, len, &bytes); + // Cached blobs and open readers describe the superseded revision. Reads + // come from the staged blob until the op drains, so just drop them. + self.cache.evict(&h.uid); + self.evict_reader(&h.uid); + self.wake_drain(); + debug!(uid = %h.uid, len = h.len, complete = filled, "queued revision upload"); + Ok(()) + } + + /// Nudge the drain worker to re-examine the queue now. + fn wake_drain(&self) { + let (lock, cv) = &*self.drain_wake; + *lock.lock().unwrap() = true; + cv.notify_all(); + } + + /// Keep a write we cannot safely queue, so the bytes are recoverable even + /// though the caller is getting an error. See [`Core::queue_revision`]. + fn stage_orphaned_write(&self, h: &WriteHandle, meta: &StagedWrite) { + match self.cache.stage_write(meta, &h.path) { + Ok(staged) => { + error!( + uid = %h.uid, + staged = %staged.display(), + "cannot queue write over an undrained edit; bytes kept in staging" + ); + let name = { + let st = self.state.lock().unwrap(); + st.entries + .get(&h.ino) + .map(|e| e.node.name.clone()) + .unwrap_or_default() + }; + self.log_activity( + ActivityKind::Upload, + &name, + format!("write not queued; changes kept at {}", staged.display()), + false, + ); + } + Err(e) => { + error!(uid = %h.uid, error = %e, "staging write failed; bytes lost"); + let _ = std::fs::remove_file(&h.path); } - } else { - self.cache.evict(&uid); } + } + + /// Drain the pending-op queue: the background half of every write + /// (offline.md Phase 3). + /// + /// Runs for the life of the mount. Ops are replayed oldest-first, each + /// retried with doubling backoff and *never* dropped on failure — the staged + /// blob is the only copy of the user's bytes, so a failed op stays queued + /// until it lands or the user deletes the file. + fn run_pending_drain(&self) { + loop { + let now = now_millis(); + let due = self + .db + .pending_ops() + .unwrap_or_default() + .into_iter() + .find(|op| op.next_attempt_at <= now); + + let Some(op) = due.filter(|_| self.online.load(Ordering::Relaxed)) else { + self.wait_for_drain_work(); + continue; + }; + if let Err(e) = self.drain_op(&op) { + // Backoff is per-op, so one wedged file cannot starve the rest: + // the next pass picks whichever op is due. + let attempts = op.attempts + 1; + let backoff = DRAIN_BACKOFF_MIN + .saturating_mul(1u32 << attempts.min(6)) + .min(DRAIN_BACKOFF_MAX); + warn!(uid = %op.uid, attempts, error = %e, "pending upload failed; will retry"); + let _ = self.db.record_op_failure( + op.id, + &e.to_string(), + now_millis() + backoff.as_millis() as i64, + ); + self.wait_for_drain_work(); + } + } + } + + /// Block until there is plausibly something to do: a new op, a reconnect, or + /// the shortest outstanding backoff elapsing. + fn wait_for_drain_work(&self) { + let (lock, cv) = &*self.drain_wake; + let mut woken = lock.lock().unwrap(); + if !*woken { + let (guard, _) = cv.wait_timeout(woken, DRAIN_IDLE_POLL).unwrap(); + woken = guard; + } + *woken = false; + } + + /// Perform one queued op and retire it. + fn drain_op(&self, op: &PendingOp) -> Result<(), Box> { + if op.kind != OP_REVISION { + return Err(format!("unknown pending op kind {:?}", op.kind).into()); + } + let blob = op + .blob_path + .clone() + .ok_or("pending op has no staged blob")?; + let blob = PathBuf::from(blob); + let meta: StagedWrite = serde_json::from_str(op.meta_json.as_deref().unwrap_or(""))?; + let uid = parse_node_uid(&meta.uid).ok_or("staged write has an unparseable uid")?; + + // An incomplete blob is authored bytes over zeros; the untouched ranges + // have to be filled from the base before it can be sent. This is the case + // the write could not resolve at release time (it was offline), and the + // reason it is safe to do now is that we are not. + if !meta.complete { + let file = std::fs::OpenOptions::new() + .read(true) + .write(true) + .open(&blob)?; + let mut written = Intervals::default(); + for &(s, e) in &meta.authored { + written.add(s, e); + } + self.fill_gaps( + &uid, + &file, + meta.len, + meta.base_mtime, + meta.base_size, + &written, + ) + .map_err(|e| format!("gap-fill from base failed: {e:?}"))?; + } + + let name = { + let st = self.state.lock().unwrap(); + st.by_uid + .get(&uid) + .and_then(|ino| st.entries.get(ino)) + .map(|e| e.node.name.clone()) + .unwrap_or_else(|| meta.uid.clone()) + }; + let guard = + self.transfers + .begin(&name, meta.uid.clone(), TransferDirection::Upload, meta.len); + let reader = CountingReader::new(File::open(&blob)?, &guard); + self.rt.block_on(self.client.upload_new_revision_from( + &uid, + reader, + meta.len as i64, + Vec::new(), + None, + ))?; + drop(guard); + + // Retire the op before dropping the blob: a crash between the two leaves + // an orphaned file (harmless), whereas the reverse would leave a queued + // op pointing at nothing. + self.db.delete_op(op.id)?; + self.pending.lock().unwrap().remove(&uid); + + // The staged blob now matches the sealed revision, so a pinned file keeps + // it as its cached content rather than re-downloading what we just sent. + if self.cache.is_pinned(&uid) + && let Ok(bytes) = std::fs::read(&blob) + { + let _ = self.cache.store(&uid, now_secs(), meta.len, &bytes); + } + self.cache.discard_staged(&blob); + self.evict_reader(&uid); + self.log_activity(ActivityKind::Upload, &name, "uploaded", true); + info!(%uid, len = meta.len, "pending upload landed"); Ok(()) } @@ -1081,6 +1613,7 @@ impl Core { for s in uids { if let Some(u) = parse_uid(&s) { self.cache.evict(&u); + self.evict_reader(&u); } } } @@ -1617,6 +2150,7 @@ impl Core { .map(|(_, n)| n) .unwrap_or_default(); self.cache.evict(&uid); + self.evict_reader(&uid); self.invalidate_parent_listing(rel); self.invalidate_trash(); Ok(name) @@ -1713,6 +2247,31 @@ impl Core { let _ = self.db.clear_state(TRASH_SYNCED_MS); } + /// Drop the persisted photos timeline's freshness stamp, so the next timeline + /// read fetches rather than serving what it already has. + fn invalidate_photos(&self) { + let _ = self.db.clear_state(PHOTOS_SYNCED_MS); + } + + /// Drop one folder's cached child listing (`rel` is mountpoint-relative), so + /// the next `ListDir`/`readdir` re-enumerates it from the server. Backs + /// [`CtlRequest::Refresh`] with a [`RefreshScope::Dir`] scope. + fn refresh_dir(&self, rel: &Path) -> Result<(), String> { + let (ino, uid) = self + .resolve_path(rel) + .map_err(|e| format!("resolve path: {e:?}"))?; + // Clear the DB flag directly rather than through + // `State::invalidate_listing`, which no-ops when the folder has no + // in-memory listing to drop — exactly the case for a folder hydrated + // from the DB but not read yet this run, which is precisely the stale + // listing worth refreshing. + self.db + .set_listed(&uid, false) + .map_err(|e| format!("db: {e:?}"))?; + self.state.lock().unwrap().children.remove(&ino); + Ok(()) + } + /// Parse wire uids (`volume~link`) into [`NodeUid`]s, rejecting the whole /// batch if any is malformed — a partial trash mutation is worse than none. fn parse_uids(uids: &[String]) -> Result, String> { @@ -1790,6 +2349,7 @@ impl Core { drop(st); for uid in uids { self.cache.evict(uid); + self.evict_reader(uid); } } @@ -2068,19 +2628,28 @@ impl Core { // ---- devices ---------------------------------------------------------- - /// List the account's registered devices. + /// 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. fn list_devices(&self) -> Result, String> { let devices = self .rt .block_on(self.client.enumerate_devices()) .map_err(|e| format!("list devices: {e}"))?; + // No cached device row yet means this machine syncs nothing, so none of + // the listed devices is ours. + let this_uid = self.db.device_get().ok().flatten().map(|d| d.uid); Ok(devices .into_iter() - .map(|d| DeviceInfo { - uid: d.uid.to_string(), - name: d.name.unwrap_or_else(|_| "(unnamed device)".to_string()), - device_type: device_type_str(d.device_type).to_string(), - last_sync: d.last_sync_time, + .map(|d| { + let uid = d.uid.to_string(); + DeviceInfo { + this_device: this_uid.as_deref() == Some(uid.as_str()), + uid, + name: d.name.unwrap_or_else(|_| "(unnamed device)".to_string()), + device_type: device_type_str(d.device_type).to_string(), + last_sync: d.last_sync_time, + } }) .collect()) } @@ -2316,13 +2885,19 @@ impl Core { .map(|n| n.to_string_lossy().into_owned()) .unwrap_or_else(|| "folder".to_string()); jobs.push(match p.phase { - // No counts exist until the walk has classified something, so a - // scan reports indeterminate rather than a fake 0 of 0. + // The scan's total is the last pass's baseline, so a folder that + // has never synced still reports indeterminate (`total: 0`) — but + // every later pass has a real bar. A grown folder can push `done` + // past the estimate; clamp so the row never reads "600 of 500". SyncPhase::Scanning => JobItem { title: format!("Checking {folder}"), detail: "Looking for changes".to_string(), - done: 0, - total: 0, + done: p.done as u64, + total: if p.total == 0 { + 0 + } else { + p.total.max(p.done) as u64 + }, }, SyncPhase::Applying => JobItem { title: format!("Syncing {folder}"), @@ -2353,6 +2928,18 @@ impl Core { .sync_folder_get(id) .map_err(|e| format!("db: {e:?}"))? .ok_or_else(|| format!("no synced folder with id {id}"))?; + // An `ondemand` folder *is* a FUSE mount over its local path, so dropping + // only the row would strand the mount: the path would keep serving a + // folder the daemon no longer tracks, and nothing would ever unmount it. + // Tear it down first — including before trashing the remote tree it + // serves, which would otherwise leave it answering for deleted nodes. + if let Some(session) = self.mounts.lock().unwrap().remove(&id) { + if let Err(e) = session.umount_and_join() { + 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])) @@ -2398,19 +2985,116 @@ impl Core { fork } + /// Ask for a synced folder to move to `mode`, applying it now if the folder is + /// free and safe to switch, and **queueing** it otherwise. Returns the human + /// message for the reply. + /// + /// Queueing rather than rejecting is what makes the toggle usable: a folder + /// that syncs continuously (a busy Downloads folder) is almost never caught in + /// the narrow window where it is both unlocked and `idle`, so a `try_lock` + /// rejection asks the user to keep retrying until they get lucky. Instead the + /// intent is recorded and the engine applies it at the end of the pass already + /// running — which, seeing a queued `ondemand`, also stops pulling down files + /// it is about to evict ([`Core::push_pass`]). + pub(crate) fn request_sync_folder_mode(&self, id: i64, mode: &str) -> Result { + if mode != "mirror" && mode != "ondemand" { + return Err(format!("unknown mode {mode:?} (want mirror|ondemand)")); + } + let folder = self + .db + .sync_folder_get(id) + .map_err(|e| format!("db: {e:?}"))? + .ok_or_else(|| 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| format!("db: {e:?}"))?; + let _ = self.sync_tx.send(sync::SyncMsg::Rewatch); + return Ok(format!("staying {mode}")); + } + return Ok(format!("already {mode}")); + } + + match self.apply_sync_folder_mode(id, mode) { + 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)) + .map_err(|e| format!("db: {e:?}"))?; + let _ = self.sync_tx.send(sync::SyncMsg::Reconcile(id)); + Ok(match mode { + "ondemand" => format!( + "{} will go on-demand once its local changes are uploaded", + base_name(&folder.local_path) + ), + _ => format!( + "{} will start mirroring once the current sync finishes", + base_name(&folder.local_path) + ), + }) + } + Err(SwitchBlocked::Failed(e)) => Err(e), + } + } + + /// Apply a queued mode switch if the folder has one and is now able to take it. + /// Called by the sync engine after every pass, so a switch the user asked for + /// mid-sync lands as soon as the pass that blocked it is done. A folder that is + /// still not ready (its push failed, so the local copy is not safe to evict) + /// keeps its `pending_mode` and is retried after the next pass. + pub(crate) fn settle_pending_mode(&self, id: i64) { + let Ok(Some(folder)) = self.db.sync_folder_get(id) else { + return; + }; + let Some(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) { + Ok(message) => { + info!(id, mode, "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"); + } + Err(SwitchBlocked::Failed(e)) => { + warn!(id, mode, error = %e, "queued mode switch failed; withdrawing"); + let _ = self.db.sync_folder_set_pending_mode(id, None); + self.log_activity( + ActivityKind::Sync, + format!("couldn't switch {}", base_name(&folder.local_path)), + e, + false, + ); + } + } + } + /// Flip a synced folder between `mirror` (full local copy + two-way sync) and /// `ondemand` (a FUSE mount over the local path; no local storage). Returns a - /// human message for the reply. + /// human message on success. /// /// - **mirror→ondemand**: require the folder in-sync (`idle`), stop watching it, /// evict the local files to reclaim disk, then mount a secondary `ProtonFs` /// rooted at the folder's remote node over its local path. /// - **ondemand→mirror**: unmount, clear the stale baseline (the local tree was /// evicted), then hand the folder back to the engine, which re-downloads it. - fn set_sync_folder_mode(&self, id: i64, mode: &str) -> Result { - if mode != "mirror" && mode != "ondemand" { - return Err(format!("unknown mode {mode:?} (want mirror|ondemand)")); - } + /// + /// [`SwitchBlocked::NotNow`] means "not yet, try after a pass" and is never an + /// error the user needs to see — callers queue on it. + fn apply_sync_folder_mode(&self, id: i64, mode: &str) -> Result { // 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` @@ -2419,15 +3103,15 @@ impl Core { // inserting the row and the engine picking it up. let lock = self.sync_lock(id); let Ok(_guard) = lock.try_lock() else { - return Err("folder is syncing right now — wait for it to finish".to_string()); + return Err(SwitchBlocked::NotNow); }; // Re-read under the lock: a pass that finished while we waited may have // changed `state`. let folder = self .db .sync_folder_get(id) - .map_err(|e| format!("db: {e:?}"))? - .ok_or_else(|| format!("no synced folder with id {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}")); } @@ -2437,28 +3121,30 @@ impl Core { "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. + // the local copy. Not an error: a pass makes this true, and the queued + // request is applied once one does. if folder.state != "idle" { - return Err(format!( - "folder is '{}', not idle — sync it before switching to on-demand", - folder.state - )); + return Err(SwitchBlocked::NotNow); } - let root_uid = parse_uid(&folder.remote_uid) - .ok_or_else(|| format!("bad remote uid: {}", folder.remote_uid))?; + let root_uid = parse_uid(&folder.remote_uid).ok_or_else(|| { + SwitchBlocked::Failed(format!("bad remote uid: {}", folder.remote_uid)) + })?; let root = self .rt .block_on(self.client.enumerate_nodes(std::slice::from_ref(&root_uid))) - .map_err(|e| format!("fetch remote root: {e}"))? + .map_err(|e| SwitchBlocked::Failed(format!("fetch remote root: {e}")))? .into_iter() .next() - .ok_or_else(|| "remote folder not found".to_string())?; + .ok_or_else(|| SwitchBlocked::Failed("remote folder not found".to_string()))?; // Reclaim the disk: empty the local dir (keep it as the mountpoint). - evict_dir_contents(&local) - .map_err(|e| format!("evict {}: {e}", local.display()))?; + evict_dir_contents(&local).map_err(|e| { + SwitchBlocked::Failed(format!("evict {}: {e}", local.display())) + })?; - let session = self.spawn_ondemand_mount(&local, root)?; + let session = self + .spawn_ondemand_mount(&local, root) + .map_err(SwitchBlocked::Failed)?; self.mounts.lock().unwrap().insert(id, session); // Persist the mode only now that the mount is actually up. Writing it // first would strand the folder on any failure below: the engine skips @@ -2468,7 +3154,7 @@ impl Core { // re-downloads whatever eviction removed. self.db .sync_folder_set_mode(id, "ondemand") - .map_err(|e| format!("db: {e:?}"))?; + .map_err(|e| SwitchBlocked::Failed(format!("db: {e:?}")))?; 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"); @@ -2485,10 +3171,10 @@ impl Core { // deleted locally"); clear it so the reconcile is a pure download. self.db .sync_entries_clear(id) - .map_err(|e| format!("db: {e:?}"))?; + .map_err(|e| SwitchBlocked::Failed(format!("db: {e:?}")))?; self.db .sync_folder_set_mode(id, "mirror") - .map_err(|e| format!("db: {e:?}"))?; + .map_err(|e| SwitchBlocked::Failed(format!("db: {e:?}")))?; let _ = self.sync_tx.send(sync::SyncMsg::Rewatch); let _ = self.sync_tx.send(sync::SyncMsg::Reconcile(id)); info!(id, path = %local.display(), "restored sync folder to mirror"); @@ -3060,11 +3746,35 @@ impl Core { } } + /// Set the number of items the scan expects to check, from the size of the + /// last pass's baseline. Only an estimate — the folder may have grown since — + /// but it turns the scan from an indeterminate pulse into a bar that moves, + /// which is the difference between "it's stuck" and "it's working" on a folder + /// whose walk takes minutes. + pub(crate) fn progress_scan_total(&self, folder_id: i64, n: usize) { + self.progress_update(folder_id, |p| p.total = n); + } + + /// Note that the scan has checked one more item, named `name`. + pub(crate) fn progress_scanned(&self, folder_id: i64, name: &str) { + self.progress_update(folder_id, |p| { + p.done += 1; + p.current = name.to_string(); + }); + } + /// Note that `n` more items have been queued for this pass, and that it has - /// moved on from scanning to applying the diff. + /// moved on from scanning to applying the diff. The scan's counts are dropped: + /// they measured a different quantity (items checked, not items to apply), so + /// carrying them over would start the applying bar at a meaningless fraction. pub(crate) fn progress_queued(&self, folder_id: i64, n: usize) { self.progress_update(folder_id, |p| { - p.phase = SyncPhase::Applying; + if p.phase == SyncPhase::Scanning { + p.phase = SyncPhase::Applying; + p.done = 0; + p.total = 0; + p.current.clear(); + } p.total += n; }); } @@ -3218,6 +3928,7 @@ impl ProtonFs { } self.core.state.lock().unwrap().forget(&uid); self.core.cache.evict(&uid); + self.core.evict_reader(&uid); self.core.invalidate_trash(); reply.ok(); } @@ -3231,6 +3942,23 @@ fn now_secs() -> i64 { .unwrap_or(0) } +/// Current wall-clock time as epoch milliseconds, the resolution `pending_op` +/// timestamps and backoff deadlines are kept in. +fn now_millis() -> i64 { + SystemTime::now() + .duration_since(UNIX_EPOCH) + .map(|d| d.as_millis() as i64) + .unwrap_or(0) +} + +/// Parse a [`NodeUid`] back from its `Display` form (`volume~link`), which is +/// how one is persisted in `pending_op.uid` and a [`StagedWrite`] sidecar. The +/// SDK has no `FromStr`, and neither id contains a `~`. +fn parse_node_uid(s: &str) -> Option { + let (vol, link) = s.split_once('~')?; + Some(NodeUid::new(VolumeId::from(vol), LinkId::from(link))) +} + /// This machine's hostname, used to name (and later recover) its Proton Drive /// Device. Reads the live kernel hostname, falling back to a generic label. fn this_hostname() -> String { @@ -3269,6 +3997,7 @@ fn sync_folder_info(f: StoredSyncFolder, progress: Option) -> Sync local_path: f.local_path, remote_uid: f.remote_uid, mode: f.mode, + pending_mode: f.pending_mode, state: f.state, last_sync: f.last_sync, progress, @@ -3853,6 +4582,7 @@ impl Filesystem for ProtonFs { } else { self.core.cache.evict(&uid); } + self.core.evict_reader(&uid); } self.core.state.lock().unwrap().set_size(ino.0, size); } @@ -3866,20 +4596,24 @@ impl Filesystem for ProtonFs { } } + /// `close(2)` calls flush before release. The upload is queued in `release` + /// and performed in the background (offline.md Phase 3), so there is nothing + /// to push here — the written bytes are already in the scratch file. fn flush( &self, _req: &Request, _ino: INodeNo, - fh: FileHandle, + _fh: FileHandle, _lock_owner: LockOwner, reply: ReplyEmpty, ) { - match self.core.commit(fh.0) { - Ok(()) => reply.ok(), - Err(e) => reply.error(e), - } + reply.ok(); } + /// Durability here means the scratch file, not the remote: a caller asking + /// for fsync wants its bytes to survive a crash, and blocking it on an upload + /// (which offline would never finish) buys nothing the queue does not already + /// guarantee. fn fsync( &self, _req: &Request, @@ -3888,9 +4622,23 @@ impl Filesystem for ProtonFs { _datasync: bool, reply: ReplyEmpty, ) { - match self.core.commit(fh.0) { - Ok(()) => reply.ok(), - Err(e) => reply.error(e), + let file = self + .core + .state + .lock() + .unwrap() + .handles + .get(&fh.0) + .map(|h| h.file.clone()); + match file { + Some(f) => match f.sync_all() { + Ok(()) => reply.ok(), + Err(e) => { + error!(fh = fh.0, error = %e, "fsync of scratch file failed"); + reply.error(Errno::EIO); + } + }, + None => reply.error(Errno::EBADF), } } @@ -3904,14 +4652,17 @@ impl Filesystem for ProtonFs { _flush: bool, reply: ReplyEmpty, ) { - let res = self.core.commit(fh.0); let handle = self.core.state.lock().unwrap().handles.remove(&fh.0); - if let Some(h) = handle { - let _ = std::fs::remove_file(&h.path); - } - match res { - Ok(()) => reply.ok(), - Err(e) => reply.error(e), + // Hand the bytes to the queue rather than uploading them here: the + // scratch file is the only copy of what was just written, and blocking + // the caller on the network is what made a copy into the mount run at + // upload speed (and fail outright offline). + match handle { + Some(h) => match self.core.queue_revision(&h) { + Ok(()) => reply.ok(), + Err(e) => reply.error(e), + }, + None => reply.ok(), } } @@ -4126,6 +4877,7 @@ impl Filesystem for ProtonFs { fn apply_event( state: &Mutex, content: &ContentCache, + pending: &Mutex>, notifier: &Notifier, event: &DriveEvent, ) { @@ -4152,6 +4904,14 @@ fn apply_event( } } } + } else if pending.lock().unwrap().contains_key(node_uid) { + // A node we owe an upload for is *ahead* of the remote, not + // behind it: this event is almost always the echo of our own + // empty-file create, and re-fetching would replace the size and + // mtime of the write we just accepted with the stale revision's + // — making a file that was copied in seconds ago read as empty + // until its upload lands (offline.md Phase 3). + debug!(uid = %node_uid, "ignoring remote event for a node with a queued write"); } else if let Some(&ino) = st.by_uid.get(node_uid) { // Known node changed: drop its cached attrs/data (and listing if // it is a directory) so the next access re-fetches. Its content @@ -4214,6 +4974,7 @@ async fn run_event_sync( state: Arc>, content: Arc, db: Arc, + pending: Arc>>, notifier: Notifier, ) { let mut cursor: Option = match db.get_event_cursor() { @@ -4222,21 +4983,30 @@ async fn run_event_sync( // First mount: a `None` cursor yields a single `CursorAdvanced` at the // server head; persist it so the next restart resumes instead of // reseeding (which would skip everything that changed offline). - Ok(None) => match client.enumerate_events(&scope, None).await { - Ok(events) => { - let head = events.last().map(|e| e.id().clone()); - if let Some(c) = &head - && let Err(e) = db.set_event_cursor(c.as_str()) - { - warn!(error = %e, "persist seed cursor failed"); + // Seeding needs the network, and this task also runs on mounts that + // started offline (offline.md Phase 1) — so retry rather than giving up, + // which used to disable live sync for the life of the daemon. + Ok(None) => { + let mut delay = ONLINE_PROBE_MIN; + loop { + match client.enumerate_events(&scope, None).await { + Ok(events) => { + let head = events.last().map(|e| e.id().clone()); + if let Some(c) = &head + && let Err(e) = db.set_event_cursor(c.as_str()) + { + warn!(error = %e, "persist seed cursor failed"); + } + break head; + } + Err(e) => { + warn!(error = %e, ?delay, "seed event cursor failed; retrying"); + tokio::time::sleep(delay).await; + delay = (delay * 2).min(ONLINE_PROBE_MAX); + } } - head } - Err(e) => { - error!(error = %e, "failed to seed event cursor; live sync disabled"); - return; - } - }, + } Err(e) => { error!(error = %e, "read persisted cursor failed; live sync disabled"); return; @@ -4258,7 +5028,7 @@ async fn run_event_sync( } debug!(count = events.len(), "applying remote events"); for event in &events { - apply_event(&state, &content, ¬ifier, event); + apply_event(&state, &content, &pending, ¬ifier, event); } cursor = events.last().map(|e| e.id().clone()); if let Some(c) = &cursor @@ -4393,6 +5163,8 @@ fn handle_control_conn(core: &Core, username: &str, mountpoint: &Path, stream: U used: core.cache.usage(), budget: core.cache.budget(), pins, + online: core.online.load(Ordering::Relaxed), + pending_uploads: core.db.pending_op_count().unwrap_or(0).max(0) as u64, } } Ok(CtlRequest::Pin { path }) => match rel_to_mount(mountpoint, &path) { @@ -4423,6 +5195,27 @@ fn handle_control_conn(core: &Core, username: &str, mountpoint: &Path, stream: U }, Err(e) => CtlResponse::Error { message: e }, }, + Ok(CtlRequest::Refresh { scope }) => { + let result = match &scope { + RefreshScope::Dir { path } => { + rel_to_mount(mountpoint, path).and_then(|rel| core.refresh_dir(&rel)) + } + RefreshScope::Trash => { + core.invalidate_trash(); + Ok(()) + } + RefreshScope::Photos => { + core.invalidate_photos(); + Ok(()) + } + }; + match result { + Ok(()) => CtlResponse::Ok { + message: "refreshed".to_string(), + }, + Err(e) => CtlResponse::Error { message: e }, + } + } Ok(CtlRequest::PhotosTimeline { offset, limit }) => { match core.photos_timeline(offset, limit) { Ok(Some(items)) => CtlResponse::Photos { @@ -4791,7 +5584,7 @@ fn handle_control_conn(core: &Core, username: &str, mountpoint: &Path, stream: U } } Ok(CtlRequest::SetSyncFolderMode { id, mode }) => { - match core.set_sync_folder_mode(id, &mode) { + match core.request_sync_folder_mode(id, &mode) { Ok(message) => { core.log_activity(ActivityKind::Upload, &message, "", true); CtlResponse::Ok { message } @@ -5034,11 +5827,54 @@ fn clear_stale_mount(mountpoint: &Path) { } } +/// `sync_state` key holding the uid of the My Files root, so a later run can +/// recover it from `nodes` without the API (offline.md Phase 1). +const ROOT_UID_KEY: &str = "root_uid"; + +/// The My Files root, and whether we got it from the API (`true`) or from the +/// cache because the API was unreachable (`false`). +/// +/// A successful fetch also records the root's uid, which is what makes the +/// fallback possible on a later run: `nodes` is keyed by uid, so without this we +/// would have the root's row on disk and no way to tell which one it is. +/// +/// Failing to fetch is only fatal on a genuinely cold start — no cached root +/// means an empty tree, and mounting that would show the user an empty Drive +/// rather than an honest error. +fn fetch_or_recall_root( + client: &ProtonDriveClient, + rt: &tokio::runtime::Handle, + db: &Db, +) -> std::io::Result<(Node, bool)> { + let err = match rt.block_on(client.get_my_files_folder()) { + Ok(root) => { + if let Err(e) = db.set_state_str(ROOT_UID_KEY, &root.uid.to_string()) { + warn!(error = %e, "persist root uid failed"); + } + return Ok((root, true)); + } + Err(e) => e, + }; + let cached = db + .state_str(ROOT_UID_KEY) + .ok() + .flatten() + .and_then(|uid| db.node_by_uid(&uid).ok().flatten()); + match cached { + Some(root) => { + warn!(error = %err, "fetch My Files root failed; mounting from cache (offline)"); + Ok((root, false)) + } + None => Err(std::io::Error::other(format!("fetch My Files root: {err}"))), + } +} + /// Mount the filesystem at `mountpoint` and block until it is unmounted or the /// daemon is asked to stop. /// -/// Fetches the My Files root up front (so an auth/network failure surfaces -/// before the kernel mount), then spawns the Phase 2 event-sync task, the +/// Resolves the My Files root up front — from the API, or from the cached tree +/// when the API is unreachable ([`fetch_or_recall_root`]) — then spawns the +/// Phase 2 event-sync task, the /// Phase 4 control socket, and runs the FUSE session on its own thread while /// this thread waits for either a stop signal (SIGTERM/SIGINT) or the kernel /// mount ending on its own. On a stop signal we lazily unmount ourselves @@ -5054,9 +5890,7 @@ pub fn mount( db: Arc, username: String, ) -> std::io::Result { - let root = rt - .block_on(client.get_my_files_folder()) - .map_err(|e| std::io::Error::other(format!("fetch My Files root: {e}")))?; + let (root, online) = fetch_or_recall_root(&client, &rt, &db)?; let scope = root.tree_event_scope_id(); // The folder-sync engine (devices.md Phase 2) runs on its own thread and is @@ -5077,7 +5911,11 @@ pub fn mount( db: db.clone(), })), cache: Arc::new(cache), + readers: Arc::new(Mutex::new(HashMap::new())), db, + online: Arc::new(AtomicBool::new(online)), + pending: Arc::new(Mutex::new(HashMap::new())), + drain_wake: Arc::new((Mutex::new(false), Condvar::new())), timeline_refreshing: Arc::new(AtomicBool::new(false)), trash_refreshing: Arc::new(AtomicBool::new(false)), thumb_gen: Arc::new(Mutex::new(HashSet::new())), @@ -5089,10 +5927,29 @@ pub fn mount( sync_locks: Arc::new(Mutex::new(HashMap::new())), }; + // 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. + core.hydrate_pending(); + { + let core = core.clone(); + std::thread::Builder::new() + .name("pdfs-drain".into()) + .spawn(move || core.run_pending_drain())?; + } + // Start the folder-sync engine. It watches every mirror folder, polls the // remotes, and reconciles on its own thread — never in front of a FUSE call. sync::spawn(core.clone(), sync_rx); + // Mounted from the cache: watch for the network coming back so the mount can + // stop being read-only-ish without the user restarting the daemon. + if !online { + let core = core.clone(); + std::thread::Builder::new() + .name("pdfs-online-probe".into()) + .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. @@ -5151,7 +6008,13 @@ pub fn mount( let bg = Session::new(fs, mountpoint, &config)?.spawn()?; let notifier = bg.notifier(); rt.spawn(run_event_sync( - client, scope, core.state, core.cache, core.db, notifier, + client, + scope, + core.state, + core.cache, + core.db, + core.pending, + notifier, )); // Stop signals (SIGTERM from `systemctl --user stop`, SIGINT from Ctrl-C) diff --git a/crates/pdfs-fuse/src/sync.rs b/crates/pdfs-fuse/src/sync.rs index 0d880da..bdb90d9 100644 --- a/crates/pdfs-fuse/src/sync.rs +++ b/crates/pdfs-fuse/src/sync.rs @@ -327,6 +327,9 @@ enum Pending { mtime: i64, size: i64, }, + /// Both sides changed during a [`push pass`](Core::push_pass): upload the local + /// copy into `parent` under a conflict name, leaving the remote file as it is. + PushConflict { rel: String, parent: NodeUid }, } impl Pending { @@ -337,7 +340,8 @@ impl Pending { | Pending::UploadNew { rel, .. } | Pending::UploadRevision { rel, .. } | Pending::Download { rel, .. } - | Pending::Conflict { rel, .. } => rel, + | Pending::Conflict { rel, .. } + | Pending::PushConflict { rel, .. } => rel, } } @@ -345,7 +349,9 @@ impl Pending { fn kind(&self) -> ActivityKind { match self { Pending::CreateDir { .. } => ActivityKind::CreateFolder, - Pending::UploadNew { .. } | Pending::UploadRevision { .. } => ActivityKind::Upload, + Pending::UploadNew { .. } + | Pending::UploadRevision { .. } + | Pending::PushConflict { .. } => ActivityKind::Upload, Pending::Download { .. } | Pending::Conflict { .. } => ActivityKind::Download, } } @@ -358,10 +364,28 @@ impl Pending { Pending::UploadRevision { .. } => "new version", Pending::Download { .. } => "from Drive", Pending::Conflict { .. } => "local changes kept as a conflict copy", + Pending::PushConflict { .. } => "local changes uploaded as a conflict copy", } } } +/// Why a full reconcile pass stopped without a result. +enum PassAbort { + /// An on-demand switch was queued while the pass was running. Everything left + /// to do is work on a local copy about to be evicted, so the pass gives up its + /// remaining work to a [`push pass`](Core::push_pass) instead. Not a failure — + /// nothing is left half-applied, because each step is applied whole. + Interrupted, + /// The pass could not establish its diff (a walk or the baseline load failed). + Failed(String), +} + +impl From for PassAbort { + fn from(e: String) -> Self { + PassAbort::Failed(e) + } +} + /// What a [`Pending`] op did, so the reconcile can update shared state on the /// engine thread (never inside a task). enum Applied { @@ -376,27 +400,42 @@ enum Applied { } impl Core { - /// Reconcile one synced folder, updating its `state` column. `ondemand` - /// folders are live FUSE mounts, not mirrored, so they are skipped here. + /// Reconcile one synced folder, then apply any mode switch queued while it was + /// busy. Both halves live here because the switch has to happen with the pass + /// finished and its lock released, and the engine thread is the only place that + /// is guaranteed — a switch requested over the control socket cannot wait for a + /// pass that may run for minutes. pub(crate) fn reconcile_folder(&self, folder: &StoredSyncFolder) { - if folder.mode != "mirror" { - return; + // `ondemand` folders are live FUSE mounts, not mirrored, so there is nothing + // to reconcile — but one may still carry a queued switch back to `mirror`, + // so the settle below is not skipped with it. + if folder.mode == "mirror" { + self.reconcile_pass(folder); } + self.settle_pending_mode(folder.id); + } + + /// One reconcile pass over a mirror folder, updating its `state` column. + fn reconcile_pass(&self, folder: &StoredSyncFolder) { // Hold the folder's lock for the whole pass so a mode switch can't evict the // local tree (and mount FUSE over it) while we walk and upload it. let lock = self.sync_lock(folder.id); let _guard = lock.lock().unwrap(); // `folder` was read before the lock; a switch may have landed in between, so // re-read and re-check the mode rather than trusting the snapshot. - match self.db.sync_folder_get(folder.id) { - Ok(Some(current)) if current.mode == "mirror" => {} + let current = match self.db.sync_folder_get(folder.id) { + Ok(Some(current)) if current.mode == "mirror" => current, Ok(Some(_)) => return, Ok(None) => return, Err(e) => { warn!(id = folder.id, error = ?e, "sync: cannot re-read folder; skipping"); return; } - } + }; + // A folder waiting to go on-demand is about to have its local copy deleted, + // so pulling the remote side down is work whose only result is more to evict. + // All that pass owes the user is getting local changes up. + let mut push_only = current.pending_mode.as_deref() == Some("ondemand"); let Some(remote_root) = parse_uid(&folder.remote_uid) else { warn!(id = folder.id, "sync: bad remote uid; skipping"); return; @@ -415,16 +454,42 @@ impl Core { .sync_folder_set_state(folder.id, "syncing", folder.last_sync); let name = base_name(&folder.local_path); self.progress_begin(folder.id); - let result = self.do_reconcile(folder.id, &local_root, &remote_root); + let result = if push_only { + self.push_pass(folder.id, &local_root, &remote_root) + } else { + match self.do_reconcile(folder.id, &local_root, &remote_root) { + Ok(outcome) => Ok(outcome), + Err(PassAbort::Failed(e)) => Err(e), + // The user asked for on-demand while this pass was running. Rather + // than make them wait out a walk-and-download whose results are + // about to be deleted, drop it here and do the only part that still + // matters — getting local changes up — so the switch lands now + // instead of after the pass and another poll. + Err(PassAbort::Interrupted) => { + info!( + id = folder.id, + "sync: on-demand queued mid-pass; pushing instead" + ); + push_only = true; + self.progress_begin(folder.id); + self.push_pass(folder.id, &local_root, &remote_root) + } + } + }; self.progress_end(folder.id); match result { Ok(outcome) => { // A folder only reaches `idle` when every path applied cleanly — // an un-uploaded file must keep it out of `idle` so it can't be // switched to on-demand (which evicts the local copy). + // + // A push pass is the exception on conflicts: it resolves them by + // uploading the local copy under a conflict name rather than leaving + // one on disk, so nothing local is left needing attention. Parking it + // in `conflict` would block the very switch the pass was run for. let state = if outcome.errors > 0 { "error" - } else if outcome.conflicts > 0 { + } else if outcome.conflicts > 0 && !push_only { "conflict" } else { "idle" @@ -478,6 +543,283 @@ impl Core { } } + /// Whether a switch to on-demand is waiting on this folder — i.e. whether a + /// pass in flight is still doing work worth doing. Read from the db rather than + /// held in memory because the request arrives on the control-socket thread while + /// the pass runs on the engine thread, and the db row is already the one place + /// both agree on. + fn ondemand_queued(&self, folder_id: i64) -> bool { + matches!( + self.db.sync_folder_get(folder_id), + Ok(Some(f)) if f.pending_mode.as_deref() == Some("ondemand") + ) + } + + /// Get every local change up to Drive, and nothing else — the pass run for a + /// folder waiting to go on-demand, whose local copy is about to be evicted. + /// + /// This exists because a full [`do_reconcile`](Self::do_reconcile) is both too + /// slow and too wasteful to stand between the user and the switch they asked + /// for. It walks the whole remote tree (the minutes-long part of a pass over a + /// large folder) to answer a question the switch makes irrelevant — what the + /// remote side has that we don't — and then downloads those files onto a disk + /// we are about to clear. A push pass answers only "is anything here not on + /// Drive yet?", which the local tree and the baseline already know, so it needs + /// no remote walk at all and finishes in the time the local walk takes. + /// + /// The remote is consulted for exactly one thing: the files whose local copy + /// changed. Those, and only those, could be a two-sided edit, so their nodes are + /// fetched (cheaply, in one batch) to check whether the remote moved too. If it + /// did, the local copy goes up under a conflict name rather than overwriting the + /// other side. Everything else — remote edits, remote deletions, remote-only + /// files — is left for the FUSE mount to show live once the switch lands. + /// + /// Per-path failures are counted, never fatal, exactly as in `do_reconcile`; a + /// non-zero [`Outcome::errors`] leaves the folder out of `idle`, which keeps the + /// queued switch waiting rather than evicting a file that never made it up. + fn push_pass( + &self, + folder_id: i64, + local_root: &Path, + remote_root: &NodeUid, + ) -> Result { + let baseline = self + .db + .sync_entries(folder_id) + .map_err(|e| format!("load baseline: {e:?}"))?; + // Only the local side is walked here, so each baseline path is checked once. + self.progress_scan_total(folder_id, baseline.len()); + + let mut local: HashMap = HashMap::new(); + self.walk_local(folder_id, local_root, local_root, &mut local)?; + + // The remote folder uids come from the baseline instead of a walk: every + // directory this device has synced recorded its uid there when it was + // created or first matched up. + let mut remote_dirs: HashMap = HashMap::new(); + remote_dirs.insert(String::new(), remote_root.clone()); + for (rel, item) in &local { + if !item.is_dir { + continue; + } + if let Some(uid) = baseline + .get(rel) + .and_then(|e| e.remote_uid.as_deref()) + .and_then(parse_uid) + { + remote_dirs.insert(rel.clone(), uid); + } + } + // A local directory the baseline has no uid for is either brand new or one + // whose first pass never finished — and those look identical from here. + // Creating it blindly would duplicate a remote folder we simply haven't + // looked at, so when any exist, the remote folder tree is walked for real. + // It costs one light listing per folder and no file keys, and a folder whose + // baseline is complete — the case this pass is written for — skips it + // entirely and touches the network only for the files it uploads. + if local + .iter() + .any(|(rel, item)| item.is_dir && !remote_dirs.contains_key(rel)) + { + self.walk_remote_dirs(folder_id, remote_root, "", &mut remote_dirs)?; + } + + // Files whose local copy has moved since the baseline: the only ones with + // anything to push, and the only ones that could be a two-sided edit. + let changed: Vec<&String> = local + .iter() + .filter(|(_, item)| !item.is_dir) + .filter(|(rel, item)| { + baseline + .get(*rel) + .is_none_or(|b| b.local_mtime != item.mtime || b.local_size != item.size) + }) + .map(|(rel, _)| rel) + .collect(); + + // One light batch tells us which of those also changed remotely. A light + // node carries the modification time without unlocking the node key, and + // mtime alone is what the baseline compares against. + let check: Vec = changed + .iter() + .filter_map(|rel| baseline.get(*rel)) + .filter_map(|e| e.remote_uid.as_deref()) + .filter_map(parse_uid) + .collect(); + let remote_mtimes: HashMap = if check.is_empty() { + HashMap::new() + } else { + self.rt + .block_on(self.client.enumerate_nodes_light(&check)) + .map_err(|e| format!("resolve changed nodes: {e}"))? + .into_iter() + .filter(|n| !n.trashed) + .map(|n| (n.uid.to_string(), n.modification_time)) + .collect() + }; + let changed: HashSet<&String> = changed.into_iter().collect(); + + let mut outcome = Outcome::default(); + let mut order: Vec<&String> = local.keys().collect(); + order.sort_by_key(|p| p.matches('/').count()); + + // Same depth-ascending batching as `do_reconcile`: a folder is created + // remotely before the paths inside it are classified, so their parent uid is + // in `remote_dirs` by the time they need it. + let mut batch: Vec = Vec::new(); + let mut batch_depth = 0usize; + for rel in order { + let depth = rel.matches('/').count(); + if depth > batch_depth { + self.flush_batch( + folder_id, + local_root, + &mut remote_dirs, + std::mem::take(&mut batch), + &mut outcome, + ); + batch_depth = depth; + } + let item = &local[rel]; + if item.is_dir { + // The folder is already on Drive, so only its children carry work — + // but if the walk above is what found it, record the uid now so the + // next pass over this folder needs no walk at all. + if let Some(uid) = remote_dirs.get(rel) { + if baseline + .get(rel) + .and_then(|e| e.remote_uid.as_deref()) + .is_none() + && let Err(e) = self.baseline_dir(folder_id, rel, uid) + { + warn!(rel, error = %e, "sync: folder step failed; continuing"); + outcome.errors += 1; + } + continue; + } + match remote_dirs.get(parent_rel(rel)) { + Some(parent) => batch.push(Pending::CreateDir { + rel: rel.clone(), + parent: parent.clone(), + }), + None => { + warn!(rel, "sync: no remote parent for folder; continuing"); + outcome.errors += 1; + } + } + continue; + } + if !changed.contains(rel) { + // Local copy matches the baseline, so it is already on Drive. + continue; + } + + let base = baseline.get(rel); + let uid = base + .and_then(|e| e.remote_uid.as_deref()) + .and_then(parse_uid); + match uid { + // Never uploaded (or the remote node is gone): it goes up as a new file. + None => match remote_dirs.get(parent_rel(rel)) { + Some(parent) => batch.push(Pending::UploadNew { + rel: rel.clone(), + parent: parent.clone(), + }), + None => { + warn!(rel, "sync: no remote parent for file; continuing"); + outcome.errors += 1; + } + }, + Some(uid) => { + let remote_moved = match remote_mtimes.get(&uid.to_string()) { + Some(mtime) => base.and_then(remote_sig).is_none_or(|(m, _)| m != *mtime), + // Trashed or unreadable remotely — treat as gone and re-upload + // the local copy as a new file rather than losing it. + None => { + match remote_dirs.get(parent_rel(rel)) { + Some(parent) => batch.push(Pending::UploadNew { + rel: rel.clone(), + parent: parent.clone(), + }), + None => { + warn!(rel, "sync: no remote parent for file; continuing"); + outcome.errors += 1; + } + } + continue; + } + }; + if remote_moved { + match remote_dirs.get(parent_rel(rel)) { + Some(parent) => batch.push(Pending::PushConflict { + rel: rel.clone(), + parent: parent.clone(), + }), + None => { + warn!(rel, "sync: no remote parent for file; continuing"); + outcome.errors += 1; + } + } + } else { + batch.push(Pending::UploadRevision { + rel: rel.clone(), + uid, + }); + } + } + } + } + self.flush_batch( + folder_id, + local_root, + &mut remote_dirs, + std::mem::take(&mut batch), + &mut outcome, + ); + + // Paths the baseline knows that are no longer on disk: the user deleted them + // locally, and the deletion has to reach Drive before the mount starts + // showing the folder's remote contents — otherwise everything they deleted + // comes back the moment the switch lands. Shallowest first, skipping anything + // under a folder already trashed (its children went with it). + let mut missing: Vec<&String> = baseline + .keys() + .filter(|rel| !local.contains_key(*rel)) + .collect(); + missing.sort_by_key(|p| p.matches('/').count()); + let mut trashed: Vec = Vec::new(); + for rel in missing { + if trashed + .iter() + .any(|dir| rel.starts_with(&format!("{dir}/"))) + { + let _ = self.db.sync_entry_remove(folder_id, rel); + continue; + } + let Some(uid) = baseline[rel].remote_uid.as_deref().and_then(parse_uid) else { + let _ = self.db.sync_entry_remove(folder_id, rel); + continue; + }; + if let Err(e) = self.rt.block_on(self.client.trash_nodes(&[uid])) { + warn!(rel, error = %e, "sync: trash remote failed"); + self.log_activity(ActivityKind::Trash, base_name(rel), e.to_string(), false); + outcome.errors += 1; + continue; + } + trashed.push(rel.clone()); + let _ = self.db.sync_entry_remove(folder_id, rel); + self.log_activity( + ActivityKind::Trash, + base_name(rel), + "removed on Drive", + true, + ); + outcome.deleted += 1; + } + + Ok(outcome) + } + /// The diff-and-apply core. A single path failing to apply (a transient /// upload error, an unreadable file, a name collision) must not abort the /// whole pass and strand every other path — otherwise one bad file in a @@ -490,21 +832,32 @@ impl Core { folder_id: i64, local_root: &Path, remote_root: &NodeUid, - ) -> Result { - let mut local: HashMap = HashMap::new(); - walk_local(local_root, local_root, &mut local)?; - - // Loaded before the walk, which uses it to tell which remote files are - // unchanged and so can skip decrypting their claimed size. + ) -> Result { + // Loaded before the walks: the remote one uses it to tell which files are + // unchanged and so can skip decrypting their claimed size, and its size is + // the only estimate available of how much this pass has to check — the walks + // discover the real figure only by finishing. let baseline = self .db .sync_entries(folder_id) .map_err(|e| format!("load baseline: {e:?}"))?; + // Both sides are walked, so each baseline path is checked about twice. + self.progress_scan_total(folder_id, baseline.len() * 2); + + let mut local: HashMap = HashMap::new(); + self.walk_local(folder_id, local_root, local_root, &mut local)?; let mut remote: HashMap = HashMap::new(); let mut remote_dirs: HashMap = HashMap::new(); remote_dirs.insert(String::new(), remote_root.clone()); - self.walk_remote(remote_root, "", &mut remote, &mut remote_dirs, &baseline)?; + self.walk_remote( + folder_id, + remote_root, + "", + &mut remote, + &mut remote_dirs, + &baseline, + )?; // Union of every path across the three states, shallow paths first so a // parent folder is created before its children are placed inside it. @@ -545,6 +898,11 @@ impl Core { &mut outcome, ); batch_depth = depth; + // A depth boundary is the pass's natural checkpoint — everything + // queued has been applied, so stopping here leaves nothing half done. + if self.ondemand_queued(folder_id) { + return Err(PassAbort::Interrupted); + } } let l = local.get(rel); @@ -745,6 +1103,94 @@ impl Core { Ok(outcome) } + /// Recursively walk a local directory into `out`, keyed by `/`-joined relative + /// path. Symlinks and other special files are skipped. Reports each entry to the + /// pass's scan progress. + fn walk_local( + &self, + folder_id: i64, + root: &Path, + dir: &Path, + out: &mut HashMap, + ) -> Result<(), String> { + let entries = std::fs::read_dir(dir).map_err(|e| format!("read {}: {e}", dir.display()))?; + for entry in entries.filter_map(|e| e.ok()) { + let path = entry.path(); + let meta = match std::fs::symlink_metadata(&path) { + Ok(m) => m, + Err(_) => continue, + }; + let Ok(stripped) = path.strip_prefix(root) else { + continue; + }; + let Some(rel) = stripped.to_str() else { + continue; + }; + // Ignore our own in-flight download temp files. + if rel.contains(".pdfs-tmp-") { + continue; + } + if meta.is_dir() { + out.insert( + rel.to_string(), + LocalItem { + is_dir: true, + mtime: 0, + size: 0, + }, + ); + self.progress_scanned(folder_id, base_name(rel)); + self.walk_local(folder_id, root, &path, out)?; + } else if meta.is_file() { + out.insert( + rel.to_string(), + LocalItem { + is_dir: false, + mtime: system_mtime(&meta), + size: meta.len() as i64, + }, + ); + self.progress_scanned(folder_id, base_name(rel)); + } + } + Ok(()) + } + + /// Map every remote folder under `folder` to its uid by rel path, ignoring + /// files. The cheap half of [`walk_remote`](Self::walk_remote): folders carry no + /// content signature, so a light listing answers this without unlocking a single + /// node key. Used by [`push_pass`](Self::push_pass) to tell a folder it must + /// create from one it just hasn't recorded yet. + fn walk_remote_dirs( + &self, + folder_id: i64, + folder: &NodeUid, + prefix: &str, + dirs: &mut HashMap, + ) -> Result<(), String> { + let uids = self + .rt + .block_on(self.client.enumerate_folder_children_node_uids(folder)) + .map_err(|e| format!("enumerate {folder}: {e}"))?; + if uids.is_empty() { + return Ok(()); + } + let nodes = self + .rt + .block_on(self.client.enumerate_nodes_light(&uids)) + .map_err(|e| format!("resolve nodes: {e}"))?; + for node in nodes { + if node.trashed || !node.is_folder() { + continue; + } + let rel = join_rel(prefix, &node.name); + self.progress_scanned(folder_id, base_name(&rel)); + dirs.insert(rel.clone(), node.uid.clone()); + self.walk_remote_dirs(folder_id, &node.uid, &rel, dirs)?; + } + Ok(()) + } + /// Recursively walk a remote folder into `out`, recording every descendant's /// relative path, and mapping each subfolder's rel path to its uid in `dirs`. /// @@ -758,12 +1204,22 @@ impl Core { /// [`enumerate_nodes_light`]: proton_drive_rs::ProtonDriveClient::enumerate_nodes_light fn walk_remote( &self, + folder_id: i64, folder: &NodeUid, prefix: &str, out: &mut HashMap, dirs: &mut HashMap, baseline: &HashMap, - ) -> Result<(), String> { + ) -> Result<(), PassAbort> { + // Checked per folder, not per pass: this walk is the long pole on a large + // tree, and a user who asks for on-demand three minutes into it should not + // wait out the rest of a survey of files they want deleted. Bailing out here + // is safe precisely because the walk has applied nothing — a *partial* + // remote map, on the other hand, would be read as "the remote deleted + // everything we haven't reached yet", so it must never reach the diff. + if self.ondemand_queued(folder_id) { + return Err(PassAbort::Interrupted); + } let uids = self .rt .block_on(self.client.enumerate_folder_children_node_uids(folder)) @@ -804,6 +1260,7 @@ impl Core { continue; } let rel = join_rel(prefix, &node.name); + self.progress_scanned(folder_id, base_name(&rel)); if node.is_folder() { dirs.insert(rel.clone(), node.uid.clone()); out.insert( @@ -815,7 +1272,7 @@ impl Core { size: 0, }, ); - self.walk_remote(&node.uid, &rel, out, dirs, baseline)?; + self.walk_remote(folder_id, &node.uid, &rel, out, dirs, baseline)?; } else { let size = if stale.contains(&node.uid.to_string()) { match sized.get(&node.uid.to_string()) { @@ -998,6 +1455,26 @@ impl Core { .await?; Ok(Applied::Conflict) } + Pending::PushConflict { rel, parent } => { + // The ordinary conflict resolution — rename the local copy aside and + // let the next pass upload it — loses the file here: the folder is + // going on-demand, so "next pass" comes after the local tree has been + // evicted. Push the conflict copy now, while it still exists, and + // leave the remote file holding the original path. + let path = local_root.join(rel_to_path(rel)); + let conflict = conflict_path(&path, now_secs()); + std::fs::rename(&path, &conflict) + .map_err(|e| format!("set aside conflict copy for {rel}: {e}"))?; + let name = conflict + .file_name() + .map(|n| n.to_string_lossy().into_owned()) + .ok_or_else(|| format!("conflict copy for {rel} has no name"))?; + let conflict_rel = join_rel(parent_rel(rel), &name); + self.upload_new(folder_id, local_root, &conflict_rel, parent) + .await?; + info!(rel, "sync: uploaded local changes as a conflict copy"); + Ok(Applied::Conflict) + } } } @@ -1162,50 +1639,6 @@ impl Core { // ---- helpers -------------------------------------------------------------- -/// Recursively walk a local directory into `out`, keyed by `/`-joined relative -/// path. Symlinks and other special files are skipped. -fn walk_local(root: &Path, dir: &Path, out: &mut HashMap) -> Result<(), String> { - let entries = std::fs::read_dir(dir).map_err(|e| format!("read {}: {e}", dir.display()))?; - for entry in entries.filter_map(|e| e.ok()) { - let path = entry.path(); - let meta = match std::fs::symlink_metadata(&path) { - Ok(m) => m, - Err(_) => continue, - }; - let Ok(stripped) = path.strip_prefix(root) else { - continue; - }; - let Some(rel) = stripped.to_str() else { - continue; - }; - // Ignore our own in-flight download temp files. - if rel.contains(".pdfs-tmp-") { - continue; - } - if meta.is_dir() { - out.insert( - rel.to_string(), - LocalItem { - is_dir: true, - mtime: 0, - size: 0, - }, - ); - walk_local(root, &path, out)?; - } else if meta.is_file() { - out.insert( - rel.to_string(), - LocalItem { - is_dir: false, - mtime: system_mtime(&meta), - size: meta.len() as i64, - }, - ); - } - } - Ok(()) -} - /// A remote node's `(mtime, size)` change signature. Size prefers the plaintext /// `claimed_size`; otherwise the encrypted storage size (stable either way, and /// only ever compared against the same node's own baseline). @@ -1273,7 +1706,7 @@ fn parent_rel(rel: &str) -> &str { } /// The final component of a `/`-joined relative path. -fn base_name(rel: &str) -> &str { +pub(crate) fn base_name(rel: &str) -> &str { match rel.rfind('/') { Some(i) => &rel[i + 1..], None => rel, diff --git a/crates/pdfs-gui/src/app.rs b/crates/pdfs-gui/src/app.rs index 7f4db1d..bc0f900 100644 --- a/crates/pdfs-gui/src/app.rs +++ b/crates/pdfs-gui/src/app.rs @@ -34,8 +34,9 @@ use pdfs_core::auth; use pdfs_core::config::AppDirs; use pdfs_core::control::{ ActivityEntry, ActivityKind, BookmarkInfo, DeviceInfo, DirEntry, InvitationInfo, JobItem, - PhotoItem, PublicLinkInfo, Request, Response, SearchHit, ShareEntry, ShareEntryKind, - SharedItem, SyncFolderInfo, SyncPhase, SyncProgress, TransferDirection, TransferItem, send, + PhotoItem, PublicLinkInfo, RefreshScope, Request, Response, SearchHit, ShareEntry, + ShareEntryKind, SharedItem, SyncFolderInfo, SyncPhase, SyncProgress, TransferDirection, + TransferItem, send, }; use pdfs_core::service; @@ -765,6 +766,18 @@ fn build_window(app: &adw::Application) { wire_shared_by_me(&ui, &shared_by_me_widgets.retry); wire_devices(&ui, &devices_widgets.retry, &devices_widgets.add_folder); wire_activity(&ui, &activity_widgets.retry); + wire_refresh( + &ui, + &[ + &browser_widgets.refresh, + &gallery_widgets.refresh, + &trash_widgets.refresh, + &shared_widgets.refresh, + &shared_by_me_widgets.refresh, + &devices_widgets.refresh, + &activity_widgets.refresh, + ], + ); wire_retry(&ui); // Lazily load the Files / Photos pages the first time they're shown, so the @@ -898,6 +911,75 @@ fn page_fresh(loaded_at: &Cell>) -> bool { loaded_at.get().is_some_and(|t| t.elapsed() < PAGE_TTL) } +/// A page header's Refresh button. Every page that can show stale rows carries +/// one, so the user never has to guess whether what they're looking at is +/// current or wait out a TTL they can't see. +fn refresh_button() -> gtk4::Button { + let button = gtk4::Button::builder() + .icon_name("view-refresh-symbolic") + .tooltip_text("Refresh (F5)") + .valign(gtk4::Align::Center) + .build(); + button.add_css_class("flat"); + button +} + +/// Point every page's Refresh button at the current page. One handler for all of +/// them: the button acts on whatever is on screen, so it can't refresh a page the +/// user has since navigated away from. +fn wire_refresh(ui: &Rc, buttons: &[>k4::Button]) { + for button in buttons { + let ui = ui.clone(); + button.connect_clicked(move |_| reload_current_page(&ui)); + } +} + +/// Re-fetch the visible page from the server, bypassing every layer of cache +/// between it and the account. +/// +/// The two layers are separate: the daemon's own persisted listings (folders, +/// trash, photos) are dropped with [`Request::Refresh`] before re-asking, while +/// the pages the daemon always fetches live (sharing, devices, activity) only +/// need this front-end's [`PAGE_TTL`] stamp cleared. +fn reload_current_page(ui: &Rc) { + match ui.stack.visible_child_name().as_deref() { + Some("browser") => { + let path = ui.browser_path.borrow().clone(); + refresh_then(ui, RefreshScope::Dir { path }, load_browser); + } + Some("gallery") => refresh_then(ui, RefreshScope::Photos, |ui| load_gallery(ui, false)), + Some("trash") => refresh_then(ui, RefreshScope::Trash, load_trash), + Some("shared") => { + ui.shared_loaded_at.set(None); + load_shared(ui); + } + Some("sharedbyme") => { + ui.shared_by_me_loaded_at.set(None); + load_shared_by_me(ui); + } + Some("devices") => { + ui.devices_loaded_at.set(None); + load_devices(ui); + } + Some("activity") => load_activity(ui), + _ => {} + } +} + +/// Drop a daemon-side cached listing, then run the page's loader to re-fetch it. +/// +/// The loader runs even when the invalidation failed: it is the loader that knows +/// how to report an unreachable daemon on its own page, and a refresh that fails +/// silently would read as a dead button. +fn refresh_then(ui: &Rc, scope: RefreshScope, load: fn(&Rc)) { + let rx = spawn_request(ui.dirs.control_socket(), Request::Refresh { scope }); + let ui = ui.clone(); + glib::spawn_future_local(async move { + let _ = rx.recv().await; + load(&ui); + }); +} + /// Highlight the sidebar row for whichever page the stack is showing, so /// navigation that doesn't start in the sidebar (login landing on Files, the tray /// raising the window) still moves the selection. @@ -1025,6 +1107,10 @@ fn install_shortcuts(ui: &Rc, window: &adw::ApplicationWindow) { let ctrl = state.contains(gtk4::gdk::ModifierType::CONTROL_MASK); let on_browser = ui.stack.visible_child_name().as_deref() == Some("browser"); match key { + // Refresh works on every page, so it is matched before the + // browser-only bindings. + gtk4::gdk::Key::F5 => reload_current_page(&ui), + gtk4::gdk::Key::r | gtk4::gdk::Key::R if ctrl => reload_current_page(&ui), gtk4::gdk::Key::f | gtk4::gdk::Key::F if ctrl && on_browser => { ui.browser_search.grab_focus(); } @@ -1877,11 +1963,22 @@ fn refresh_status(ui: &Rc) { used, budget, pins, + online, + pending_uploads, .. })) => { set_mounted(&ui, true); - ui.mount_row - .set_subtitle(&format!("Mounted at {mountpoint}")); + // Queued uploads are the more useful thing to say when there are + // any: they are why a file that looks saved is not on the remote + // yet, and offline is usually the reason they are still queued. + ui.mount_row.set_subtitle(&match (online, pending_uploads) { + (true, 0) => format!("Mounted at {mountpoint}"), + (true, n) => format!("Mounted at {mountpoint} — uploading {n} file(s)"), + (false, 0) => format!("Mounted at {mountpoint} — offline, cached files only"), + (false, n) => { + format!("Mounted at {mountpoint} — offline, {n} file(s) waiting to upload") + } + }); let fraction = if budget == 0 { 0.0 } else { @@ -2029,6 +2126,7 @@ struct BrowserWidgets { new_folder: gtk4::Button, upload: gtk4::Button, upload_folder: gtk4::Button, + refresh: gtk4::Button, /// Wraps the views + the details pane; the pane slides in on selection. split: adw::OverlaySplitView, details: DetailsWidgets, @@ -2103,9 +2201,12 @@ fn build_browser_page() -> (gtk4::Widget, BrowserWidgets) { .build(); search.set_width_chars(18); + let refresh = refresh_button(); + let header = gtk4::Box::new(gtk4::Orientation::Horizontal, 8); header.append(&back); header.append(&crumb_scroll); + header.append(&refresh); header.append(&new_folder); header.append(&upload); header.append(&upload_folder); @@ -2216,6 +2317,7 @@ fn build_browser_page() -> (gtk4::Widget, BrowserWidgets) { new_folder, upload, upload_folder, + refresh, split, details, grid_selection, @@ -3560,6 +3662,7 @@ struct TrashWidgets { status: adw::StatusPage, retry: gtk4::Button, empty: gtk4::Button, + refresh: gtk4::Button, subtitle: gtk4::Label, } @@ -3572,6 +3675,7 @@ struct SharedWidgets { bookmarks: adw::PreferencesGroup, retry: gtk4::Button, add_bookmark: gtk4::Button, + refresh: gtk4::Button, } /// Widgets the Devices page's load/repaint touch. @@ -3580,10 +3684,13 @@ struct DevicesWidgets { status: adw::StatusPage, /// "This computer" — the local folders synced to this machine's device. sync_group: adw::PreferencesGroup, - /// "Other devices" — the account's other registered devices (read-only). + /// "Other computers" — the account's *other* registered devices. This + /// machine's own device is deliberately not among them; see + /// [`repaint_devices`]. group: adw::PreferencesGroup, add_folder: gtk4::Button, retry: gtk4::Button, + refresh: gtk4::Button, } /// The Shared page: three stacked sections — items shared *with* me (each with a @@ -3605,9 +3712,11 @@ fn build_shared_page() -> (gtk4::Widget, SharedWidgets) { .valign(gtk4::Align::Center) .build(); add_bookmark.add_css_class("flat"); + let refresh = refresh_button(); let header = gtk4::Box::new(gtk4::Orientation::Horizontal, 8); header.append(&titles); + header.append(&refresh); header.append(&add_bookmark); let shared_with_me = adw::PreferencesGroup::builder() @@ -3666,6 +3775,7 @@ fn build_shared_page() -> (gtk4::Widget, SharedWidgets) { bookmarks, retry, add_bookmark, + refresh, }, ) } @@ -3688,17 +3798,27 @@ fn build_devices_page() -> (gtk4::Widget, DevicesWidgets) { .valign(gtk4::Align::Center) .build(); add_folder.add_css_class("flat"); + let refresh = refresh_button(); let header = gtk4::Box::new(gtk4::Orientation::Horizontal, 8); header.append(&titles); + header.append(&refresh); header.append(&add_folder); + // The description spells out what the per-row On-demand switch does to the + // local copy, because the switch itself can't: turning it on *deletes* the + // files from this disk, which is not something to discover afterwards. let sync_group = adw::PreferencesGroup::builder() .title("This computer") - .description("Local folders backed up to this device.") + .description( + "Local folders backed up to this device. Synced folders keep a full copy on \ + this disk; on-demand folders keep the files in Proton Drive only and fetch \ + them as you open them.", + ) .build(); let group = adw::PreferencesGroup::builder() - .title("Other devices") + .title("Other computers") + .description("Other devices backing up to this account.") .build(); let groups = gtk4::Box::new(gtk4::Orientation::Vertical, 18); @@ -3747,6 +3867,7 @@ fn build_devices_page() -> (gtk4::Widget, DevicesWidgets) { group, add_folder, retry, + refresh, }, ) } @@ -3779,9 +3900,11 @@ fn build_trash_page() -> (gtk4::Widget, TrashWidgets) { .sensitive(false) .build(); empty.add_css_class("destructive-action"); + let refresh = refresh_button(); let header = gtk4::Box::new(gtk4::Orientation::Horizontal, 8); header.append(&titles); + header.append(&refresh); header.append(&empty); let retry = gtk4::Button::builder() @@ -3831,6 +3954,7 @@ fn build_trash_page() -> (gtk4::Widget, TrashWidgets) { status, retry, empty, + refresh, subtitle, }, ) @@ -4619,16 +4743,24 @@ fn devices_unreachable(ui: &Rc) { }); } -/// Rebuild the "Other devices" section from a fresh listing. Empty is a normal +/// Rebuild the "Other computers" section from a fresh listing. Empty is a normal /// state (this machine may be the only device), so it shows a placeholder row /// rather than collapsing the page. +/// +/// This machine's own device is filtered out. Deleting a device deletes its root +/// folder — every file backed up from it — and for *this* machine that would also +/// pull the ground out from under the synced folders listed directly above, so it +/// is not offered as a peer of "remove some other laptop". Removing this +/// computer's backup means removing its folders, each of which asks about its +/// cloud copy on its own terms. fn repaint_devices(ui: &Rc, devices: &[DeviceInfo]) { for row in ui.devices_rows.borrow_mut().drain(..) { ui.devices_group.remove(&row); } - if devices.is_empty() { + let others: Vec<&DeviceInfo> = devices.iter().filter(|d| !d.this_device).collect(); + if others.is_empty() { let row = adw::ActionRow::builder() - .title("No other devices") + .title("No other computers") .subtitle("Desktop apps syncing to this account appear here.") .build(); row.add_prefix(>k4::Image::from_icon_name("computer-symbolic")); @@ -4637,21 +4769,21 @@ fn repaint_devices(ui: &Rc, devices: &[DeviceInfo]) { return; } let mut rows: Vec = Vec::new(); - for dev in devices { + for dev in others { let row = adw::ActionRow::builder() .title(&dev.name) - .subtitle(&dev.device_type) + .subtitle(device_subtitle(dev)) .build(); row.add_prefix(>k4::Image::from_icon_name("computer-symbolic")); let remove = gtk4::Button::builder() .icon_name("user-trash-symbolic") - .tooltip_text("Remove device") + .tooltip_text("Remove this computer and everything it backed up") .valign(gtk4::Align::Center) .build(); remove.add_css_class("flat"); let rename = gtk4::Button::builder() .icon_name("document-edit-symbolic") - .tooltip_text("Rename device") + .tooltip_text("Rename computer") .valign(gtk4::Align::Center) .build(); rename.add_css_class("flat"); @@ -4671,6 +4803,18 @@ fn repaint_devices(ui: &Rc, devices: &[DeviceInfo]) { *ui.devices_rows.borrow_mut() = rows; } +/// A device row's subtitle: its platform, and when it last synced. A device that +/// has never synced says so — an unexplained missing date reads as a bug, and +/// "never" is the fact that tells the user this computer isn't backing anything up. +fn device_subtitle(dev: &DeviceInfo) -> String { + match dev.last_sync { + Some(secs) if secs > 0 => { + format!("{} · last synced {}", dev.device_type, activity_time(secs)) + } + _ => format!("{} · never synced", dev.device_type), + } +} + /// Rebuild the "This computer" section from a fresh synced-folders listing. fn repaint_sync_folders(ui: &Rc, folders: &[SyncFolderInfo]) { for row in ui.devices_sync_rows.borrow_mut().drain(..) { @@ -4694,9 +4838,14 @@ fn repaint_sync_folders(ui: &Rc, folders: &[SyncFolderInfo]) { Some(p) => sync_progress_label(p), None => sync_state_label(&f.state).to_string(), }; - let subtitle = match f.mode.as_str() { - "ondemand" => format!("On-demand · {status}"), - _ => format!("Synced · {status}"), + // A queued switch is where the folder is *going*, which is what the user + // just asked for and is waiting to see happen — so it leads the subtitle + // instead of the mode the folder is still technically in. + let subtitle = match (f.pending_mode.as_deref(), f.mode.as_str()) { + (Some("ondemand"), _) => format!("Going on-demand · {status}"), + (Some(_), _) => format!("Switching to synced · {status}"), + (None, "ondemand") => format!("On-demand · {status}"), + (None, _) => format!("Synced · {status}"), }; let row = adw::ActionRow::builder() .title(&f.local_path) @@ -4705,17 +4854,17 @@ fn repaint_sync_folders(ui: &Rc, folders: &[SyncFolderInfo]) { row.add_prefix(>k4::Image::from_icon_name("folder-symbolic")); let id = f.id; - // A pass that is applying knows how far it has got, so show it: the - // subtitle alone ("syncing photo.jpg — 12 of 40") never conveys that the - // end is near. A scanning pass has no counts to draw, and this row is - // repainted on the 2s tick — too slow for a pulse to read as motion — so - // it stays text-only until the counts exist. + // A pass that knows how far it has got shows it: the subtitle alone + // ("syncing photo.jpg — 12 of 40") never conveys that the end is near. This + // holds for scanning as much as applying — a big folder spends minutes there + // — but a folder syncing for the first time has no estimate to draw against + // (`total == 0`), and this row is repainted on the 2s tick, too slow for a + // pulse to read as motion. So it stays text-only until real counts exist. if let Some(p) = &f.progress - && p.phase == SyncPhase::Applying && p.total > 0 { let bar = gtk4::ProgressBar::builder() - .fraction((p.done as f64 / p.total as f64).min(1.0)) + .fraction((p.done as f64 / p.total.max(p.done) as f64).min(1.0)) .valign(gtk4::Align::Center) .width_request(120) .build(); @@ -4759,10 +4908,19 @@ fn repaint_sync_folders(ui: &Rc, folders: &[SyncFolderInfo]) { // On-demand toggle: off = full local copy (mirror), on = FUSE mount that // frees the disk. Set the state before wiring the handler so painting the // current mode doesn't fire a spurious request. + // + // A queued switch paints as already flipped: the daemon accepted the + // request and will act on it, so snapping the switch back to the current + // mode would read as "it didn't take" and invite the user to toggle again. + let target = f.pending_mode.as_deref().unwrap_or(&f.mode); let ondemand = gtk4::Switch::builder() - .tooltip_text("On-demand: keep files in the cloud, free local disk") + .tooltip_text( + "On-demand: free this disk by keeping the files in Proton Drive only, \ + fetching each as you open it. Turn off to download them back and keep a \ + full local copy.", + ) .valign(gtk4::Align::Center) - .active(f.mode == "ondemand") + .active(target == "ondemand") .build(); let ui_mode = ui.clone(); ondemand.connect_state_set(move |_, on| { @@ -4779,7 +4937,10 @@ fn repaint_sync_folders(ui: &Rc, folders: &[SyncFolderInfo]) { remove.add_css_class("flat"); let ui_rm = ui.clone(); let path = f.local_path.clone(); - remove.connect_clicked(move |_| prompt_remove_sync_folder(&ui_rm, id, &path)); + // The folder's *current* mode, not a queued one: a switch that hasn't + // landed yet has not moved the files anywhere. + let ondemand = f.mode == "ondemand"; + remove.connect_clicked(move |_| prompt_remove_sync_folder(&ui_rm, id, &path, ondemand)); row.add_suffix(&remove); ui.devices_sync_group.add(&row); rows.push(row.upcast()); @@ -4798,11 +4959,19 @@ fn sync_state_label(state: &str) -> &str { } /// Human label for a sync pass in flight: what it is doing, to which file, and -/// how far along it is. The total grows as deeper paths are classified, so this -/// reads "12 of 40" rather than showing a percentage that could go backwards. +/// how far along it is. Neither phase's total is exact — the scan's is an estimate +/// from the last pass, and the applying total grows as deeper paths are classified +/// — so both read "12 of 40" rather than a percentage that could go backwards. fn sync_progress_label(p: &SyncProgress) -> String { match p.phase { - SyncPhase::Scanning => "checking for changes…".to_string(), + // Before the first pass finishes there is no estimate, so the count would + // be "checked 12 of 12" — worse than saying nothing. + SyncPhase::Scanning if p.total == 0 => "checking for changes…".to_string(), + SyncPhase::Scanning => format!( + "checking for changes — {} of {}", + p.done, + p.total.max(p.done) + ), SyncPhase::Applying => { let count = format!("{} of {}", p.done + 1, p.total.max(p.done + 1)); if p.current.is_empty() { @@ -4890,18 +5059,39 @@ fn set_sync_folder_mode(ui: &Rc, id: i64, mode: &str) { } /// Confirm, then stop syncing a folder. Offers to also delete the cloud copy. -fn prompt_remove_sync_folder(ui: &Rc, id: i64, path: &str) { +fn prompt_remove_sync_folder(ui: &Rc, id: i64, path: &str, ondemand: bool) { let win = ui_window(ui); + // The two modes leave the user in opposite places, so they can't share a + // sentence. A mirror folder's files are already on this disk and stay there. + // An on-demand folder's are not: the path is a mount over content that lives + // in Proton Drive, so unmounting it leaves an empty directory — which is a + // nasty surprise if the dialog claimed the local files were safe, and is + // recoverable only by turning On-demand off *first* and letting it download. + let body = if ondemand { + format!( + "Stop syncing “{path}”?\n\nThis folder is on-demand: its files live in Proton \ + Drive, not on this disk, so the folder will be empty once it is unmounted. To \ + keep a local copy, cancel, turn off On-demand, and wait for the download to \ + finish before removing it." + ) + } else { + format!( + "Stop syncing “{path}”?\n\nThe local files stay on this disk and simply stop \ + being synced. Choose whether to also delete the copy in Proton Drive." + ) + }; let dialog = adw::AlertDialog::builder() .heading("Stop syncing folder") - .body(format!( - "Stop syncing “{path}”? Your local files stay; choose whether to also \ - delete the copy in Proton Drive." - )) + .body(body) .build(); let group = adw::PreferencesGroup::new(); let delete_remote = adw::SwitchRow::builder() .title("Also delete from Proton Drive") + .subtitle(if ondemand { + "Deletes the only copy of these files." + } else { + "The local copy is unaffected." + }) .active(false) .build(); group.add(&delete_remote); @@ -4973,9 +5163,11 @@ fn prompt_rename_device(ui: &Rc, uid: &str, current: &str) { fn prompt_remove_device(ui: &Rc, uid: &str, name: &str) { let win = ui_window(ui); let dialog = adw::AlertDialog::builder() - .heading("Remove device") + .heading("Remove computer") .body(format!( - "Remove “{name}”? Its synced folder registration is deleted." + "Remove “{name}” from this account?\n\nEverything it backed up to Proton Drive is \ + deleted along with it. The files on that computer itself are not touched — but \ + this cannot be undone from here." )) .build(); dialog.add_response("cancel", "Cancel"); @@ -5026,6 +5218,7 @@ struct SharedByMeWidgets { status: adw::StatusPage, group: adw::PreferencesGroup, retry: gtk4::Button, + refresh: gtk4::Button, } /// The Shared page: one section listing the items I have shared with others — @@ -5035,11 +5228,14 @@ fn build_shared_by_me_page() -> (gtk4::Widget, SharedByMeWidgets) { let title = gtk4::Label::builder() .label("Shared") .halign(gtk4::Align::Start) + .hexpand(true) .build(); title.add_css_class("title-2"); + let refresh = refresh_button(); let header = gtk4::Box::new(gtk4::Orientation::Horizontal, 8); header.set_hexpand(true); header.append(&title); + header.append(&refresh); let group = adw::PreferencesGroup::new(); let clamp = adw::Clamp::builder().child(&group).build(); @@ -5083,6 +5279,7 @@ fn build_shared_by_me_page() -> (gtk4::Widget, SharedByMeWidgets) { status, group, retry, + refresh, }, ) } @@ -5303,6 +5500,7 @@ struct ActivityWidgets { status: adw::StatusPage, group: adw::PreferencesGroup, retry: gtk4::Button, + refresh: gtk4::Button, } /// The Activity page: a newest-first feed of the mutations and transfers the @@ -5311,11 +5509,14 @@ fn build_activity_page() -> (gtk4::Widget, ActivityWidgets) { let title = gtk4::Label::builder() .label("Activity") .halign(gtk4::Align::Start) + .hexpand(true) .build(); title.add_css_class("title-2"); + let refresh = refresh_button(); let header = gtk4::Box::new(gtk4::Orientation::Horizontal, 8); header.set_hexpand(true); header.append(&title); + header.append(&refresh); let group = adw::PreferencesGroup::new(); let clamp = adw::Clamp::builder().child(&group).build(); @@ -5359,6 +5560,7 @@ fn build_activity_page() -> (gtk4::Widget, ActivityWidgets) { status, group, retry, + refresh, }, ) } @@ -6082,6 +6284,7 @@ struct GalleryWidgets { scroll: gtk4::ScrolledWindow, retry: gtk4::Button, upload: gtk4::Button, + refresh: gtk4::Button, } /// The Photos page: a [`gtk4::ListView`] of day sections, each a heading over @@ -6164,9 +6367,11 @@ fn build_gallery_page() -> (gtk4::Widget, GalleryWidgets) { .build(); upload.add_css_class("pill"); upload.add_css_class("suggested-action"); + let refresh = refresh_button(); let header_box = gtk4::Box::new(gtk4::Orientation::Horizontal, 8); header_box.append(&titles); + header_box.append(&refresh); header_box.append(&upload); // The timeline (plus its pager) or the status page, never both. @@ -6201,6 +6406,7 @@ fn build_gallery_page() -> (gtk4::Widget, GalleryWidgets) { scroll, retry, upload, + refresh, }, ) } diff --git a/crates/pdfs-gui/src/main.rs b/crates/pdfs-gui/src/main.rs index 0c9ff22..f2b5e54 100644 --- a/crates/pdfs-gui/src/main.rs +++ b/crates/pdfs-gui/src/main.rs @@ -86,9 +86,18 @@ struct DriveTray { fn poll_state(socket: &Path, default_mountpoint: &Path) -> DriveState { match send(socket, &Request::Status) { Ok(Response::Status { - mountpoint, pinned, .. + mountpoint, + pinned, + online, + pending_uploads, + .. }) => DriveState { - line: format!("Mounted at {mountpoint} ({pinned} pinned)"), + line: match (online, pending_uploads) { + (true, 0) => format!("Mounted at {mountpoint} ({pinned} pinned)"), + (true, n) => format!("Uploading {n} file(s) ({pinned} pinned)"), + (false, 0) => format!("Offline — cached files only ({pinned} pinned)"), + (false, n) => format!("Offline — {n} file(s) waiting to upload"), + }, mounted: true, mountpoint: PathBuf::from(mountpoint), // Same daemon is up, so a cheap follow-up poll gives the sync line.