implemented offline scaffold

This commit is contained in:
nvrl committed 2026-07-17 10:23:24 +02:00
1 parent bb9804498c
commit 68e07ed426
9 files changed
+2412 -275

No files matched your search

Generated
+34
View File
@@ -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",
+54 -6
View File
@@ -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<String>,
},
/// 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<String>) -> 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.
+133
View File
@@ -82,6 +82,34 @@ struct PinFile {
pins: BTreeMap<String, Pin>,
}
/// What a staged file actually contains, written beside it as `<name>.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 `<key>` blobs and `<key>.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<PathBuf> {
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 _};
+76 -6
View File
@@ -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<i64>,
/// 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<String>,
/// `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<Pin>,
/// 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();
+350 -9
View File
@@ -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<String>,
/// Serialized [`StagedWrite`](crate::cache::StagedWrite).
pub meta_json: Option<String>,
/// When the op was queued (ms since epoch).
pub created_at: i64,
pub attempts: i64,
pub last_error: Option<String>,
/// 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<String>,
/// `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<Option<Node>> {
let conn = self.conn.lock().unwrap();
let json: Option<String> = 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<Option<String>> {
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<String>)> {
let conn = self.conn.lock().unwrap();
let superseded: Option<String> = 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<Vec<PendingOp>> {
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::<std::result::Result<Vec<_>, _>>()?;
Ok(rows)
}
/// How many uploads are still waiting, for `Response::Status`.
pub fn pending_op_count(&self) -> Result<i64> {
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<Option<i64>> {
@@ -644,7 +807,7 @@ impl Db {
pub fn sync_folder_list(&self) -> Result<Vec<StoredSyncFolder>> {
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<Option<StoredSyncFolder>> {
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();
+1013 -150
View File
File diff suppressed because it is too large. Load diff
+498 -65
View File
@@ -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<String> 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<Outcome, String> {
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<String, LocalItem> = 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<String, NodeUid> = 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<NodeUid> = 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<String, i64> = 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<Pending> = 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<String> = 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<Outcome, String> {
let mut local: HashMap<String, LocalItem> = 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<Outcome, PassAbort> {
// 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<String, LocalItem> = HashMap::new();
self.walk_local(folder_id, local_root, local_root, &mut local)?;
let mut remote: HashMap<String, RemoteItem> = HashMap::new();
let mut remote_dirs: HashMap<String, NodeUid> = 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<String, LocalItem>,
) -> 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<String, NodeUid>,
) -> 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<String, RemoteItem>,
dirs: &mut HashMap<String, NodeUid>,
baseline: &HashMap<String, StoredSyncEntry>,
) -> 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<String, LocalItem>) -> 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,
+243 -37
View File
@@ -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<Option<Instant>>) -> 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<Ui>, buttons: &[&gtk4::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<Ui>) {
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<Ui>, scope: RefreshScope, load: fn(&Rc<Ui>)) {
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<Ui>, 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<Ui>) {
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<Ui>) {
});
}
/// 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<Ui>, 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(&gtk4::Image::from_icon_name("computer-symbolic"));
@@ -4637,21 +4769,21 @@ fn repaint_devices(ui: &Rc<Ui>, devices: &[DeviceInfo]) {
return;
}
let mut rows: Vec<gtk4::Widget> = 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(&gtk4::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<Ui>, 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<Ui>, folders: &[SyncFolderInfo]) {
for row in ui.devices_sync_rows.borrow_mut().drain(..) {
@@ -4694,9 +4838,14 @@ fn repaint_sync_folders(ui: &Rc<Ui>, 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<Ui>, folders: &[SyncFolderInfo]) {
row.add_prefix(&gtk4::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<Ui>, 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<Ui>, 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<Ui>, id: i64, mode: &str) {
}
/// Confirm, then stop syncing a folder. Offers to also delete the cloud copy.
fn prompt_remove_sync_folder(ui: &Rc<Ui>, id: i64, path: &str) {
fn prompt_remove_sync_folder(ui: &Rc<Ui>, 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<Ui>, uid: &str, current: &str) {
fn prompt_remove_device(ui: &Rc<Ui>, 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,
},
)
}
+11 -2
View File
@@ -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.