From 067561564e03c67611eae739e128845fd176d4e8 Mon Sep 17 00:00:00 2001 From: Nils Pukropp Date: Wed, 23 Sep 2026 02:06:41 +0200 Subject: [PATCH] feat(core,fuse,gui,cli): pause/resume sync and a visible upload queue (phase 2) Backend (docs/UI_UX_PLAN.md B-1, B-2): - Request::SetSyncPaused { paused, until } pauses the drain worker and the mirror reconcile. FUSE reads, write staging, the watcher and the poll keep running. The pause is persisted in the `state` table (`sync_paused_until`), so it survives a restart; a timed pause ends by itself through a timer thread, re-armed on mount. Resuming wakes the drain and runs a full reconcile. Response::Status gains `paused` and `paused_until`. - Request::ListPendingOps lists the queue with paths, attempts, last error, next attempt and parked/failing flags. Request::RetryPendingOp { id } makes one backed-off op (or every failed one) due now. Parked ops are never un-parked and nothing is dropped from the queue. Front ends: - Sync page: a status card with Pause/Resume (split button with 1 h, 8 h and 24 h pauses) and a "Waiting to Upload" list with Retry now and Retry All. - The sidebar strip and the tray show the paused state; the tray gains Pause Sync / Resume Sync and names failing changes. - CLI: `pdfs sync pause [--for 30m|1h|2d]`, `pdfs sync resume`, `pdfs sync queue`, `pdfs sync retry [id]`; `pdfs status` reports a pause. Not in this phase: merging Activity into the Sync page, discarding a queued op (needs a recovery copy first), and per-folder pause. --- crates/pdfs-cli/src/main.rs | 120 +++++++- crates/pdfs-core/src/control.rs | 45 +++ crates/pdfs-core/src/db/mod.rs | 4 +- crates/pdfs-core/src/db/ops.rs | 25 ++ crates/pdfs-core/src/db/tests.rs | 55 ++++ crates/pdfs-fuse/src/control.rs | 28 ++ crates/pdfs-fuse/src/drain.rs | 11 +- crates/pdfs-fuse/src/lib.rs | 8 +- crates/pdfs-fuse/src/mount.rs | 6 + crates/pdfs-fuse/src/pause.rs | 149 +++++++++ crates/pdfs-fuse/src/queue.rs | 60 ++++ crates/pdfs-fuse/src/sync/engine.rs | 14 +- crates/pdfs-gui/src/app/main.rs | 21 +- crates/pdfs-gui/src/app/pages/locations.rs | 342 +++++++++++++++++++++ crates/pdfs-gui/src/app/pages/status.rs | 66 +++- crates/pdfs-gui/src/main.rs | 35 +++ docs/CHANGELOG.md | 19 ++ 17 files changed, 995 insertions(+), 13 deletions(-) create mode 100644 crates/pdfs-fuse/src/pause.rs create mode 100644 crates/pdfs-fuse/src/queue.rs diff --git a/crates/pdfs-cli/src/main.rs b/crates/pdfs-cli/src/main.rs index 947e15f..ae52362 100644 --- a/crates/pdfs-cli/src/main.rs +++ b/crates/pdfs-cli/src/main.rs @@ -543,6 +543,22 @@ enum SyncCmd { /// Optional synced-folder id; omit to reconcile every folder. id: Option, }, + /// Pause syncing: queued uploads and mirror folders wait, reads keep working. + Pause { + /// End the pause by itself after this long, e.g. `30m`, `1h`, `2d`. + /// Omit to pause until `pdfs sync resume`. + #[arg(long = "for", value_parser = parse_pause_duration)] + duration: Option, + }, + /// Resume syncing after `pdfs sync pause`. + Resume, + /// List uploads and changes not yet on Proton Drive. + Queue, + /// Retry a queued operation now (by id from `sync queue`), or every failed one. + Retry { + /// Queue id; omit to retry every failed operation. + id: Option, + }, /// Switch a folder between full-copy sync and on-demand (FUSE, no local storage). Mode { /// Synced-folder id (from `sync list`). @@ -834,6 +850,19 @@ fn cmd_sync(action: SyncCmd) -> Result<()> { })?)? } SyncCmd::Now { id } => ok_or_bail(control_request(CtlRequest::SyncNow { id })?)?, + SyncCmd::Pause { duration } => { + let until = duration.map(|secs| unix_now().saturating_add(secs as i64)); + ok_or_bail(control_request(CtlRequest::SetSyncPaused { + paused: true, + until, + })?)? + } + SyncCmd::Queue => return cmd_sync_queue(), + SyncCmd::Retry { id } => ok_or_bail(control_request(CtlRequest::RetryPendingOp { id })?)?, + SyncCmd::Resume => ok_or_bail(control_request(CtlRequest::SetSyncPaused { + paused: false, + until: None, + })?)?, SyncCmd::Mode { id, mode } => { ok_or_bail(control_request(CtlRequest::SetSyncFolderMode { id, mode })?)? } @@ -842,6 +871,38 @@ fn cmd_sync(action: SyncCmd) -> Result<()> { Ok(()) } +/// `pdfs sync queue`: what has not reached Proton Drive yet, and why. +fn cmd_sync_queue() -> Result<()> { + let response = control_request(CtlRequest::ListPendingOps)?; + if emit_json(&response)? { + return Ok(()); + } + match response { + CtlResponse::PendingOps { items } if items.is_empty() => { + println!("Nothing queued: everything is on Proton Drive."); + } + CtlResponse::PendingOps { items } => { + let now = unix_now(); + for op in items { + let state = match (op.parked, op.next_attempt_at) { + (true, _) => "waiting for its final name".to_string(), + (false, Some(at)) if at > now => { + format!("next try in {}", human_age((at - now) as u64)) + } + (false, _) => "due now".to_string(), + }; + println!("[{}] {} {} ({state})", op.id, op.kind, op.path); + if let Some(error) = &op.last_error { + println!(" failed {} time(s): {error}", op.attempts); + } + } + } + CtlResponse::Error { message, kind } => bail!("{}", cli_error(kind, &message)), + other => bail!("unexpected response: {other:?}"), + } + Ok(()) +} + /// `pdfs sync restore`: list what this machine's device holds, confirm where /// each folder should land, then hand the mapping to the daemon. /// @@ -1472,9 +1533,15 @@ fn cmd_status() -> Result<()> { failing_error, staged_bytes, staged_oldest_secs, + paused, + paused_until, .. }) => { - let state = if online { "" } else { ", offline" }; + let state = match (online, paused) { + (_, true) => ", sync paused", + (false, false) => ", offline", + (true, false) => "", + }; let queued = match pending_summary(pending_uploads, pending_changes) { Some(s) => format!(", {s}"), None => String::new(), @@ -1490,6 +1557,10 @@ fn cmd_status() -> Result<()> { human_age(staged_oldest_secs), ); } + if let Some(until) = paused_until { + let left = until.saturating_sub(unix_now()).max(0) as u64; + println!("Paused sync resumes in {}", human_age(left.max(1))); + } if parked_uploads > 0 { println!("Parked {parked_uploads} upload(s) waiting for a final filename"); } @@ -2384,6 +2455,31 @@ fn path_arg(path: &Path) -> Result { /// Format an age in seconds for a report line, coarsely — the point of the /// number is whether it is minutes or weeks. +/// Parse a `sync pause --for` duration: a whole number followed by `m`, `h` or +/// `d`. Returns seconds. +fn parse_pause_duration(text: &str) -> Result { + let text = text.trim(); + let split = text.char_indices().last().map_or(0, |(at, _)| at); + let (number, unit) = text.split_at(split); + let per_unit: u64 = match unit { + "m" => 60, + "h" => 3600, + "d" => 86_400, + _ => return Err("use a number followed by m, h or d, e.g. 30m".to_string()), + }; + match number.parse::() { + Ok(n) if n > 0 => Ok(n.saturating_mul(per_unit)), + _ => Err("use a positive whole number, e.g. 30m".to_string()), + } +} + +fn unix_now() -> i64 { + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map(|d| d.as_secs() as i64) + .unwrap_or(0) +} + fn human_age(secs: u64) -> String { const UNITS: [(&str, u64); 4] = [("d", 86_400), ("h", 3_600), ("m", 60), ("s", 1)]; for (unit, scale) in UNITS { @@ -2847,6 +2943,28 @@ fn control_request(req: CtlRequest) -> Result { .with_context(|| format!("talk to mount daemon at {}", socket.display())) } +#[cfg(test)] +mod pause_tests { + use super::*; + + #[test] + fn pause_durations_read_minutes_hours_and_days() { + assert_eq!(parse_pause_duration("30m"), Ok(1800)); + assert_eq!(parse_pause_duration("1h"), Ok(3600)); + assert_eq!(parse_pause_duration("2d"), Ok(172_800)); + } + + #[test] + fn pause_durations_reject_what_they_cannot_read() { + assert!(parse_pause_duration("").is_err()); + assert!(parse_pause_duration("0h").is_err()); + assert!(parse_pause_duration("1w").is_err()); + assert!(parse_pause_duration("h").is_err()); + assert!(parse_pause_duration("-1h").is_err()); + assert!(parse_pause_duration("1ä").is_err()); + } +} + #[cfg(test)] mod diagnose_tests { use super::*; diff --git a/crates/pdfs-core/src/control.rs b/crates/pdfs-core/src/control.rs index 75e4129..629c731 100644 --- a/crates/pdfs-core/src/control.rs +++ b/crates/pdfs-core/src/control.rs @@ -384,6 +384,18 @@ pub enum Request { /// Force a reconcile pass: one folder by id, or all when `id` is `None`. /// Replies with [`Response::Ok`]. SyncNow { id: Option }, + /// Pause or resume syncing: queued uploads and changes stop draining and + /// mirror folders stop reconciling, while reads through the mount keep + /// working. `until` is the Unix second a pause ends by itself; `None` + /// pauses until resumed. Ignored when resuming. Replies with + /// [`Response::Ok`]. + SetSyncPaused { paused: bool, until: Option }, + /// List the queue of uploads and changes not yet on the remote. Replies + /// with [`Response::PendingOps`]. + ListPendingOps, + /// Retry a backed-off queued op now instead of waiting out its backoff; + /// every failed op when `id` is `None`. Replies with [`Response::Ok`]. + RetryPendingOp { id: Option }, /// List the folders under this machine's device that can be synced here, /// each with a proposed local path (features.md 5.2). Replies with /// [`Response::RestorableFolders`]. @@ -626,6 +638,30 @@ pub struct DeviceInfo { pub adopted: bool, } +/// One queued upload or change (in [`Response::PendingOps`]). +#[derive(Serialize, Deserialize, Debug, Clone)] +pub struct PendingOpInfo { + /// Queue row id, for [`Request::RetryPendingOp`]. + pub id: i64, + /// `revision`, `create`, `mkdir`, `rename` or `trash`. + pub kind: String, + /// Mountpoint-relative path of the file or folder, or its name when the + /// path is no longer known. + pub path: String, + /// Failed attempts so far. + pub attempts: i64, + /// The latest failure, when there has been one. + pub last_error: Option, + /// Unix second the op was queued. + pub queued_at: i64, + /// Unix second of the next attempt; `None` while parked. + pub next_attempt_at: Option, + /// Held back until a transient file gets its final name. + pub parked: bool, + /// Failed often enough to count as stuck. + pub failing: bool, +} + /// One synced local folder on this machine's device (in [`Response::SyncFolders`]). #[derive(Serialize, Deserialize, Debug, Clone)] pub struct SyncFolderInfo { @@ -1562,6 +1598,13 @@ pub enum Response { /// Age of the oldest staged write, in seconds. #[serde(default)] staged_oldest_secs: u64, + /// True while the user has sync paused (see [`Request::SetSyncPaused`]). + #[serde(default)] + paused: bool, + /// The Unix second a timed pause ends; `None` when paused until + /// resumed, or not paused. + #[serde(default)] + paused_until: Option, }, /// A human-readable success message. Ok { message: String }, @@ -1691,6 +1734,8 @@ pub enum Response { Locations { items: Vec }, /// This device's synced folders (reply to [`Request::ListSyncFolders`]). SyncFolders { items: Vec }, + /// The upload/change queue (reply to [`Request::ListPendingOps`]). + PendingOps { items: Vec }, /// Folders offered for restore (reply to [`Request::ListRestorableFolders`]). RestorableFolders { items: Vec }, /// A node's share: members + pending invitations, and its public link if any diff --git a/crates/pdfs-core/src/db/mod.rs b/crates/pdfs-core/src/db/mod.rs index 5f5559c..c3b29d5 100644 --- a/crates/pdfs-core/src/db/mod.rs +++ b/crates/pdfs-core/src/db/mod.rs @@ -56,8 +56,8 @@ pub use local::LocalFileHit; pub use maintenance::{DbStats, VacuumOutcome}; pub use nodes::{PublishedSharedRoot, SearchHit, StoredNode}; pub use ops::{ - AttachedBlob, LOCAL_VOLUME, OP_CREATE, OP_MKDIR, OP_RENAME, OP_REVISION, OP_TRASH, - PARK_EXPIRY_MS, PARK_UNTIL, PendingCounts, PendingOp, RenameMeta, op_supersedes, + AttachedBlob, FAILING_ATTEMPTS, LOCAL_VOLUME, OP_CREATE, OP_MKDIR, OP_RENAME, OP_REVISION, + OP_TRASH, PARK_EXPIRY_MS, PARK_UNTIL, PendingCounts, PendingOp, RenameMeta, op_supersedes, }; pub use photos::{StoredPhoto, THUMB_HAVE, THUMB_NONE, THUMB_UNKNOWN, TimelineRow}; pub use pins::PinRow; diff --git a/crates/pdfs-core/src/db/ops.rs b/crates/pdfs-core/src/db/ops.rs index b21d87a..6f85027 100644 --- a/crates/pdfs-core/src/db/ops.rs +++ b/crates/pdfs-core/src/db/ops.rs @@ -772,6 +772,31 @@ impl Db { Ok(()) } + /// Make one backed-off op due now, for the user's "Retry now". Parked ops are + /// left alone: they wait for a rename, not for time, and un-parking one would + /// upload a file its writer has not finished. Returns whether a row changed. + pub fn retry_op_now(&self, id: i64, now: i64) -> Result { + let conn = self.conn.lock(); + let changed = conn.execute( + "UPDATE pending_op SET next_attempt_at = ?2 + WHERE id = ?1 AND next_attempt_at > ?2 AND next_attempt_at < ?3", + params![id, now, PARK_UNTIL], + )?; + Ok(changed > 0) + } + + /// [`retry_op_now`](Self::retry_op_now) for every op that has failed at + /// least once. Returns how many rows changed. + pub fn retry_failed_ops_now(&self, now: i64) -> Result { + let conn = self.conn.lock(); + let changed = conn.execute( + "UPDATE pending_op SET next_attempt_at = ?1 + WHERE attempts > 0 AND next_attempt_at > ?1 AND next_attempt_at < ?2", + params![now, PARK_UNTIL], + )?; + Ok(changed) + } + /// 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. /// diff --git a/crates/pdfs-core/src/db/tests.rs b/crates/pdfs-core/src/db/tests.rs index 7c10593..a42b204 100644 --- a/crates/pdfs-core/src/db/tests.rs +++ b/crates/pdfs-core/src/db/tests.rs @@ -4480,3 +4480,58 @@ fn migration_v31_treats_an_existing_timeline_as_resolved() { let _ = std::fs::remove_file(&path); let _ = std::fs::remove_file(format!("{}.lock", path.display())); } + +fn queued_op(db: &Db, uid: &str, failures: i64, next_attempt_at: i64) -> i64 { + let (id, _) = db + .enqueue_op(&ops::PendingOp { + id: 0, + kind: ops::OP_REVISION.to_string(), + uid: uid.to_string(), + parent_uid: None, + name: None, + blob_path: None, + meta_json: None, + created_at: 1000, + attempts: 0, + last_error: None, + next_attempt_at, + }) + .unwrap(); + for _ in 0..failures { + db.record_op_failure(id, "boom", next_attempt_at).unwrap(); + } + id +} + +/// "Retry now" must make a backed-off op due, but never un-park a transient +/// file's op: that one waits for its finishing rename, not for time. +#[test] +fn retry_now_makes_a_backed_off_op_due_but_leaves_parked_ops_parked() { + let db = Db::open_in_memory().unwrap(); + let backed_off = queued_op(&db, "vol~a", 3, 90_000); + let parked = queued_op(&db, "vol~b", 0, ops::PARK_UNTIL); + + assert!(db.retry_op_now(backed_off, 5_000).unwrap()); + assert!(!db.retry_op_now(parked, 5_000).unwrap()); + + let ops = db.pending_ops().unwrap(); + let due = |id: i64| ops.iter().find(|op| op.id == id).unwrap().next_attempt_at; + assert_eq!(due(backed_off), 5_000); + assert_eq!(due(parked), ops::PARK_UNTIL); +} + +#[test] +fn retry_all_touches_only_ops_that_have_failed() { + let db = Db::open_in_memory().unwrap(); + let failed = queued_op(&db, "vol~a", 2, 90_000); + let debounced = queued_op(&db, "vol~b", 0, 7_000); + let parked = queued_op(&db, "vol~c", 4, ops::PARK_UNTIL); + + assert_eq!(db.retry_failed_ops_now(5_000).unwrap(), 1); + + let ops = db.pending_ops().unwrap(); + let due = |id: i64| ops.iter().find(|op| op.id == id).unwrap().next_attempt_at; + assert_eq!(due(failed), 5_000); + assert_eq!(due(debounced), 7_000); + assert_eq!(due(parked), ops::PARK_UNTIL); +} diff --git a/crates/pdfs-fuse/src/control.rs b/crates/pdfs-fuse/src/control.rs index 92fed0f..728d05f 100644 --- a/crates/pdfs-fuse/src/control.rs +++ b/crates/pdfs-fuse/src/control.rs @@ -246,6 +246,8 @@ fn handle_control_conn(core: &Core, username: &str, mountpoint: &Path, stream: U failing_error: queued.last_error.clone(), staged_bytes: staging.bytes, staged_oldest_secs: staging.oldest_secs, + paused: core.sync_pause().is_some(), + paused_until: core.sync_pause().flatten(), username: username.to_string(), mountpoint: mountpoint.display().to_string(), pinned: pins.len(), @@ -989,6 +991,32 @@ fn handle_control_conn(core: &Core, username: &str, mountpoint: &Path, stream: U }, } } + Ok(CtlRequest::SetSyncPaused { paused, until }) => { + match core.set_sync_paused(paused, until) { + Ok(()) => CtlResponse::Ok { + message: match (paused, until) { + (false, _) => "sync resumed".to_string(), + (true, None) => "sync paused until resumed".to_string(), + (true, Some(_)) => "sync paused".to_string(), + }, + }, + Err(e) => CtlResponse::error(e), + } + } + Ok(CtlRequest::ListPendingOps) => match core.pending_op_infos() { + Ok(items) => CtlResponse::PendingOps { items }, + Err(e) => CtlResponse::error(e), + }, + Ok(CtlRequest::RetryPendingOp { id }) => match core.retry_pending_ops(id) { + Ok(0) if id.is_some() => CtlResponse::Ok { + message: "nothing to retry: the operation is already due or waiting for a rename" + .to_string(), + }, + Ok(moved) => CtlResponse::Ok { + message: format!("retrying {moved} queued operation(s) now"), + }, + Err(e) => CtlResponse::error(e), + }, Ok(CtlRequest::ListRestorableFolders) => match core.list_restorable_folders() { Ok(items) => CtlResponse::RestorableFolders { items }, Err(e) => CtlResponse::error(e), diff --git a/crates/pdfs-fuse/src/drain.rs b/crates/pdfs-fuse/src/drain.rs index 62cebaf..2cc44ef 100644 --- a/crates/pdfs-fuse/src/drain.rs +++ b/crates/pdfs-fuse/src/drain.rs @@ -285,8 +285,9 @@ impl Core { } }; let online = self.online.load(Ordering::Relaxed); + let paused = self.sync_paused(); let op = match due { - Some(op) if online => op, + Some(op) if online && !paused => op, idle => { // Claimed but offline: hand it straight back, or it stays // invisible to every worker until this one happens to be @@ -306,6 +307,14 @@ impl Core { // its writer has to be noticed. self.sweep_parked_creates(); } + // Paused, nothing due changes that: sleep until a resume + // wakes the worker (or the idle poll notices a timed pause + // ran out), not until the next op falls due — that is + // "now" for everything already queued. + if paused { + self.wait_for_drain_work(); + continue; + } // A debounced or backed-off op may be waiting: sleep only // until it becomes due rather than the full idle-poll. self.wait_for_drain_work_or_due(); diff --git a/crates/pdfs-fuse/src/lib.rs b/crates/pdfs-fuse/src/lib.rs index 86727e9..4a4c531 100644 --- a/crates/pdfs-fuse/src/lib.rs +++ b/crates/pdfs-fuse/src/lib.rs @@ -42,7 +42,7 @@ use std::os::unix::net::UnixListener; use std::path::{Component, Path, PathBuf}; use std::sync::Arc; use std::sync::OnceLock; -use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; +use std::sync::atomic::{AtomicBool, AtomicI64, AtomicU64, Ordering}; use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; use fuser::ReplyXattr; @@ -88,8 +88,10 @@ mod drain; mod filesystem; pub use filesystem::ProtonFs; mod mount; +mod pause; mod photos; mod profile; +mod queue; mod reads; mod redate; mod revisions; @@ -508,6 +510,10 @@ struct Core { /// `Response::Status` so the UI can say so rather than leaving the user to /// infer it from a wall of EIO. online: Arc, + /// When the user paused syncing, the Unix second it resumes by itself; + /// [`pause::PAUSED_INDEFINITELY`] for "until resumed", `0` when not paused. + /// Persisted, so a pause survives a restart (see [`Core::sync_paused`]). + sync_paused_until: 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. diff --git a/crates/pdfs-fuse/src/mount.rs b/crates/pdfs-fuse/src/mount.rs index 9a6451c..2a6652f 100644 --- a/crates/pdfs-fuse/src/mount.rs +++ b/crates/pdfs-fuse/src/mount.rs @@ -357,6 +357,7 @@ pub fn mount( // handlers can trigger reconciles. let (sync_tx, sync_rx) = std::sync::mpsc::channel::(); + let paused_until = pause::load_paused_until(&db); let core = Core { client: client.clone(), rt: rt.clone(), @@ -377,6 +378,7 @@ pub fn mount( shared_generation: Arc::new(AtomicU64::new(0)), shared_refresh_deadlines: Arc::new(Mutex::new(SharedRefreshDeadlines::default())), online: Arc::new(AtomicBool::new(online)), + sync_paused_until: Arc::new(AtomicI64::new(paused_until)), pending: Arc::new(Mutex::new(HashMap::new())), hidden: Arc::new(Mutex::new(HashSet::new())), drain_wake: Arc::new((Mutex::new(false), Condvar::new())), @@ -451,6 +453,10 @@ pub fn mount( // 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. workers.extend(sync::spawn(core.clone(), sync_rx)); + // A timed pause persisted by an earlier run still has to end on time. + if pause::paused_at(paused_until, now_secs()) && paused_until != pause::PAUSED_INDEFINITELY { + core.arm_pause_timer(paused_until); + } // Reconcile leftover `(sync-conflict …)` copies: drop the ones that turned // out identical to the live file, surface the ones that genuinely diverge. diff --git a/crates/pdfs-fuse/src/pause.rs b/crates/pdfs-fuse/src/pause.rs new file mode 100644 index 0000000..7800fa5 --- /dev/null +++ b/crates/pdfs-fuse/src/pause.rs @@ -0,0 +1,149 @@ +//! Pausing sync: the user's "stop uploading for a while". +//! +//! A pause holds back the two things that push local state to the remote — the +//! drain worker's queued uploads and changes, and the mirror-folder reconcile. +//! Nothing else stops: reads through the mount keep hydrating on demand (a read +//! is not a sync), writes keep being accepted and staged, and the watcher and +//! poll keep noticing changes. Resuming lets all of it catch up at once. +//! +//! The pause is persisted in the database, so a restart does not quietly undo +//! it. A timed pause ends by itself; [`Core::set_sync_paused`] arms a timer for +//! that, and [`Core::sync_paused`] compares against the clock anyway, so a +//! deadline that passed while the daemon was down is already over on mount. + +use super::*; + +/// The `state` key holding the resume time, in Unix seconds. +const PAUSED_UNTIL_KEY: &str = "sync_paused_until"; + +/// A pause with no end time: it lasts until the user resumes. +pub(crate) const PAUSED_INDEFINITELY: i64 = i64::MAX; + +/// The persisted pause, or `0` when there is none (or it cannot be read — a +/// broken state row must not keep sync paused forever). +pub(crate) fn load_paused_until(db: &Db) -> i64 { + match db.state_i64(PAUSED_UNTIL_KEY) { + Ok(Some(until)) => until, + Ok(None) => 0, + Err(error) => { + warn!(%error, "reading the sync pause failed; treating sync as running"); + 0 + } + } +} + +/// Whether a pause lasting until `until` is still in force at `now` (both Unix +/// seconds). `0` is "not paused". +pub(crate) fn paused_at(until: i64, now: i64) -> bool { + until != 0 && now < until +} + +impl Core { + /// True while the user has sync paused. + pub(crate) fn sync_paused(&self) -> bool { + paused_at(self.sync_paused_until.load(Ordering::Relaxed), now_secs()) + } + + /// The pause's end, for status reporting: `None` when not paused, + /// `Some(None)` when paused until resumed, `Some(Some(t))` when it ends at + /// Unix second `t`. + pub(crate) fn sync_pause(&self) -> Option> { + let until = self.sync_paused_until.load(Ordering::Relaxed); + match paused_at(until, now_secs()) { + false => None, + true if until == PAUSED_INDEFINITELY => Some(None), + true => Some(Some(until)), + } + } + + /// Pause sync until Unix second `until` (`None`: until resumed), or resume + /// it when `paused` is false. + pub(crate) fn set_sync_paused( + &self, + paused: bool, + until: Option, + ) -> Result<(), CoreError> { + let value = match (paused, until) { + (false, _) => 0, + (true, None) => PAUSED_INDEFINITELY, + (true, Some(until)) if until <= now_secs() => { + return Err(CoreError::invalid("the pause would already be over")); + } + (true, Some(until)) => until, + }; + let persisted = match value { + 0 => self.db.clear_state(PAUSED_UNTIL_KEY), + value => self.db.set_state_i64(PAUSED_UNTIL_KEY, value), + }; + if let Err(error) = persisted { + return Err(CoreError::internal(format!( + "saving the sync pause: {error}" + ))); + } + self.sync_paused_until.store(value, Ordering::Relaxed); + match value { + 0 => { + info!("sync resumed"); + self.resume_sync_work(); + } + PAUSED_INDEFINITELY => info!("sync paused until resumed"), + until => { + info!(until, "sync paused"); + self.arm_pause_timer(until); + } + } + Ok(()) + } + + /// Wake everything a pause held back: the drain worker, and a full + /// reconcile to pick up whatever the watcher saw meanwhile. + fn resume_sync_work(&self) { + self.wake_drain(); + self.sync_now(None); + } + + /// End a timed pause when its time comes. A later pause or resume replaces + /// the stored deadline, which the timer checks before acting, so a stale + /// timer does nothing. + pub(crate) fn arm_pause_timer(&self, until: i64) { + let core = self.clone(); + let wait = Duration::from_secs(until.saturating_sub(now_secs()).max(0) as u64); + let spawned = std::thread::Builder::new() + .name("pdfs-pause".into()) + .spawn(move || { + if !core.shutdown.sleep(wait) { + return; + } + if core + .sync_paused_until + .compare_exchange(until, 0, Ordering::Relaxed, Ordering::Relaxed) + .is_ok() + { + if let Err(error) = core.db.clear_state(PAUSED_UNTIL_KEY) { + warn!(%error, "clearing an expired sync pause failed"); + } + info!("sync pause ended"); + core.resume_sync_work(); + } + }); + if let Err(error) = spawned { + // The drain and the reconcile still read the clock, so the pause + // still ends — just at the next idle poll instead of on time. + warn!(%error, "starting the sync pause timer failed"); + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn a_pause_is_in_force_until_its_deadline() { + assert!(!paused_at(0, 100)); + assert!(paused_at(200, 100)); + assert!(!paused_at(200, 200)); + assert!(!paused_at(200, 300)); + assert!(paused_at(PAUSED_INDEFINITELY, i64::MAX - 1)); + } +} diff --git a/crates/pdfs-fuse/src/queue.rs b/crates/pdfs-fuse/src/queue.rs new file mode 100644 index 0000000..6c6a12d --- /dev/null +++ b/crates/pdfs-fuse/src/queue.rs @@ -0,0 +1,60 @@ +//! The user's view of the pending-op queue: what has not reached the remote +//! yet, why, and a way to stop waiting out a backoff. +//! +//! Read-only apart from "retry now", which only moves a row's next-attempt time. +//! Nothing here drops a queued op: its staged blob is the only copy of the +//! user's write (see `docs/RECOVERY.md`). + +use super::*; +use pdfs_core::control::PendingOpInfo; +use pdfs_core::db::{FAILING_ATTEMPTS, PARK_UNTIL}; + +impl Core { + /// Every queued op, oldest first, with a path a person can read. + pub(crate) fn pending_op_infos(&self) -> CoreResult> { + let ops = self + .db + .pending_ops() + .map_err(|e| CoreError::internal(format!("listing the queue: {e}")))?; + Ok(ops + .into_iter() + .map(|op| { + let parked = op.next_attempt_at >= PARK_UNTIL; + let path = self + .db + .node_path(&op.uid) + .ok() + .flatten() + .filter(|path| !path.is_empty()) + .or_else(|| op.name.clone()) + .unwrap_or_else(|| op.uid.clone()); + PendingOpInfo { + id: op.id, + kind: op.kind, + path, + attempts: op.attempts, + last_error: op.last_error, + queued_at: op.created_at / 1000, + next_attempt_at: (!parked).then_some(op.next_attempt_at / 1000), + parked, + failing: op.attempts >= FAILING_ATTEMPTS, + } + }) + .collect()) + } + + /// Retry one backed-off op now, or every failed one when `id` is `None`. + /// Returns how many ops were moved up. + pub(crate) fn retry_pending_ops(&self, id: Option) -> CoreResult { + let now = now_millis(); + let moved = match id { + Some(id) => self.db.retry_op_now(id, now).map(usize::from), + None => self.db.retry_failed_ops_now(now), + } + .map_err(|e| CoreError::internal(format!("retrying the queue: {e}")))?; + if moved > 0 { + self.wake_drain(); + } + Ok(moved) + } +} diff --git a/crates/pdfs-fuse/src/sync/engine.rs b/crates/pdfs-fuse/src/sync/engine.rs index 8725dd6..2fac7df 100644 --- a/crates/pdfs-fuse/src/sync/engine.rs +++ b/crates/pdfs-fuse/src/sync/engine.rs @@ -115,6 +115,9 @@ fn engine_loop(core: Core, rx: Receiver) { reconcile_all(&core); } else { for id in ids { + if core.sync_paused() { + break; + } if let Ok(Some(folder)) = core.db.sync_folder_get(id) { core.reconcile_folder(&folder); } @@ -151,6 +154,9 @@ fn poll_only_loop(core: &Core, rx: Receiver) { reconcile_all(core); } else { for id in ids { + if core.sync_paused() { + break; + } if let Ok(Some(folder)) = core.db.sync_folder_get(id) { core.reconcile_folder(&folder); } @@ -230,8 +236,14 @@ pub(super) fn classify( true } -/// Reconcile every mirror folder in turn. +/// Reconcile every mirror folder in turn. Skipped while sync is paused: the +/// resume sends a fresh [`SyncMsg::ReconcileAll`], which walks everything the +/// watcher reported meanwhile. fn reconcile_all(core: &Core) { + if core.sync_paused() { + debug!("sync: paused, skipping reconcile"); + return; + } let folders = match core.db.sync_folder_list() { Ok(f) => f, Err(e) => { diff --git a/crates/pdfs-gui/src/app/main.rs b/crates/pdfs-gui/src/app/main.rs index 4af4c6c..05ec48e 100644 --- a/crates/pdfs-gui/src/app/main.rs +++ b/crates/pdfs-gui/src/app/main.rs @@ -48,10 +48,10 @@ use pdfs_core::config::AppDirs; use pdfs_core::control::{ ActivityEntry, ActivityKind, AlbumInfo, BookmarkInfo, DeviceInfo, DirEntry, ErrorKind, - ImportSummary, InvitationInfo, JobItem, PhotoItem, PhotoKind, PublicLinkInfo, RefreshScope, - Request, Response, RestorableFolder, RestoreItem, SearchHit, ShareEntry, ShareEntryKind, - SharedItem, SyncFolderInfo, SyncPhase, SyncProgress, ThumbnailBuildStatus, TransferDirection, - TransferItem, pending_summary, send, + ImportSummary, InvitationInfo, JobItem, PendingOpInfo, PhotoItem, PhotoKind, PublicLinkInfo, + RefreshScope, Request, Response, RestorableFolder, RestoreItem, SearchHit, ShareEntry, + ShareEntryKind, SharedItem, SyncFolderInfo, SyncPhase, SyncProgress, ThumbnailBuildStatus, + TransferDirection, TransferItem, pending_summary, send, }; use pdfs_core::mounts::{MountAccess, MountKind, MountMode, MountSpec}; @@ -542,6 +542,19 @@ fn build_window(app: &adw::Application) { rows: RefCell::new(Vec::new()), inflight: Cell::new(false), loaded_at: Cell::new(None), + card: SyncCard { + icon: locations_widgets.card_icon.clone(), + row: locations_widgets.card_row.clone(), + pause: locations_widgets.pause.clone(), + paused: Cell::new(false), + }, + queue: QueueState { + group: locations_widgets.queue_group.clone(), + retry_all: locations_widgets.retry_all.clone(), + rows: RefCell::new(Vec::new()), + painted: RefCell::new(Vec::new()), + inflight: Cell::new(false), + }, }, activity: ActivityState { content: activity_widgets.content.clone(), diff --git a/crates/pdfs-gui/src/app/pages/locations.rs b/crates/pdfs-gui/src/app/pages/locations.rs index 0cd97b8..26c3ec5 100644 --- a/crates/pdfs-gui/src/app/pages/locations.rs +++ b/crates/pdfs-gui/src/app/pages/locations.rs @@ -22,6 +22,31 @@ pub(crate) struct LocationsState { pub(crate) rows: RefCell>, pub(crate) inflight: Cell, pub(crate) loaded_at: Cell>, + pub(crate) card: SyncCard, + pub(crate) queue: QueueState, +} + +/// The status card heading the Sync page: what sync is doing, and Pause/Resume. +pub(crate) struct SyncCard { + pub(crate) icon: gtk4::Image, + pub(crate) row: adw::ActionRow, + pub(crate) pause: adw::SplitButton, + /// Whether the last status said paused, so the button knows which way to go. + pub(crate) paused: Cell, +} + +/// What a queue row shows that can change: id, attempts, next attempt, parked. +pub(crate) type QueueKey = (i64, i64, Option, bool); + +/// The "Waiting to Upload" list: the daemon's pending-op queue. +pub(crate) struct QueueState { + pub(crate) group: adw::PreferencesGroup, + pub(crate) retry_all: gtk4::Button, + pub(crate) rows: RefCell>, + /// What the rows were built from, so an unchanged queue is not rebuilt on + /// every tick. + pub(crate) painted: RefCell>, + pub(crate) inflight: Cell, } /// Widgets the Locations page's load/repaint touch. @@ -34,6 +59,11 @@ pub(crate) struct LocationsWidgets { pub(crate) add_folder: gtk4::Button, /// Live transfers, above the folder list; painted by the refresh loop. pub(crate) transfers_group: adw::PreferencesGroup, + pub(crate) card_icon: gtk4::Image, + pub(crate) card_row: adw::ActionRow, + pub(crate) pause: adw::SplitButton, + pub(crate) queue_group: adw::PreferencesGroup, + pub(crate) retry_all: gtk4::Button, } pub(crate) fn build_locations_page() -> (gtk4::Widget, LocationsWidgets) { @@ -48,6 +78,8 @@ pub(crate) fn build_locations_page() -> (gtk4::Widget, LocationsWidgets) { .build(); let refresh = refresh_button(); let transfers_group = build_transfers_group(); + let (card, card_icon, card_row, pause) = build_sync_card(); + let (queue_group, retry_all) = build_queue_group(); // Same warning the Computers page carried, for the same reason: the // on-demand switch removes the local copy, which is not a thing to discover @@ -62,6 +94,8 @@ pub(crate) fn build_locations_page() -> (gtk4::Widget, LocationsWidgets) { .build(); let groups = gtk4::Box::new(gtk4::Orientation::Vertical, 18); + groups.append(&card); + groups.append(&queue_group); groups.append(&transfers_group); groups.append(&group); let clamp = adw::Clamp::builder().child(&groups).build(); @@ -118,6 +152,11 @@ pub(crate) fn build_locations_page() -> (gtk4::Widget, LocationsWidgets) { refresh, add_folder, transfers_group, + card_icon, + card_row, + pause, + queue_group, + retry_all, }; (frame.upcast(), widgets) } @@ -130,6 +169,309 @@ pub(crate) fn wire_locations(ui: &Rc, retry: >k4::Button, add_folder: >k }); let ui_add = ui.clone(); add_folder.connect_clicked(move |_| prompt_add_sync_folder(&ui_add)); + wire_pause(ui); + let ui_retry = ui.clone(); + ui.locations + .queue + .retry_all + .connect_clicked(move |_| retry_queued(&ui_retry, None)); +} + +/// The status card: a large state icon, the state in words, and Pause/Resume +/// with timed pauses in its menu. +fn build_sync_card() -> ( + adw::PreferencesGroup, + gtk4::Image, + adw::ActionRow, + adw::SplitButton, +) { + let icon = gtk4::Image::builder() + .icon_name("emblem-synchronizing-symbolic") + .pixel_size(32) + .margin_top(6) + .margin_bottom(6) + .build(); + icon.add_css_class("sidebar-status"); + let menu = gio::Menu::new(); + menu.append(Some("Pause for 1 Hour"), Some("sync.pause-for(int64 3600)")); + menu.append( + Some("Pause for 8 Hours"), + Some("sync.pause-for(int64 28800)"), + ); + menu.append( + Some("Pause for 24 Hours"), + Some("sync.pause-for(int64 86400)"), + ); + let pause = adw::SplitButton::builder() + .label("Pause") + .menu_model(&menu) + .valign(gtk4::Align::Center) + .tooltip_text("Stop uploading until you resume. Files still open as usual.") + .dropdown_tooltip("Pause for a while") + .build(); + let row = adw::ActionRow::builder() + .title("Checking…") + .title_lines(1) + .subtitle_lines(2) + .build(); + row.add_css_class("property"); + row.add_prefix(&icon); + row.add_suffix(&pause); + let group = adw::PreferencesGroup::new(); + group.add(&row); + (group, icon, row, pause) +} + +/// Paint the Sync page's status card from what the sidebar strip shows. +pub(crate) fn paint_sync_card( + ui: &Rc, + icon: &str, + class: Option<&str>, + title: &str, + detail: Option<&str>, + paused: bool, + connected: bool, +) { + let card = &ui.locations.card; + card.icon.set_icon_name(Some(icon)); + for c in ["success", "warning", "error"] { + card.icon.remove_css_class(c); + } + if let Some(class) = class { + card.icon.add_css_class(class); + } + card.row.set_title(title); + card.row.set_subtitle(detail.unwrap_or_default()); + card.paused.set(paused); + card.pause + .set_label(if paused { "Resume" } else { "Pause" }); + card.pause.set_tooltip_text(Some(if paused { + "Upload everything that waited while sync was paused" + } else { + "Stop uploading until you resume. Files still open as usual." + })); + if paused { + card.pause.add_css_class("suggested-action"); + } else { + card.pause.remove_css_class("suggested-action"); + } + card.pause.set_sensitive(connected); +} + +/// Pause/Resume on the card: the button flips, the menu pauses for a while. +fn wire_pause(ui: &Rc) { + let ui_click = ui.clone(); + ui.locations.card.pause.connect_clicked(move |_| { + let resume = ui_click.locations.card.paused.get(); + set_sync_paused(&ui_click, !resume, None); + }); + let actions = gio::SimpleActionGroup::new(); + let pause_for = gio::SimpleAction::new("pause-for", Some(glib::VariantTy::INT64)); + let ui_for = ui.clone(); + pause_for.connect_activate(move |_, param| { + let Some(secs) = param.and_then(|p| p.get::()) else { + return; + }; + let until = glib::DateTime::now_utc() + .map(|now| now.to_unix()) + .unwrap_or_default() + + secs; + set_sync_paused(&ui_for, true, Some(until)); + }); + actions.add_action(&pause_for); + ui.locations + .card + .pause + .insert_action_group("sync", Some(&actions)); +} + +fn set_sync_paused(ui: &Rc, paused: bool, until: Option) { + ui.locations.card.pause.set_sensitive(false); + let rx = spawn_request( + ui.dirs.control_socket(), + Request::SetSyncPaused { paused, until }, + ); + let ui = ui.clone(); + glib::spawn_future_local(async move { + let what = if paused { + "Couldn't pause sync" + } else { + "Couldn't resume sync" + }; + match rx.recv().await { + Ok(Ok(Response::Ok { .. })) => toast( + &ui, + match (paused, until) { + (false, _) => "Sync resumed", + (true, None) => "Sync paused until you resume", + (true, Some(_)) => "Sync paused", + }, + ), + Ok(Ok(Response::Error { message, kind })) => toast_failure(&ui, what, &message, kind), + _ => toast_error(&ui, what, "The mount service didn't respond."), + } + ui.locations.card.pause.set_sensitive(true); + refresh_status(&ui); + }); +} + +/// The queue list, hidden while nothing waits. +fn build_queue_group() -> (adw::PreferencesGroup, gtk4::Button) { + let retry_all = gtk4::Button::builder() + .label("Retry All") + .valign(gtk4::Align::Center) + .tooltip_text("Try every failed change again now") + .visible(false) + .build(); + retry_all.add_css_class("flat"); + let group = adw::PreferencesGroup::builder() + .title("Waiting to Upload") + .description("Changes made on this computer that are not on Proton Drive yet.") + .header_suffix(&retry_all) + .visible(false) + .build(); + (group, retry_all) +} + +/// Most queue rows shown at once. The queue can hold thousands after a big copy; +/// the first screenful says what is going on, and the count says the rest. +const QUEUE_ROWS_SHOWN: usize = 50; + +/// Poll the pending-op queue while the Sync page is on screen. +pub(crate) fn refresh_queue(ui: &Rc) { + if ui.locations.queue.inflight.get() { + return; + } + ui.locations.queue.inflight.set(true); + let rx = spawn_request(ui.dirs.control_socket(), Request::ListPendingOps); + let ui = ui.clone(); + glib::spawn_future_local(async move { + let result = rx.recv().await; + ui.locations.queue.inflight.set(false); + match result { + Ok(Ok(Response::PendingOps { items })) => repaint_queue(&ui, &items), + // An older daemon without the request, or none at all: say nothing + // rather than show a list that cannot be kept current. + _ => repaint_queue(&ui, &[]), + } + }); +} + +fn repaint_queue(ui: &Rc, items: &[PendingOpInfo]) { + let queue = &ui.locations.queue; + // Stuck first — they are why the user is here — then oldest first. + let mut items: Vec<&PendingOpInfo> = items.iter().collect(); + items.sort_by_key(|op| (!op.failing, op.attempts == 0, op.id)); + let key: Vec = items + .iter() + .map(|op| (op.id, op.attempts, op.next_attempt_at, op.parked)) + .collect(); + if *queue.painted.borrow() == key { + return; + } + *queue.painted.borrow_mut() = key; + + for row in queue.rows.borrow_mut().drain(..) { + queue.group.remove(&row); + } + queue.group.set_visible(!items.is_empty()); + queue + .retry_all + .set_visible(items.iter().any(|op| op.attempts > 0 && !op.parked)); + let hidden = items.len().saturating_sub(QUEUE_ROWS_SHOWN); + queue.group.set_description(Some(&if hidden > 0 { + format!( + "Changes made on this computer that are not on Proton Drive yet. \ + Showing the first {QUEUE_ROWS_SHOWN} of {}.", + items.len() + ) + } else { + "Changes made on this computer that are not on Proton Drive yet.".to_string() + })); + let now = glib::DateTime::now_utc() + .map(|now| now.to_unix()) + .unwrap_or_default(); + let mut rows = queue.rows.borrow_mut(); + for op in items.into_iter().take(QUEUE_ROWS_SHOWN) { + let row = queue_row(ui, op, now); + queue.group.add(&row); + rows.push(row); + } +} + +fn queue_row(ui: &Rc, op: &PendingOpInfo, now: i64) -> adw::ActionRow { + let (icon, action) = match op.kind.as_str() { + "revision" => ("pdfs-upload-symbolic", "Upload changes"), + "create" => ("pdfs-upload-symbolic", "Upload new file"), + "mkdir" => ("folder-new-symbolic", "Create folder"), + "rename" => ("document-edit-symbolic", "Rename or move"), + "trash" => ("user-trash-symbolic", "Move to trash"), + _ => ("emblem-synchronizing-symbolic", "Change"), + }; + let state = match (op.parked, op.next_attempt_at) { + (true, _) => "waiting for the app writing it to finish".to_string(), + (false, Some(at)) if at > now && op.attempts > 0 => { + format!("retrying {}", clock_time(at)) + } + (false, _) if ui.locations.card.paused.get() => "waiting for sync to resume".to_string(), + (false, _) => "up next".to_string(), + }; + let mut subtitle = format!("{action} · {state}"); + if let Some(error) = &op.last_error { + subtitle.push_str(&format!( + "\nFailed {}: {error}", + count_noun(op.attempts.max(0) as usize, "time", "times") + )); + } + let row = adw::ActionRow::builder() + .title(glib::markup_escape_text(&op.path).as_str()) + .subtitle(glib::markup_escape_text(&subtitle).as_str()) + .title_lines(1) + .subtitle_lines(3) + .tooltip_text(&op.path) + .build(); + let image = gtk4::Image::from_icon_name(if op.failing { + "dialog-warning-symbolic" + } else { + icon + }); + if op.failing { + image.add_css_class("error"); + } + row.add_prefix(&image); + let waiting = !op.parked && op.attempts > 0 && op.next_attempt_at.is_some_and(|at| at > now); + if waiting { + let retry = gtk4::Button::builder() + .icon_name("view-refresh-symbolic") + .tooltip_text("Retry now") + .valign(gtk4::Align::Center) + .build(); + retry.add_css_class("flat"); + let ui = ui.clone(); + let id = op.id; + retry.connect_clicked(move |_| retry_queued(&ui, Some(id))); + row.add_suffix(&retry); + } + row +} + +/// Ask the daemon to stop waiting out one op's backoff, or every failed op's. +fn retry_queued(ui: &Rc, id: Option) { + let rx = spawn_request(ui.dirs.control_socket(), Request::RetryPendingOp { id }); + let ui = ui.clone(); + glib::spawn_future_local(async move { + match rx.recv().await { + Ok(Ok(Response::Ok { .. })) => { + toast(&ui, "Retrying now…"); + ui.locations.queue.painted.borrow_mut().clear(); + refresh_queue(&ui); + } + Ok(Ok(Response::Error { message, kind })) => { + toast_failure(&ui, "Couldn't retry", &message, kind) + } + _ => toast_error(&ui, "Couldn't retry", "The mount service didn't respond."), + } + }); } /// Show a status page in place of the locations list. diff --git a/crates/pdfs-gui/src/app/pages/status.rs b/crates/pdfs-gui/src/app/pages/status.rs index 1971aa1..0cb6f02 100644 --- a/crates/pdfs-gui/src/app/pages/status.rs +++ b/crates/pdfs-gui/src/app/pages/status.rs @@ -615,7 +615,10 @@ pub(crate) fn refresh(ui: &Rc) { // keeps this from asking more than once a minute. refresh_quota(ui); match ui.stack.visible_child_name().as_deref() { - Some("locations") => refresh_locations(ui), + Some("locations") => { + refresh_locations(ui); + refresh_queue(ui); + } Some("activity") => refresh_activity(ui), _ => {} } @@ -940,6 +943,8 @@ pub(crate) fn refresh_status(ui: &Rc) { pending_changes, failing_ops, failing_error, + paused, + paused_until, .. })) => { set_mounted(&ui, true); @@ -947,7 +952,12 @@ pub(crate) fn refresh_status(ui: &Rc) { // in it: it is why a file that looks saved is not on the remote // yet, and offline is usually the reason it is still queued. let queued = pending_summary(pending_uploads, pending_changes); - let state = if failing_ops > 0 { + let state = if paused { + SyncState::Paused { + until: paused_until, + queued, + } + } else if failing_ops > 0 { SyncState::Attention { count: failing_ops, error: failing_error, @@ -994,6 +1004,13 @@ pub(crate) fn refresh_status(ui: &Rc) { /// What the sidebar's status strip reports, most urgent first. pub(crate) enum SyncState { + /// The user paused syncing; nothing goes up until it resumes. Outranks + /// everything else, because it is the reason for everything else. + Paused { + /// Unix second the pause ends by itself; `None` until resumed. + until: Option, + queued: Option, + }, /// Operations keep failing; the user may have to act. Attention { count: u64, @@ -1014,9 +1031,23 @@ pub(crate) enum SyncState { Disconnected, } -/// Paint the sidebar status strip: icon, one-line state, and a detail line. +/// Paint the sidebar status strip and the Sync page's status card: icon, +/// one-line state, and a detail line. fn paint_sync_status(ui: &Rc, state: SyncState) { + let paused = matches!(state, SyncState::Paused { .. }); + let connected = !matches!(state, SyncState::Disconnected); let (icon, class, title, detail) = match state { + SyncState::Paused { until, queued } => ( + "media-playback-pause-symbolic", + Some("warning"), + "Sync paused".to_string(), + Some(match (until, queued) { + (Some(until), Some(queued)) => format!("{queued} · resumes {}", clock_time(until)), + (Some(until), None) => format!("Resumes {}", clock_time(until)), + (None, Some(queued)) => queued, + (None, None) => "Until you resume".to_string(), + }), + ), SyncState::Attention { count, error } => ( "dialog-warning-symbolic", Some("error"), @@ -1066,6 +1097,35 @@ fn paint_sync_status(ui: &Rc, state: SyncState) { .status_detail .set_label(detail.as_deref().unwrap_or_default()); ui.status.status_detail.set_tooltip_text(detail.as_deref()); + paint_sync_card( + ui, + icon, + class, + &title, + detail.as_deref(), + paused, + connected, + ); +} + +/// "at 14:30" today, "Tue 09:00" within the week, a date beyond that. +pub(crate) fn clock_time(unix: i64) -> String { + let (Ok(at), Ok(now)) = ( + glib::DateTime::from_unix_local(unix), + glib::DateTime::now_local(), + ) else { + return "later".to_string(); + }; + let format = if at.ymd() == now.ymd() { + "at %H:%M" + } else if at.difference(&now).as_seconds() < 6 * 86_400 { + "%a %H:%M" + } else { + "%e %b" + }; + at.format(format) + .map(|s| s.trim().to_string()) + .unwrap_or_else(|_| "later".to_string()) } /// Render the pins group from `pins`, with the unpin buttons enabled only while a diff --git a/crates/pdfs-gui/src/main.rs b/crates/pdfs-gui/src/main.rs index 05b05f2..a22a8cd 100644 --- a/crates/pdfs-gui/src/main.rs +++ b/crates/pdfs-gui/src/main.rs @@ -39,6 +39,8 @@ struct DriveState { /// One-line sync summary of in-flight transfers, empty when idle. Shown as a /// disabled menu line under the status line. sync: String, + /// Whether the user has sync paused; decides Pause vs Resume. + paused: bool, } /// Summarise a work snapshot into one menu line, or empty when idle. A single @@ -82,6 +84,7 @@ fn sync_line(items: &[TransferItem], jobs: &[JobItem]) -> String { struct DriveTray { state: DriveState, + socket: PathBuf, } /// Ask the daemon for its status, falling back to the stored session so the @@ -94,15 +97,23 @@ fn poll_state(socket: &Path, default_mountpoint: &Path) -> DriveState { online, pending_uploads, pending_changes, + failing_ops, + paused, .. }) => DriveState { line: match (online, pending_summary(pending_uploads, pending_changes)) { + _ if paused => "Sync paused".to_string(), + _ if failing_ops > 0 => match failing_ops { + 1 => "1 change needs attention".to_string(), + n => format!("{n} changes need attention"), + }, (true, None) => format!("Mounted at {mountpoint} ({pinned} pinned)"), (true, Some(q)) => format!("Syncing — {q} ({pinned} pinned)"), (false, None) => format!("Offline — cached files only ({pinned} pinned)"), (false, Some(q)) => format!("Offline — {q}"), }, mounted: true, + paused, mountpoint: PathBuf::from(mountpoint), // Same daemon is up, so a cheap follow-up poll gives the sync line. sync: match send(socket, &Request::GetQueueStatus) { @@ -116,6 +127,7 @@ fn poll_state(socket: &Path, default_mountpoint: &Path) -> DriveState { mounted: true, mountpoint: default_mountpoint.to_path_buf(), sync: String::new(), + paused: false, }, // No daemon: describe login state instead so the menu is still useful. Err(_) => { @@ -129,6 +141,7 @@ fn poll_state(socket: &Path, default_mountpoint: &Path) -> DriveState { mounted: false, mountpoint: default_mountpoint.to_path_buf(), sync: String::new(), + paused: false, } } } @@ -209,6 +222,27 @@ impl Tray for DriveTray { } .into(), ); + // Pause holds uploads back but leaves the mount readable, so it + // sits apart from Disconnect, which takes the whole mount away. + let paused = self.state.paused; + items.push( + StandardItem { + label: if paused { "Resume Sync" } else { "Pause Sync" }.into(), + activate: Box::new(move |this: &mut Self| { + let request = Request::SetSyncPaused { + paused: !paused, + until: None, + }; + match send(&this.socket, &request) { + Ok(Response::Ok { .. }) => this.state.paused = !paused, + Ok(other) => tracing::warn!("pause sync: {other:?}"), + Err(e) => tracing::warn!("pause sync: {e}"), + } + }), + ..Default::default() + } + .into(), + ); // Stop the systemd service (clean SIGTERM → lazy unmount). The next // login or reboot brings it back; this is a deliberate disconnect. items.push( @@ -281,6 +315,7 @@ fn main() -> Result<()> { let tray = DriveTray { state: poll_state(&socket, &default_mountpoint), + socket: socket.clone(), }; // The tray autostarts with the session and can beat the desktop's SNI diff --git a/docs/CHANGELOG.md b/docs/CHANGELOG.md index 5586cd9..9f80e75 100644 --- a/docs/CHANGELOG.md +++ b/docs/CHANGELOG.md @@ -9,6 +9,25 @@ release ships. Migrations are forward-only — a database written by a newer cli refuse-to-open, not a downgrade, so rolling back a release means restoring the cache from scratch (user data in `staging/` and `recovery/` is never touched by this). +## [Unreleased] + +Schema: **30** (unchanged). + +### Added +- **Pause and resume sync.** The Sync page, the tray and `pdfs sync pause [--for 1h]` / + `pdfs sync resume` hold back queued uploads and mirror-folder reconciles. Reads through the mount + keep working, writes keep being accepted and staged, and a timed pause ends by itself. The pause + survives a restart. +- **See what is waiting to upload.** The Sync page lists the queue with each item's last error and + next retry, with Retry now per item and Retry All; `pdfs sync queue` and `pdfs sync retry [id]` + do the same from the CLI. Nothing in the queue is ever dropped from here. + +### Changed +- **New app shell.** Every page has its own header bar; the sidebar ends in a footer with sync + state, storage and the account; the Settings page became a Preferences dialog (Ctrl+,), with + transfers on the Sync page and developer details in About. The app follows the system accent + colour; the Proton purple accent is a preference. + ## [1.11.1] — 2026-09-19 Schema: **30** (unchanged).