Compare commits

...
7 Commits
Author SHA1 Message Date
nvrl 172ec4b0b7 version 0.2.0 2026-06-30 06:23:36 +02:00
nvrl ec4f5ea32f fmt 2026-06-30 06:23:00 +02:00
nvrl 81a0ecce54 added block cache 2026-06-30 06:22:40 +02:00
nvrl cc2f370670 lazy 2fa 2026-06-30 06:01:36 +02:00
nvrl b3e8e8ab96 performance fixes 2026-06-30 05:56:21 +02:00
nvrl 8e8d42ad3b updated ux 2026-06-30 05:50:20 +02:00
nvrl 99e80a7206 implemented robust fuse mount 2026-06-30 05:43:14 +02:00
13 changed files with 1392 additions and 329 deletions

No files matched your search

Generated
+4 -4
View File
@@ -2772,7 +2772,7 @@ dependencies = [
[[package]]
name = "pdfs-cli"
version = "0.1.2"
version = "0.2.0"
dependencies = [
"anyhow",
"clap 4.6.1",
@@ -2787,7 +2787,7 @@ dependencies = [
[[package]]
name = "pdfs-core"
version = "0.1.2"
version = "0.2.0"
dependencies = [
"directories",
"keyring",
@@ -2803,7 +2803,7 @@ dependencies = [
[[package]]
name = "pdfs-fuse"
version = "0.1.2"
version = "0.2.0"
dependencies = [
"fuser",
"libc",
@@ -2816,7 +2816,7 @@ dependencies = [
[[package]]
name = "pdfs-gui"
version = "0.1.2"
version = "0.2.0"
dependencies = [
"anyhow",
"async-channel",
+1 -1
View File
@@ -3,7 +3,7 @@ resolver = "2"
members = ["crates/pdfs-core", "crates/pdfs-fuse", "crates/pdfs-cli", "crates/pdfs-gui"]
[workspace.package]
version = "0.1.2"
version = "0.2.0"
edition = "2024"
license = "MIT"
authors = ["Nils Pukropp <contact@narl.io>"]
+54 -2
View File
@@ -38,6 +38,13 @@ enum Command {
/// Mountpoint; defaults to ~/ProtonDrive.
mountpoint: Option<PathBuf>,
},
/// Run the auto-mount daemon: wait for login, mount, and keep it mounted
/// until stopped. This is what the systemd user service runs; you normally
/// don't invoke it by hand.
Daemon {
/// Mountpoint; defaults to ~/ProtonDrive.
mountpoint: Option<PathBuf>,
},
/// Keep a file on this device (download + cache it). Needs a running mount.
Pin {
/// File path, inside the mountpoint or relative to it.
@@ -88,6 +95,7 @@ fn main() -> Result<()> {
Command::Logout => cmd_logout(),
Command::Status => cmd_status(),
Command::Mount { mountpoint } => cmd_mount(mountpoint),
Command::Daemon { mountpoint } => cmd_daemon(mountpoint),
Command::Pin { path } => cmd_pin(path),
Command::Unpin { path } => cmd_unpin(path),
Command::Pins => cmd_pins(),
@@ -158,6 +166,50 @@ fn cmd_status() -> Result<()> {
}
fn cmd_mount(mountpoint: Option<PathBuf>) -> Result<()> {
mount_once(mountpoint)?;
Ok(())
}
/// Run the auto-mount daemon loop. Waits for a stored login, mounts, and keeps
/// the mount alive: a clean stop (SIGTERM via `systemctl --user stop`) exits 0;
/// an external unmount triggers a remount; errors back off and retry. This is
/// the entry point for the systemd user service.
fn cmd_daemon(mountpoint: Option<PathBuf>) -> Result<()> {
loop {
// Wait until a session is stored. The GUI enables this service on login,
// but the service may also start at boot before the user has logged in.
loop {
match auth::load() {
Ok(_) => break,
Err(pdfs_core::Error::NotLoggedIn) => {
tracing::info!("not logged in; waiting…");
std::thread::sleep(std::time::Duration::from_secs(3));
}
Err(e) => return Err(e.into()),
}
}
match mount_once(mountpoint.clone()) {
Ok(pdfs_fuse::MountOutcome::Shutdown) => {
tracing::info!("daemon stopping");
return Ok(());
}
Ok(pdfs_fuse::MountOutcome::Unmounted) => {
tracing::warn!("mount ended externally; remounting in 2s");
std::thread::sleep(std::time::Duration::from_secs(2));
}
Err(e) => {
tracing::error!(error = %e, "mount failed; retrying in 5s");
std::thread::sleep(std::time::Duration::from_secs(5));
}
}
}
}
/// One mount run: resume the session, mount, and block until the mount ends
/// (clean shutdown or external unmount). Returns the [`MountOutcome`] so the
/// daemon loop can decide whether to exit or remount.
fn mount_once(mountpoint: Option<PathBuf>) -> Result<pdfs_fuse::MountOutcome> {
let dirs = AppDirs::new()?;
dirs.ensure()?;
let mountpoint = mountpoint.unwrap_or_else(|| dirs.default_mountpoint());
@@ -230,8 +282,8 @@ fn cmd_mount(mountpoint: Option<PathBuf>) -> Result<()> {
tracing::warn!(error = %e, "failed to persist tokens on shutdown");
}
result.context("mount failed")?;
Ok(())
let outcome = result.context("mount failed")?;
Ok(outcome)
}
fn cmd_pin(path: PathBuf) -> Result<()> {
+210
View File
@@ -27,6 +27,11 @@ use sha2::{Digest, Sha256};
use crate::error::Result;
/// Block size for the on-demand block cache. Matches the SDK content block size
/// (`DEFAULT_BLOCK_SIZE`, 4 MiB) so each cached block maps to exactly one
/// `download_range` fetch with no straddling.
pub const BLOCK_SIZE: u64 = 1 << 22;
/// Validity tag stored alongside a cached blob. A blob is fresh only if both
/// fields still match the node's current metadata.
#[derive(Serialize, Deserialize, Clone, Copy, PartialEq, Eq)]
@@ -61,6 +66,13 @@ pub struct ContentCache {
/// Subdirectory holding cached thumbnails (`<key>.t<n>` + `.meta`). Kept out
/// of `content_dir` so the budget scan never sees thumbnail files.
thumb_dir: PathBuf,
/// Subdirectory holding on-demand block-cache files (`<key>.b<idx>` +
/// `.meta`). Kept out of `content_dir` so the whole-file budget scan never
/// sees them; blocks carry their own LRU budget.
block_dir: PathBuf,
/// Subdirectory for write-handle scratch files (disk-backed write buffers).
/// Emptied on open so a crashed run leaves no orphans.
scratch_dir: PathBuf,
/// JSON pin registry path.
pins_path: PathBuf,
/// Soft cap on total blob bytes. Exceeded only transiently: a `store`
@@ -78,12 +90,21 @@ impl ContentCache {
std::fs::create_dir_all(&content_dir)?;
let thumb_dir = content_dir.join("thumbs");
std::fs::create_dir_all(&thumb_dir)?;
let block_dir = content_dir.join("blocks");
std::fs::create_dir_all(&block_dir)?;
// Scratch holds disk-backed write buffers; a previous run's leftovers are
// worthless, so start clean.
let scratch_dir = content_dir.join("scratch");
let _ = std::fs::remove_dir_all(&scratch_dir);
std::fs::create_dir_all(&scratch_dir)?;
if let Some(parent) = pins_path.parent() {
std::fs::create_dir_all(parent)?;
}
Ok(Self {
content_dir,
thumb_dir,
block_dir,
scratch_dir,
pins_path,
max_bytes,
})
@@ -180,6 +201,123 @@ impl ContentCache {
Ok(())
}
fn block_blob(&self, uid: &NodeUid, idx: u64) -> PathBuf {
self.block_dir.join(format!("{}.b{idx}", Self::key(uid)))
}
fn block_meta(&self, uid: &NodeUid, idx: u64) -> PathBuf {
self.block_dir
.join(format!("{}.b{idx}.meta", Self::key(uid)))
}
/// Serve cached block `idx` (a [`BLOCK_SIZE`]-aligned chunk) of `uid`, or
/// `None` on miss/stale. Validated against `(mtime, size)` like a whole-file
/// blob, so a new revision (which bumps the mtime) is detected. Bumps the
/// block's mtime for LRU, best effort.
pub fn cached_block(&self, uid: &NodeUid, mtime: i64, size: u64, idx: u64) -> Option<Vec<u8>> {
let want = Meta { mtime, size };
let meta: Meta =
serde_json::from_slice(&std::fs::read(self.block_meta(uid, idx)).ok()?).ok()?;
if meta != want {
return None;
}
let mut f = std::fs::File::open(self.block_blob(uid, idx)).ok()?;
let _ = f.set_times(std::fs::FileTimes::new().set_modified(SystemTime::now()));
let mut buf = Vec::new();
f.read_to_end(&mut buf).ok()?;
Some(buf)
}
/// Store `bytes` as cached block `idx` of `uid`, tagged `(mtime, size)`.
/// Temp-file-then-rename like [`store`](Self::store); meta written last so a
/// crash mid-store fails validation. Enforces the block-cache LRU budget.
pub fn store_block(
&self,
uid: &NodeUid,
mtime: i64,
size: u64,
idx: u64,
bytes: &[u8],
) -> Result<()> {
let blob = self.block_blob(uid, idx);
let tmp = self
.block_dir
.join(format!("{}.b{idx}.tmp", Self::key(uid)));
std::fs::write(&tmp, bytes)?;
std::fs::rename(&tmp, &blob)?;
std::fs::write(
self.block_meta(uid, idx),
serde_json::to_vec(&Meta { mtime, size })?,
)?;
self.enforce_block_budget();
Ok(())
}
/// Evict least-recently-used block-cache files until the block dir fits
/// `max_bytes`. No-op when the cap is disabled (`0`). All blocks are
/// evictable — pinned files are served from whole-file blobs, never blocks.
fn enforce_block_budget(&self) {
if self.max_bytes == 0 {
return;
}
let mut total: u64 = 0;
let mut blobs: Vec<(PathBuf, u64, SystemTime)> = Vec::new();
let Ok(rd) = std::fs::read_dir(&self.block_dir) else {
return;
};
for entry in rd.flatten() {
let path = entry.path();
let Some(name) = path.file_name().and_then(|n| n.to_str()) else {
continue;
};
if name.ends_with(".meta") || name.ends_with(".tmp") {
continue;
}
let Ok(meta) = entry.metadata() else { continue };
if !meta.is_file() {
continue;
}
total += meta.len();
let atime = meta.modified().unwrap_or(SystemTime::UNIX_EPOCH);
blobs.push((path, meta.len(), atime));
}
if total <= self.max_bytes {
return;
}
blobs.sort_by_key(|(_, _, atime)| *atime);
for (path, len, _) in blobs {
if total <= self.max_bytes {
break;
}
let meta = path.with_file_name(format!(
"{}.meta",
path.file_name().unwrap_or_default().to_string_lossy()
));
let _ = std::fs::remove_file(&path);
let _ = std::fs::remove_file(meta);
total = total.saturating_sub(len);
}
}
/// Create a fresh, empty read-write scratch file for a disk-backed write
/// handle. Returns the open file and its path (for cleanup on release).
pub fn create_scratch(&self) -> Result<(std::fs::File, PathBuf)> {
use std::sync::atomic::{AtomicU64, Ordering};
static N: AtomicU64 = AtomicU64::new(0);
let path = self.scratch_dir.join(format!(
"w-{}-{}",
std::process::id(),
N.fetch_add(1, Ordering::Relaxed)
));
let file = std::fs::OpenOptions::new()
.create(true)
.truncate(true)
.read(true)
.write(true)
.open(&path)?;
Ok((file, path))
}
/// 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.
@@ -190,6 +328,19 @@ impl ContentCache {
let _ = std::fs::remove_file(self.thumb_blob(uid, ttype));
let _ = std::fs::remove_file(self.thumb_meta(uid, ttype));
}
// Drop every cached block (and its meta/tmp) for this uid.
let prefix = format!("{}.b", Self::key(uid));
if let Ok(rd) = std::fs::read_dir(&self.block_dir) {
for entry in rd.flatten() {
if entry
.file_name()
.to_str()
.is_some_and(|n| n.starts_with(&prefix))
{
let _ = std::fs::remove_file(entry.path());
}
}
}
}
fn thumb_blob(&self, uid: &NodeUid, ttype: i32) -> PathBuf {
@@ -559,4 +710,63 @@ mod tests {
c.evict(&u);
assert!(!c.is_cached(&u, 1, 3));
}
#[test]
fn block_store_then_read() {
let (c, _d) = cache();
let u = uid("a");
c.store_block(&u, 100, 4096, 0, b"block-zero").unwrap();
assert_eq!(c.cached_block(&u, 100, 4096, 0).unwrap(), b"block-zero");
// A different index is a separate cache entry, absent here.
assert!(c.cached_block(&u, 100, 4096, 1).is_none());
// A new revision (mtime/size bump) invalidates the block.
assert!(c.cached_block(&u, 101, 4096, 0).is_none());
assert!(c.cached_block(&u, 100, 5000, 0).is_none());
}
#[test]
fn evict_drops_blocks() {
let (c, _d) = cache();
let u = uid("a");
c.store_block(&u, 1, 8, 0, b"aaaa").unwrap();
c.store_block(&u, 1, 8, 1, b"bbbb").unwrap();
c.evict(&u);
assert!(c.cached_block(&u, 1, 8, 0).is_none());
assert!(c.cached_block(&u, 1, 8, 1).is_none());
}
#[test]
fn block_budget_evicts_lru() {
// Cap fits two 4-byte blocks but not three.
let (c, _d) = cache_capped(8);
let u = uid("a");
c.store_block(&u, 1, 12, 0, b"aaaa").unwrap();
std::thread::sleep(std::time::Duration::from_millis(10));
c.store_block(&u, 1, 12, 1, b"bbbb").unwrap();
// Touch block 0 so block 1 is the least-recently-used.
std::thread::sleep(std::time::Duration::from_millis(10));
assert!(c.cached_block(&u, 1, 12, 0).is_some());
std::thread::sleep(std::time::Duration::from_millis(10));
c.store_block(&u, 1, 12, 2, b"cccc").unwrap();
assert!(
c.cached_block(&u, 1, 12, 0).is_some(),
"recently-read survives"
);
assert!(c.cached_block(&u, 1, 12, 1).is_none(), "LRU block evicted");
assert!(c.cached_block(&u, 1, 12, 2).is_some(), "newest survives");
}
#[test]
fn scratch_file_is_writable_and_isolated() {
use std::io::{Read as _, Seek as _, SeekFrom, Write as _};
let (c, _d) = cache();
let (mut f1, p1) = c.create_scratch().unwrap();
let (_f2, p2) = c.create_scratch().unwrap();
assert_ne!(p1, p2, "each scratch file is unique");
f1.write_all(b"scratch").unwrap();
f1.seek(SeekFrom::Start(0)).unwrap();
let mut s = String::new();
f1.read_to_string(&mut s).unwrap();
assert_eq!(s, "scratch");
}
}
+4 -3
View File
@@ -57,9 +57,10 @@ impl AppDirs {
let path = self.config_path();
if path.exists()
&& let Ok(content) = std::fs::read_to_string(&path)
&& let Ok(config) = serde_json::from_str::<AppConfig>(&content) {
return config;
}
&& let Ok(config) = serde_json::from_str::<AppConfig>(&content)
{
return config;
}
let default_config = AppConfig::default();
if let Ok(content) = serde_json::to_string_pretty(&default_config) {
+9 -1
View File
@@ -85,11 +85,19 @@ pub struct PhotoItem {
/// The daemon's reply to a [`Request`].
#[derive(Serialize, Deserialize, Debug)]
pub enum Response {
/// Current mount status.
/// Current mount status. Carries the cache stats the daemon already holds
/// (`used`/`budget` bytes and the pin list) so a front-end never has to open
/// the on-disk cache itself on its UI thread.
Status {
username: String,
mountpoint: String,
pinned: usize,
/// Bytes of cached content blobs (see [`crate::cache::ContentCache::usage`]).
used: u64,
/// Configured soft byte cap (`0` = unlimited).
budget: u64,
/// The pin registry.
pins: Vec<Pin>,
},
/// A human-readable success message.
Ok { message: String },
+1
View File
@@ -7,6 +7,7 @@ pub mod cache;
pub mod config;
pub mod control;
pub mod error;
pub mod service;
pub use error::{Error, Result};
+86
View File
@@ -0,0 +1,86 @@
//! Thin wrappers over `systemctl --user` for the auto-mount daemon.
//!
//! The mount is owned by a systemd *user* service (`proton-drive.service`, see
//! `packaging/`). Front-ends never spawn `pdfs mount` directly anymore: logging
//! in enables+starts the unit, logging out disables+stops it, and the tray's
//! "Disconnect" just stops it. systemd then handles restart-on-failure and a
//! clean SIGTERM stop (which the daemon turns into a lazy unmount).
//!
//! Every call is best-effort: a failure is logged but not fatal, since the user
//! can always drive `systemctl --user` by hand.
use std::process::{Command, Stdio};
/// The systemd user unit that runs `pdfs daemon`.
pub const SERVICE: &str = "proton-drive.service";
/// Run one `systemctl --user <args> proton-drive.service`, returning whether it
/// succeeded. Output is suppressed; failures are logged.
fn systemctl(args: &[&str]) -> bool {
let ok = Command::new("systemctl")
.arg("--user")
.args(args)
.arg(SERVICE)
.stdout(Stdio::null())
.stderr(Stdio::null())
.status()
.map(|s| s.success())
.unwrap_or(false);
if !ok {
tracing::warn!(args = ?args, "systemctl --user {SERVICE} failed");
}
ok
}
/// Like [`systemctl`] but treats a non-zero exit as a plain `false` answer
/// rather than a failure to log. For query verbs (`is-active`, `is-failed`)
/// where "no" is the normal, expected result.
fn systemctl_query(args: &[&str]) -> bool {
Command::new("systemctl")
.arg("--user")
.args(args)
.arg(SERVICE)
.stdout(Stdio::null())
.stderr(Stdio::null())
.status()
.map(|s| s.success())
.unwrap_or(false)
}
/// Enable the service so it survives reboots/logins and start it now. Called on
/// successful login.
pub fn enable_start() -> bool {
systemctl(&["enable", "--now"])
}
/// Stop the running mount without disabling it. Called by the tray's
/// "Disconnect": the next login (or reboot) brings it back.
pub fn stop() -> bool {
systemctl(&["stop"])
}
/// Restart the unit now. Used by a front-end "Retry" action when the mount has
/// failed and the user wants to bring it back without re-logging-in.
pub fn restart() -> bool {
systemctl(&["restart"])
}
/// Whether the unit currently reports `active` (running, or starting up).
/// `systemctl is-active` exits 0 only while the unit is active, so a front-end
/// can use this to tell "service still coming up" from "service is down".
pub fn is_active() -> bool {
systemctl_query(&["is-active"])
}
/// Whether the unit is in the `failed` state. `systemctl is-failed` exits 0
/// only when the unit has failed, letting a front-end show an actionable error
/// (with Retry) rather than an indefinite "connecting…" spinner.
pub fn is_failed() -> bool {
systemctl_query(&["is-failed"])
}
/// Disable the service so it no longer auto-starts, and stop it now. Called on
/// logout.
pub fn disable_stop() -> bool {
systemctl(&["disable", "--now"])
}
+557 -114
View File
@@ -23,10 +23,19 @@
//! into the on-disk [`ContentCache`] and every later `read` is served from disk
//! instead of the network. Writes and remote events evict the cache so it never
//! goes stale.
//!
//! Reads of unpinned files no longer hit the network per call: [`Core::read_range`]
//! fetches and caches [`BLOCK_SIZE`]-aligned blocks, so sequential/sparse reads
//! reuse the on-disk block cache. Writes are disk-backed: each [`WriteHandle`]
//! stages authored bytes in a scratch file and tracks them with an [`Intervals`]
//! set, so a multi-GiB write never buffers in RAM and only the untouched
//! remainder of the file is pulled from the remote — lazily, at commit.
use std::collections::HashMap;
use std::ffi::OsStr;
use std::fs::File;
use std::io::{BufRead, BufReader, Write as _};
use std::os::unix::fs::FileExt;
use std::os::unix::net::{UnixListener, UnixStream};
use std::path::{Component, Path, PathBuf};
use std::sync::{Arc, Mutex};
@@ -39,7 +48,7 @@ use fuser::{
ReplyAttr, ReplyCreate, ReplyData, ReplyDirectory, ReplyEmpty, ReplyEntry, ReplyOpen,
ReplyWrite, Request, Session, TimeOrNow, WriteFlags,
};
use pdfs_core::cache::ContentCache;
use pdfs_core::cache::{BLOCK_SIZE, ContentCache};
use pdfs_core::control::{DirEntry, PhotoItem, Request as CtlRequest, Response as CtlResponse};
use proton_drive_rs::proton_sdk::ids::{DriveEventId, LinkId, NodeUid, VolumeId};
use proton_drive_rs::{
@@ -75,18 +84,90 @@ struct Entry {
node: Node,
}
/// Buffered state for a file opened for writing. The full plaintext is
/// assembled here and uploaded as one new revision on flush/release, because
/// the SDK seals whole revisions rather than byte ranges.
/// A set of non-overlapping `[start, end)` byte ranges, kept sorted and merged.
/// Tracks which bytes of a [`WriteHandle`]'s scratch file were authored locally
/// (vs. still living only in the remote base), so reads and the commit gap-fill
/// know which regions to pull from the network.
#[derive(Clone, Default)]
struct Intervals(Vec<(u64, u64)>);
impl Intervals {
/// Mark `[start, end)` as authored, coalescing with any touching ranges.
fn add(&mut self, start: u64, end: u64) {
if start >= end {
return;
}
self.0.push((start, end));
self.0.sort_by_key(|&(s, _)| s);
let mut merged: Vec<(u64, u64)> = Vec::with_capacity(self.0.len());
for &(s, e) in &self.0 {
match merged.last_mut() {
Some(last) if s <= last.1 => last.1 = last.1.max(e),
_ => merged.push((s, e)),
}
}
self.0 = merged;
}
/// Drop everything at or beyond `len` (a shrink/truncate).
fn clip(&mut self, len: u64) {
self.0.retain(|&(s, _)| s < len);
for iv in &mut self.0 {
iv.1 = iv.1.min(len);
}
}
/// Split `[start, end)` into contiguous `(s, e, authored)` segments, in
/// order. `authored == true` means the bytes live in the scratch file;
/// `false` means they must come from the remote base (or are a hole).
fn segments(&self, start: u64, end: u64) -> Vec<(u64, u64, bool)> {
let mut out = Vec::new();
let mut pos = start;
for &(s, e) in &self.0 {
if e <= start {
continue;
}
if s >= end {
break;
}
let ws = s.max(start);
let we = e.min(end);
if pos < ws {
out.push((pos, ws, false));
}
out.push((ws, we, true));
pos = we;
}
if pos < end {
out.push((pos, end, false));
}
out
}
}
/// State for a file opened for writing. Authored bytes are staged in an on-disk
/// scratch file (positional reads/writes) rather than RAM, so a multi-GiB write
/// never balloons the daemon. On flush/release the scratch file — gap-filled
/// from the remote base where untouched — is streamed up as one new revision,
/// since the SDK seals whole revisions rather than byte ranges.
struct WriteHandle {
ino: u64,
uid: NodeUid,
/// The plaintext being assembled; valid only once `seeded`.
buf: Vec<u8>,
/// Whether `buf` holds the file's current content (or was emptied by a
/// truncate). Until then a partial write must hydrate the base first.
seeded: bool,
/// Whether `buf` diverged from the remote and needs an upload.
/// Disk-backed staging buffer. Shared (`Arc`) so reads can use it without
/// holding the state lock across I/O. Accessed positionally (`read_at`/
/// `write_at`), so a clone never disturbs another's file offset.
file: Arc<File>,
/// Scratch file path, removed on release.
path: PathBuf,
/// Byte ranges authored into `file`. Everything else in `[0, len)` is base.
written: Intervals,
/// Logical file size (may exceed authored bytes after a truncate-extend).
len: u64,
/// Size of the remote base at open, for serving untouched ranges.
base_size: u64,
/// Modification time of the remote base at open, validating its block cache.
base_mtime: i64,
/// Whether anything diverged from the remote and needs an upload.
dirty: bool,
}
@@ -289,35 +370,183 @@ impl Core {
nodes.into_iter().next().ok_or(Errno::ENOENT)
}
/// Upload a write handle's buffer as a new revision if it is dirty, clearing
/// the dirty flag and refreshing the entry's size/mtime on success. No-op for
/// a clean (or unknown) handle. Network I/O happens without the lock held.
/// Serve bytes `[offset, offset + len)` of `uid`'s active revision, hitting
/// the on-disk caches before the network: a whole-file blob (pinned files)
/// first, then the block cache — fetching only the [`BLOCK_SIZE`]-aligned
/// blocks that overlap the request and caching each. `mtime`/`fsize` validate
/// both caches. Network I/O runs without any lock held.
fn read_range(
&self,
uid: &NodeUid,
mtime: i64,
fsize: u64,
offset: u64,
len: u64,
) -> Result<Vec<u8>, Errno> {
if let Some(bytes) = self.cache.read_range(uid, mtime, fsize, offset, len) {
return Ok(bytes);
}
if offset >= fsize || len == 0 {
return Ok(Vec::new());
}
let end = offset.saturating_add(len).min(fsize);
let mut out = Vec::with_capacity((end - offset) as usize);
let first = offset / BLOCK_SIZE;
let last = (end - 1) / BLOCK_SIZE;
for bidx in first..=last {
let bstart = bidx * BLOCK_SIZE;
let block = match self.cache.cached_block(uid, mtime, fsize, bidx) {
Some(b) => b,
None => {
let blen = BLOCK_SIZE.min(fsize - bstart);
let bytes = self
.rt
.block_on(self.client.download_range(uid, bstart, blen))
.map_err(|e| {
warn!(%uid, bstart, blen, error = %e, "download_range failed");
Errno::EIO
})?;
let _ = self.cache.store_block(uid, mtime, fsize, bidx, &bytes);
bytes
}
};
let s = (offset.max(bstart) - bstart) as usize;
let e = (end.min(bstart + block.len() as u64) - bstart) as usize;
if s < e {
out.extend_from_slice(&block[s..e]);
}
}
Ok(out)
}
/// Serve a read against an open write handle: stitch authored ranges (from
/// the scratch file) and untouched ranges (from the remote base via the
/// block cache) in order, zero-filling any region past the base.
#[allow(clippy::too_many_arguments)]
fn serve_open_read(
&self,
file: &Arc<File>,
len: u64,
uid: &NodeUid,
base_mtime: i64,
base_size: u64,
written: &Intervals,
offset: u64,
size: u64,
) -> Result<Vec<u8>, Errno> {
if offset >= len || size == 0 {
return Ok(Vec::new());
}
let end = offset.saturating_add(size).min(len);
let mut out = Vec::with_capacity((end - offset) as usize);
for (s, e, authored) in written.segments(offset, end) {
if authored {
let mut buf = vec![0u8; (e - s) as usize];
file.read_exact_at(&mut buf, s).map_err(|err| {
warn!(%uid, error = %err, "scratch read failed");
Errno::EIO
})?;
out.extend_from_slice(&buf);
} else {
let bend = e.min(base_size);
if s < bend {
out.extend_from_slice(&self.read_range(
uid,
base_mtime,
base_size,
s,
bend - s,
)?);
}
// Anything past the base is a hole: zero-fill.
let zeros = e.saturating_sub(s.max(base_size));
out.resize(out.len() + zeros as usize, 0);
}
}
Ok(out)
}
/// Upload a write handle's scratch file as a new revision if it is dirty.
/// Untouched bytes within the base are filled from the remote first (reusing
/// the block cache), so a partial overwrite never had to pre-download the
/// whole file. On success the handle's base is advanced to the just-sealed
/// revision and `written` cleared, so later reads of untouched regions see
/// the new content. No-op for a clean (or unknown) handle. Network I/O runs
/// without the lock held.
fn commit(&self, fh: u64) -> Result<(), Errno> {
let (uid, buf, ino) = {
let (uid, file, path, len, base_mtime, base_size, written, ino) = {
let st = self.state.lock().unwrap();
match st.handles.get(&fh) {
Some(h) if h.dirty => (h.uid.clone(), h.buf.clone(), h.ino),
Some(h) if h.dirty => (
h.uid.clone(),
h.file.clone(),
h.path.clone(),
h.len,
h.base_mtime,
h.base_size,
h.written.clone(),
h.ino,
),
_ => return Ok(()),
}
};
// Materialize the full new content in the scratch file: ensure its length,
// then fill every untouched range that overlaps the base with base bytes.
file.set_len(len).map_err(|e| {
error!(%uid, error = %e, "resize scratch file failed");
Errno::EIO
})?;
for (s, e, authored) in written.segments(0, len) {
if authored {
continue;
}
let bend = e.min(base_size);
if s >= bend {
continue; // wholly past the base: already zero-filled on disk
}
let bytes = self.read_range(&uid, base_mtime, base_size, s, bend - s)?;
file.write_all_at(&bytes, s).map_err(|err| {
error!(%uid, error = %err, "scratch gap-fill write failed");
Errno::EIO
})?;
}
// Stream the scratch file up as one revision (fresh handle reads from 0).
let reader = File::open(&path).map_err(|e| {
error!(%uid, error = %e, "reopen scratch file failed");
Errno::EIO
})?;
self.rt
.block_on(self.client.upload_new_revision(&uid, &buf))
.block_on(self.client.upload_new_revision_from(
&uid,
reader,
len as i64,
Vec::new(),
None,
))
.map_err(|e| {
error!(%uid, error = %e, "upload new revision failed");
Errno::EIO
})?;
let now = now_secs();
let mut st = self.state.lock().unwrap();
if let Some(h) = st.handles.get_mut(&fh) {
h.dirty = false;
{
let mut st = self.state.lock().unwrap();
if let Some(h) = st.handles.get_mut(&fh) {
h.dirty = false;
// The scratch file now equals the sealed revision; treat it all as
// base so further reads of untouched bytes hit the new content.
h.written = Intervals::default();
h.base_mtime = now;
h.base_size = len;
}
st.set_size(ino, len);
st.touch_mtime(ino, now);
}
st.set_size(ino, buf.len() as u64);
st.touch_mtime(ino, now);
drop(st);
// The sealed content differs from any cached blob; if the file is
// pinned, refresh the cache so reads stay served from disk.
// The sealed content differs from any cached blob/blocks; refresh a pinned
// file's whole-file blob, otherwise evict so reads re-fetch fresh.
if self.cache.is_pinned(&uid) {
let _ = self.cache.store(&uid, now, buf.len() as u64, &buf);
if let Ok(bytes) = std::fs::read(&path) {
let _ = self.cache.store(&uid, now, len, &bytes);
}
} else {
self.cache.evict(&uid);
}
@@ -821,10 +1050,12 @@ impl Filesystem for ProtonFs {
}
fn open(&self, _req: &Request, ino: INodeNo, flags: OpenFlags, reply: ReplyOpen) {
let uid = {
let (uid, base_mtime, base_size) = {
let st = self.core.state.lock().unwrap();
match st.entries.get(&ino.0) {
Some(e) if e.node.is_file() => e.uid.clone(),
Some(e) if e.node.is_file() => {
(e.uid.clone(), e.node.modification_time, node_size(&e.node))
}
Some(_) => {
reply.error(Errno::EISDIR);
return;
@@ -835,12 +1066,21 @@ impl Filesystem for ProtonFs {
}
}
};
// Read-only opens stay stateless (fh 0). A write open allocates a buffer
// handle, hydrated lazily on the first write that needs the base content.
// Read-only opens stay stateless (fh 0). A write open allocates a
// disk-backed handle; bytes are authored into a scratch file and the
// untouched remainder is pulled from the base lazily (on read / commit).
if flags.acc_mode() == OpenAccMode::O_RDONLY {
reply.opened(FileHandle(0), FopenFlags::empty());
return;
}
let (file, path) = match self.core.cache.create_scratch() {
Ok(x) => x,
Err(e) => {
error!(%uid, error = %e, "create scratch file failed");
reply.error(Errno::EIO);
return;
}
};
let mut st = self.core.state.lock().unwrap();
let fh = st.next_fh;
st.next_fh += 1;
@@ -849,8 +1089,14 @@ impl Filesystem for ProtonFs {
WriteHandle {
ino: ino.0,
uid,
buf: Vec::new(),
seeded: false,
file: Arc::new(file),
path,
written: Intervals::default(),
// Starts at the current size; reads in [0, base_size) come from
// the base until overwritten.
len: base_size,
base_size,
base_mtime,
dirty: false,
},
);
@@ -868,16 +1114,40 @@ impl Filesystem for ProtonFs {
_lock_owner: Option<LockOwner>,
reply: ReplyData,
) {
// A file open for writing is served from its handle so reads see the
// in-flight (possibly unsaved) content: authored bytes come from the
// scratch file, untouched bytes from the base.
let handle = {
let st = self.core.state.lock().unwrap();
st.handles.values().find(|h| h.ino == ino.0).map(|h| {
(
h.file.clone(),
h.len,
h.uid.clone(),
h.base_mtime,
h.base_size,
h.written.clone(),
)
})
};
if let Some((file, len, uid, base_mtime, base_size, written)) = handle {
match self.core.serve_open_read(
&file,
len,
&uid,
base_mtime,
base_size,
&written,
offset,
size as u64,
) {
Ok(bytes) => reply.data(&bytes),
Err(e) => reply.error(e),
}
return;
}
let (uid, mtime, fsize) = {
let st = self.core.state.lock().unwrap();
// If the file is open for writing and its buffer is populated, serve
// from there so reads see the in-flight (possibly unsaved) content.
if let Some(h) = st.handles.values().find(|h| h.ino == ino.0 && h.seeded) {
let off = (offset as usize).min(h.buf.len());
let end = (off + size as usize).min(h.buf.len());
reply.data(&h.buf[off..end]);
return;
}
match st.entries.get(&ino.0) {
Some(e) if e.node.is_file() => {
(e.uid.clone(), e.node.modification_time, node_size(&e.node))
@@ -892,25 +1162,12 @@ impl Filesystem for ProtonFs {
}
}
};
// Pinned (or otherwise cached) files are served straight from disk.
if let Some(bytes) = self
.core
.cache
.read_range(&uid, mtime, fsize, offset, size as u64)
{
reply.data(&bytes);
return;
}
match self
.core
.rt
.block_on(self.core.client.download_range(&uid, offset, size as u64))
.read_range(&uid, mtime, fsize, offset, size as u64)
{
Ok(bytes) => reply.data(&bytes),
Err(e) => {
warn!(%uid, offset, size, error = %e, "download_range failed");
reply.error(Errno::EIO);
}
Err(e) => reply.error(e),
}
}
@@ -962,6 +1219,14 @@ impl Filesystem for ProtonFs {
return;
}
};
let (file, path) = match self.core.cache.create_scratch() {
Ok(x) => x,
Err(e) => {
error!(%new_uid, error = %e, "create scratch file failed");
reply.error(Errno::EIO);
return;
}
};
let mut st = self.core.state.lock().unwrap();
let ino = st.intern(parent, node);
if let Some(kids) = st.children.get_mut(&parent)
@@ -973,11 +1238,16 @@ impl Filesystem for ProtonFs {
st.next_fh += 1;
st.handles.insert(
fh,
// A brand-new file: empty base, everything written is authored.
WriteHandle {
ino,
uid: new_uid,
buf: Vec::new(),
seeded: true,
file: Arc::new(file),
path,
written: Intervals::default(),
len: 0,
base_size: 0,
base_mtime: now_secs(),
dirty: false,
},
);
@@ -1004,60 +1274,34 @@ impl Filesystem for ProtonFs {
reply: ReplyWrite,
) {
let fh = fh.0;
// Partial writes need the file's current content; hydrate it once.
let need_seed = {
// Stage the bytes straight into the scratch file (no base download): only
// the untouched remainder is pulled from the remote, and only at commit.
let file = {
let st = self.core.state.lock().unwrap();
match st.handles.get(&fh) {
Some(h) => !h.seeded,
Some(h) => h.file.clone(),
None => {
reply.error(Errno::EBADF);
return;
}
}
};
if need_seed {
let uid = self
.core
.state
.lock()
.unwrap()
.handles
.get(&fh)
.map(|h| h.uid.clone());
let Some(uid) = uid else {
reply.error(Errno::EBADF);
return;
};
let base = match self.core.rt.block_on(self.core.client.download_file(&uid)) {
Ok(b) => b,
Err(e) => {
warn!(%uid, error = %e, "seed write buffer failed");
reply.error(Errno::EIO);
return;
}
};
let mut st = self.core.state.lock().unwrap();
if let Some(h) = st.handles.get_mut(&fh)
&& !h.seeded
{
h.buf = base;
h.seeded = true;
}
if let Err(e) = file.write_all_at(data, offset) {
error!(ino = ino.0, fh, error = %e, "scratch write failed");
reply.error(Errno::EIO);
return;
}
let off = offset as usize;
let new_len = {
let mut st = self.core.state.lock().unwrap();
let Some(h) = st.handles.get_mut(&fh) else {
reply.error(Errno::EBADF);
return;
};
let end = off + data.len();
if h.buf.len() < end {
h.buf.resize(end, 0);
}
h.buf[off..end].copy_from_slice(data);
let end = offset + data.len() as u64;
h.written.add(offset, end);
h.len = h.len.max(end);
h.dirty = true;
let len = h.buf.len() as u64;
let len = h.len;
st.set_size(ino.0, len);
len
};
@@ -1067,7 +1311,7 @@ impl Filesystem for ProtonFs {
offset,
len = data.len(),
new_len,
"buffered write"
"staged write"
);
reply.written(data.len() as u32);
}
@@ -1100,8 +1344,16 @@ impl Filesystem for ProtonFs {
let mut st = self.core.state.lock().unwrap();
match st.handles.get_mut(&fh) {
Some(h) => {
h.buf.resize(size as usize, 0);
h.seeded = true;
if size < h.len {
// Shrink: drop authored ranges past the new end.
h.written.clip(size);
} else if size > h.len {
// Grow: the new tail is defined as zeros, so claim
// it as authored rather than base content.
h.written.add(h.len, size);
}
let _ = h.file.set_len(size);
h.len = size;
h.dirty = true;
true
}
@@ -1207,7 +1459,10 @@ impl Filesystem for ProtonFs {
reply: ReplyEmpty,
) {
let res = self.core.commit(fh.0);
self.core.state.lock().unwrap().handles.remove(&fh.0);
let handle = self.core.state.lock().unwrap().handles.remove(&fh.0);
if let Some(h) = handle {
let _ = std::fs::remove_file(&h.path);
}
match res {
Ok(()) => reply.ok(),
Err(e) => reply.error(e),
@@ -1571,11 +1826,17 @@ fn handle_control_conn(core: &Core, username: &str, mountpoint: &Path, stream: U
return;
}
let response = match serde_json::from_str::<CtlRequest>(line.trim()) {
Ok(CtlRequest::Status) => CtlResponse::Status {
username: username.to_string(),
mountpoint: mountpoint.display().to_string(),
pinned: core.cache.list_pins().len(),
},
Ok(CtlRequest::Status) => {
let pins = core.cache.list_pins();
CtlResponse::Status {
username: username.to_string(),
mountpoint: mountpoint.display().to_string(),
pinned: pins.len(),
used: core.cache.usage(),
budget: core.cache.budget(),
pins,
}
}
Ok(CtlRequest::Pin { path }) => match rel_to_mount(mountpoint, &path) {
Ok(rel) => match core.pin(&rel) {
Ok(name) => CtlResponse::Ok {
@@ -1667,12 +1928,53 @@ fn run_control_socket(core: Core, username: String, mountpoint: PathBuf, listene
}
}
/// Mount the filesystem at `mountpoint` and block until it is unmounted.
/// Why a [`mount`] call returned. Lets the daemon decide whether to exit (clean
/// shutdown) or remount (the mount went away under it).
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum MountOutcome {
/// The daemon was asked to stop (SIGTERM/SIGINT) and we lazily unmounted
/// ourselves. The caller should exit cleanly.
Shutdown,
/// The kernel mount ended on its own (e.g. an external `fusermount -u`).
/// The caller may want to remount.
Unmounted,
}
/// Best-effort teardown of a stale mount left behind by a crashed daemon. A
/// previous run that died without unmounting leaves the kernel mount in place,
/// so the fresh `Session::new` below would fail with EBUSY ("Device or resource
/// busy"). `fusermount3 -u -z` is the lazy (detach) unmount, which succeeds even
/// when the old mount is still busy. Swallow all output/errors: if there is no
/// stale mount this is simply a no-op.
fn clear_stale_mount(mountpoint: &Path) {
for bin in ["fusermount3", "fusermount"] {
let ok = std::process::Command::new(bin)
.arg("-u")
.arg("-z")
.arg(mountpoint)
.stdout(std::process::Stdio::null())
.stderr(std::process::Stdio::null())
.status()
.map(|s| s.success())
.unwrap_or(false);
if ok {
info!(mountpoint = %mountpoint.display(), "cleared stale mount before remount");
return;
}
}
}
/// Mount the filesystem at `mountpoint` and block until it is unmounted or the
/// daemon is asked to stop.
///
/// Fetches the My Files root up front (so an auth/network failure surfaces
/// before the kernel mount), then spawns the Phase 2 event-sync task, the
/// Phase 4 control socket, and takes over the calling thread in the FUSE session
/// loop. `rt` must be a handle to a *running* multi-threaded runtime.
/// Phase 4 control socket, and runs the FUSE session on its own thread while
/// this thread waits for either a stop signal (SIGTERM/SIGINT) or the kernel
/// mount ending on its own. On a stop signal we lazily unmount ourselves
/// (`umount_and_join`, the MNT_DETACH path that succeeds even while downloads
/// are in flight), so `systemctl --user stop` is always a clean teardown.
/// `rt` must be a handle to a *running* multi-threaded runtime.
pub fn mount(
client: ProtonDriveClient,
rt: tokio::runtime::Handle,
@@ -1680,7 +1982,7 @@ pub fn mount(
cache: ContentCache,
control_socket: &Path,
username: String,
) -> std::io::Result<()> {
) -> std::io::Result<MountOutcome> {
let root = rt
.block_on(client.get_my_files_folder())
.map_err(|e| std::io::Error::other(format!("fetch My Files root: {e}")))?;
@@ -1707,6 +2009,10 @@ pub fn mount(
MountOption::Subtype("protondrive".to_string()),
MountOption::DefaultPermissions,
];
// A crashed previous run can leave the kernel mount in place, which makes
// the fresh mount below fail with EBUSY. Lazily detach any leftover first.
clear_stale_mount(mountpoint);
info!(mountpoint = %mountpoint.display(), "mounting Proton Drive");
// Bind the control socket before the FUSE session takes over the thread. A
@@ -1725,16 +2031,153 @@ pub fn mount(
let fs = ProtonFs::new(core.clone(), root);
// Build the session explicitly (not `mount2`) so we can grab a `Notifier`
// for the event task. `spawn` runs the session loop on its own thread;
// `join` then blocks here until the filesystem is unmounted, preserving
// `mount`'s blocking contract.
let session = Session::new(fs, mountpoint, &config)?.spawn()?;
let notifier = session.notifier();
// for the event task. `spawn` runs the session loop on its own background
// thread; we then wait here for either a stop signal or the mount ending.
let bg = Session::new(fs, mountpoint, &config)?.spawn()?;
let notifier = bg.notifier();
rt.spawn(run_event_sync(
client, scope, core.state, core.cache, notifier,
));
let result = session.join();
// Stop signals (SIGTERM from `systemctl --user stop`, SIGINT from Ctrl-C)
// are delivered onto the async runtime; bridge them onto a sync channel so
// the loop below can react without blocking a worker thread. A bounded
// channel of 1 is enough — we only need to know that *a* stop arrived.
let (sig_tx, sig_rx) = std::sync::mpsc::sync_channel::<()>(1);
rt.spawn(async move {
let mut sigterm =
match tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate()) {
Ok(s) => s,
Err(e) => {
warn!(error = %e, "install SIGTERM handler failed");
return;
}
};
let mut sigint =
match tokio::signal::unix::signal(tokio::signal::unix::SignalKind::interrupt()) {
Ok(s) => s,
Err(e) => {
warn!(error = %e, "install SIGINT handler failed");
return;
}
};
tokio::select! {
_ = sigterm.recv() => info!("received SIGTERM"),
_ = sigint.recv() => info!("received SIGINT"),
}
let _ = sig_tx.try_send(());
});
// Wait for whichever happens first: a stop signal (→ we unmount ourselves
// via the lazy MNT_DETACH path, clean even mid-download), or the kernel
// mount ending on its own (→ the session thread finishes). Poll instead of
// blocking on `join` so we can also notice the signal.
let outcome = loop {
match sig_rx.recv_timeout(Duration::from_millis(500)) {
Ok(()) => {
info!("stop requested; unmounting");
if let Err(e) = bg.umount_and_join() {
warn!(error = %e, "umount_and_join failed");
}
break MountOutcome::Shutdown;
}
Err(std::sync::mpsc::RecvTimeoutError::Timeout) => {
if bg.guard.is_finished() {
info!("mount ended externally");
if let Err(e) = bg.join() {
warn!(error = %e, "session join failed");
}
break MountOutcome::Unmounted;
}
}
Err(std::sync::mpsc::RecvTimeoutError::Disconnected) => {
// Signal task gone (failed to install); fall back to join.
let _ = bg.join();
break MountOutcome::Unmounted;
}
}
};
let _ = std::fs::remove_file(control_socket);
result
Ok(outcome)
}
#[cfg(test)]
mod tests {
use super::Intervals;
/// Flatten `segments` into a readable form for assertions.
fn segs(iv: &Intervals, start: u64, end: u64) -> Vec<(u64, u64, bool)> {
iv.segments(start, end)
}
#[test]
fn append_merges_into_one_run() {
// Sequential writes (an append) coalesce into a single authored range.
let mut iv = Intervals::default();
iv.add(0, 10);
iv.add(10, 20);
iv.add(20, 25);
assert_eq!(segs(&iv, 0, 25), vec![(0, 25, true)]);
}
#[test]
fn partial_overwrite_leaves_base_gap() {
// Author [0,4) and [8,12); [4,8) stays base. A read of [0,12) must stitch
// authored / base / authored in order.
let mut iv = Intervals::default();
iv.add(0, 4);
iv.add(8, 12);
assert_eq!(
segs(&iv, 0, 12),
vec![(0, 4, true), (4, 8, false), (8, 12, true)]
);
}
#[test]
fn overlapping_writes_coalesce() {
let mut iv = Intervals::default();
iv.add(0, 10);
iv.add(5, 15);
iv.add(14, 20);
assert_eq!(segs(&iv, 0, 20), vec![(0, 20, true)]);
}
#[test]
fn segments_clamp_to_request_window() {
let mut iv = Intervals::default();
iv.add(0, 100);
// A sub-window of one big authored range is a single authored segment.
assert_eq!(segs(&iv, 20, 50), vec![(20, 50, true)]);
// A window entirely outside any authored range is all base.
let empty = Intervals::default();
assert_eq!(segs(&empty, 0, 8), vec![(0, 8, false)]);
}
#[test]
fn truncate_shrink_drops_tail() {
// Grow-then-shrink: clip removes/truncates authored ranges past the end.
let mut iv = Intervals::default();
iv.add(0, 100);
iv.clip(40);
assert_eq!(segs(&iv, 0, 40), vec![(0, 40, true)]);
// Authored ranges wholly past the new end disappear.
let mut iv2 = Intervals::default();
iv2.add(0, 10);
iv2.add(50, 60);
iv2.clip(40);
assert_eq!(segs(&iv2, 0, 40), vec![(0, 10, true), (10, 40, false)]);
}
#[test]
fn truncate_extend_authors_zero_tail() {
// setattr grow claims the new tail as authored (defined zeros), so commit
// never pulls it from the base.
let mut iv = Intervals::default();
iv.add(0, 10); // base content authored over
let old_len = 10u64;
let new_len = 30u64;
iv.add(old_len, new_len);
assert_eq!(segs(&iv, 0, 30), vec![(0, 30, true)]);
}
}
+429 -174
View File
@@ -5,18 +5,21 @@
//! [`pdfs-tray`](crate) SNI icon. It presents four things in a Proton-purple,
//! Google-Drive-style layout:
//!
//! * a **login form** (email / password / optional 2FA) driving
//! [`pdfs_core::auth::login`];
//! * a **mount toggle** that spawns `pdfs mount` / `fusermount3 -u`;
//! * a **cache usage** read-out from the on-disk [`ContentCache`]; and
//! * a **login form** (email / password, with a lazy 2FA dialog shown only when
//! the account requires it) driving [`pdfs_core::auth::login`]; logging in
//! enables the systemd mount service;
//! * a **read-only mount status** line (the mount is managed automatically);
//! * a **cache usage** read-out; and
//! * the **pin list** with per-file unpin.
//!
//! State the GUI only displays (cache size, pins, login identity) is read
//! straight off disk via `pdfs-core`, so the window is useful even with no mount
//! daemon running; live mount status comes from the daemon over the control
//! socket the CLI and tray already speak.
//! Mount status, cache usage and the pin list all ride along on one
//! [`Request::Status`] round-trip to the daemon (which owns the cache), fetched on
//! a worker thread so the periodic refresh never blocks the GTK main loop. Login
//! identity is cached in [`Ui`] and refreshed only on login/logout, so the 2s tick
//! never touches the keyring.
use std::cell::RefCell;
use std::cell::{Cell, RefCell};
use std::collections::HashSet;
use std::path::{Path, PathBuf};
use std::process::Command;
use std::rc::Rc;
@@ -25,12 +28,12 @@ use std::time::Duration;
use adw::prelude::*;
use gtk4::gio;
use gtk4::glib;
use gtk4::glib::{BoxedAnyObject, SignalHandlerId};
use gtk4::glib::BoxedAnyObject;
use pdfs_core::auth;
use pdfs_core::cache::ContentCache;
use pdfs_core::config::{AppDirs, DEFAULT_CACHE_BUDGET_BYTES};
use pdfs_core::config::AppDirs;
use pdfs_core::control::{DirEntry, PhotoItem, Request, Response, send};
use pdfs_core::service;
/// How many photos to pull per [`Request::PhotosTimeline`] page.
const PHOTOS_PAGE: usize = 60;
@@ -41,6 +44,17 @@ const APP_ID: &str = "io.narl.proton-drive-linux";
const PROTON_PURPLE: &str = "#6d4aff";
/// How often the window re-reads mount status, cache usage and the pin list.
const REFRESH_INTERVAL: Duration = Duration::from_secs(2);
/// Backoff between auto-retries of a Files/Photos load while the mount service
/// is still coming up (see [`load_browser`] / [`load_gallery`]).
const CONNECT_RETRY_INTERVAL: Duration = Duration::from_secs(2);
/// One rendered pin row, retained so [`repaint_pins`] can flip the unpin button's
/// `sensitive` in place (when the pin set is unchanged) instead of rebuilding.
struct PinRow {
row: adw::ActionRow,
/// The unpin button, absent on the placeholder row.
unpin: Option<gtk4::Button>,
}
/// All widgets the periodic refresh and the action handlers mutate, plus the
/// resolved paths they act on. Wrapped in an [`Rc`] so handlers and the timeout
@@ -48,22 +62,39 @@ const REFRESH_INTERVAL: Duration = Duration::from_secs(2);
struct Ui {
dirs: AppDirs,
stack: adw::ViewStack,
/// Header spinner shown while any open/load round-trip is in flight; ref-
/// counted via [`Self::busy`] so concurrent operations don't stop it early.
spinner: gtk4::Spinner,
busy: Cell<u32>,
/// Keys (relative path / photo uid) of open requests currently in flight, so
/// a double-click on the same entry is a no-op instead of a second download.
opening: RefCell<HashSet<String>>,
/// Resolved login identity, cached so the periodic [`refresh`] never hits the
/// keyring (a DBus round-trip). Populated at startup and updated only on
/// login / logout. `None` = signed out.
session: RefCell<Option<auth::StoredSession>>,
/// Whether a [`Request::Status`] round-trip is already in flight, so the 2s
/// refresh tick doesn't pile worker threads up on a slow/wedged daemon.
status_inflight: Cell<bool>,
// Login page.
email: adw::EntryRow,
password: adw::PasswordEntryRow,
totp: adw::EntryRow,
login_button: gtk4::Button,
login_status: gtk4::Label,
// Main page.
account_row: adw::ActionRow,
mount_row: adw::SwitchRow,
mount_handler: RefCell<Option<SignalHandlerId>>,
/// Read-only mount status line. The mount is driven by the systemd user
/// service (enabled on login), not by the user — so this only reports.
mount_row: adw::ActionRow,
cache_bar: gtk4::ProgressBar,
cache_label: gtk4::Label,
pins_group: adw::PreferencesGroup,
/// Rows currently shown under [`Self::pins_group`], so a refresh can clear
/// the previous batch before re-adding.
pin_rows: RefCell<Vec<adw::ActionRow>>,
/// Rows currently shown under [`Self::pins_group`], retained so a refresh can
/// diff against them and only rebuild when the pin set actually changes.
pin_rows: RefCell<Vec<PinRow>>,
/// The pin paths last rendered, the diff baseline for [`repaint_pins`].
/// `None` = nothing built yet; `Some(empty)` = the placeholder is shown.
pins_state: RefCell<Option<Vec<String>>>,
/// Whether the last refresh saw a live mount daemon. Gates the unpin
/// buttons, which need the daemon to evict + re-hydrate.
mounted: RefCell<bool>,
@@ -76,14 +107,37 @@ struct Ui {
browser_back: gtk4::Button,
browser_crumb: gtk4::Label,
browser_status: gtk4::Label,
/// Shown beside [`Self::browser_status`] when a load failed because the mount
/// service is down (not merely starting); restarts the service and reloads.
browser_retry: gtk4::Button,
/// Mountpoint-relative path the browser is showing (empty = root).
browser_path: RefCell<String>,
// Photos (gallery) page.
gallery_model: gio::ListStore,
gallery_status: gtk4::Label,
gallery_retry: gtk4::Button,
gallery_more: gtk4::Button,
}
impl Ui {
/// Begin a unit of background work: show + spin the header spinner.
fn busy_begin(&self) {
self.busy.set(self.busy.get() + 1);
self.spinner.set_visible(true);
self.spinner.start();
}
/// End a unit of background work: stop the spinner once the last one is done.
fn busy_end(&self) {
let remaining = self.busy.get().saturating_sub(1);
self.busy.set(remaining);
if remaining == 0 {
self.spinner.stop();
self.spinner.set_visible(false);
}
}
}
fn main() -> glib::ExitCode {
tracing_subscriber::fmt()
.with_env_filter(
@@ -167,38 +221,48 @@ fn build_window(app: &adw::Application) {
let switcher = adw::ViewSwitcherBar::builder().stack(&stack).build();
// Header spinner, hidden until a background open/load is in flight.
let spinner = gtk4::Spinner::new();
spinner.set_visible(false);
let ui = Rc::new(Ui {
dirs,
stack: stack.clone(),
spinner: spinner.clone(),
busy: Cell::new(0),
opening: RefCell::new(HashSet::new()),
session: RefCell::new(auth::load().ok()),
status_inflight: Cell::new(false),
email: login_widgets.0,
password: login_widgets.1,
totp: login_widgets.2,
login_button: login_widgets.3,
login_status: login_widgets.4,
login_button: login_widgets.2,
login_status: login_widgets.3,
account_row: main_widgets.0,
mount_row: main_widgets.1,
mount_handler: RefCell::new(None),
cache_bar: main_widgets.2,
cache_label: main_widgets.3,
pins_group: main_widgets.4,
pin_rows: RefCell::new(Vec::new()),
pins_state: RefCell::new(None),
mounted: RefCell::new(false),
switcher: switcher.clone(),
browser_model: browser_widgets.0,
browser_back: browser_widgets.1,
browser_crumb: browser_widgets.2,
browser_status: browser_widgets.3,
browser_retry: browser_widgets.6,
browser_path: RefCell::new(String::new()),
gallery_model: gallery_widgets.0,
gallery_status: gallery_widgets.1,
gallery_retry: gallery_widgets.4,
gallery_more: gallery_widgets.2,
});
wire_login(&ui);
wire_logout(&ui, &main_widgets.5);
wire_mount_switch(&ui);
wire_browser(&ui, &browser_widgets.4, &browser_widgets.5);
wire_gallery(&ui, &gallery_widgets.3);
wire_retry(&ui);
// Lazily load the Files / Photos pages the first time they're shown, so the
// network round-trip only happens on demand rather than on every refresh.
@@ -218,6 +282,7 @@ fn build_window(app: &adw::Application) {
brand.append(&icon);
brand.append(&gtk4::Label::new(Some("Proton Drive")));
header.set_title_widget(Some(&brand));
header.pack_end(&spinner);
let toolbar = adw::ToolbarView::new();
toolbar.add_top_bar(&header);
@@ -251,15 +316,15 @@ fn build_window(app: &adw::Application) {
window.present();
}
/// The login page: an email row, password row, optional 2FA row, a primary
/// "Sign in" button and a status label, centred in a clamp.
/// The login page: an email row, password row, a primary "Sign in" button and a
/// status label, centred in a clamp. The 2FA code is prompted lazily in a
/// dialog (see [`prompt_2fa`]) only when the account actually requires it.
#[allow(clippy::type_complexity)]
fn build_login_page() -> (
gtk4::Widget,
(
adw::EntryRow,
adw::PasswordEntryRow,
adw::EntryRow,
gtk4::Button,
gtk4::Label,
),
@@ -271,12 +336,8 @@ fn build_login_page() -> (
let email = adw::EntryRow::builder().title("Email or username").build();
let password = adw::PasswordEntryRow::builder().title("Password").build();
let totp = adw::EntryRow::builder()
.title("Two-factor code (if enabled)")
.build();
group.add(&email);
group.add(&password);
group.add(&totp);
let login_button = gtk4::Button::builder()
.label("Sign in")
@@ -319,7 +380,7 @@ fn build_login_page() -> (
(
scroll.upcast(),
(email, password, totp, login_button, login_status),
(email, password, login_button, login_status),
)
}
@@ -331,7 +392,7 @@ fn build_main_page() -> (
gtk4::Widget,
(
adw::ActionRow,
adw::SwitchRow,
adw::ActionRow,
gtk4::ProgressBar,
gtk4::Label,
adw::PreferencesGroup,
@@ -351,10 +412,11 @@ fn build_main_page() -> (
account_row.add_suffix(&logout_button);
account_group.add(&account_row);
// Mount group: a switch row driving the FUSE mount.
// Mount group: a read-only status line. The mount is managed automatically
// by the systemd user service; there is no toggle to fiddle with.
let mount_group = adw::PreferencesGroup::builder().title("Drive").build();
let mount_row = adw::SwitchRow::builder()
.title("Mount Proton Drive")
let mount_row = adw::ActionRow::builder()
.title("Proton Drive")
.subtitle("Not mounted")
.build();
mount_group.add(&mount_row);
@@ -417,7 +479,6 @@ fn wire_login(ui: &Rc<Ui>) {
button.connect_clicked(move |_| {
let username = ui.email.text().to_string();
let password = ui.password.text().to_string();
let totp = ui.totp.text().to_string();
if username.is_empty() || password.is_empty() {
ui.login_status.set_text("Enter your email and password.");
return;
@@ -425,7 +486,17 @@ fn wire_login(ui: &Rc<Ui>) {
ui.login_button.set_sensitive(false);
ui.login_status.set_text("Signing in…");
let rx = spawn_login(username, password, totp);
let (rx, totp_req_rx) = spawn_login(username, password);
// Surface the 2FA dialog only if the SDK actually asks for a code (i.e.
// the account has 2FA enabled). The worker blocks until the dialog feeds
// back a code (or is cancelled, dropping the sender).
let ui_2fa = ui.clone();
glib::spawn_future_local(async move {
if let Ok(code_tx) = totp_req_rx.recv().await {
prompt_2fa(&ui_2fa, code_tx);
}
});
let ui = ui.clone();
glib::spawn_future_local(async move {
@@ -438,7 +509,10 @@ fn wire_login(ui: &Rc<Ui>) {
Ok(()) => {
ui.login_status.set_text("");
ui.password.set_text("");
ui.totp.set_text("");
// Cache the new identity so `refresh` never hits the keyring.
*ui.session.borrow_mut() = auth::load().ok();
// Enable+start the mount service now that we have a session.
service::enable_start();
refresh(&ui);
}
Err(e) => ui.login_status.set_text(&format!("Sign-in failed: {e}")),
@@ -448,15 +522,21 @@ fn wire_login(ui: &Rc<Ui>) {
}
/// Run the async SRP + optional 2FA login on a dedicated current-thread Tokio
/// runtime, returning a channel that yields the result once. The TOTP closure
/// supplies whatever was typed in the 2FA field, erroring if the account needs a
/// code but none was entered.
/// runtime. Returns two channels: the first yields the final login result once;
/// the second fires *only if* the account needs a 2FA code, carrying a
/// [`std::sync::mpsc::Sender`] the UI uses to feed the code back. The login
/// closure blocks the worker on that sender until the dialog answers, so the
/// code is requested lazily and can't expire before the password proof.
#[allow(clippy::type_complexity)]
fn spawn_login(
username: String,
password: String,
totp: String,
) -> async_channel::Receiver<Result<(), String>> {
) -> (
async_channel::Receiver<Result<(), String>>,
async_channel::Receiver<std::sync::mpsc::Sender<String>>,
) {
let (tx, rx) = async_channel::bounded(1);
let (totp_req_tx, totp_req_rx) = async_channel::bounded(1);
std::thread::spawn(move || {
let rt = match tokio::runtime::Builder::new_current_thread()
.enable_all()
@@ -470,138 +550,210 @@ fn spawn_login(
};
let result = rt.block_on(async move {
auth::login(&username, &password, || {
let code = totp.trim();
if code.is_empty() {
Err(pdfs_core::Error::Other("two-factor code required".into()))
} else {
Ok(code.to_string())
}
// 2FA required: hand a one-shot sender to the UI and block until
// the dialog supplies the code. A dropped sender (cancelled
// dialog) surfaces as a cancelled login.
let (code_tx, code_rx) = std::sync::mpsc::channel::<String>();
totp_req_tx
.send_blocking(code_tx)
.map_err(|_| pdfs_core::Error::Other("two-factor prompt closed".into()))?;
code_rx
.recv()
.map_err(|_| pdfs_core::Error::Other("two-factor entry cancelled".into()))
})
.await
.map_err(|e| e.to_string())
});
let _ = tx.send_blocking(result);
});
rx
(rx, totp_req_rx)
}
/// Connect the sign-out button: forget the stored session and drop back to the
/// login page. Unmounts first so the daemon isn't left without credentials.
/// Show the lazy two-factor dialog and feed the entered code back to the waiting
/// login worker via `code_tx`. Cancelling (or closing) drops the sender, which
/// the worker reads as a cancelled login.
fn prompt_2fa(ui: &Rc<Ui>, code_tx: std::sync::mpsc::Sender<String>) {
let dialog = adw::AlertDialog::builder()
.heading("Two-factor authentication")
.body("Enter the code from your authenticator app.")
.build();
let group = adw::PreferencesGroup::new();
let entry = adw::EntryRow::builder()
.title("Authentication code")
.build();
group.add(&entry);
dialog.set_extra_child(Some(&group));
dialog.add_response("cancel", "Cancel");
dialog.add_response("confirm", "Confirm");
dialog.set_response_appearance("confirm", adw::ResponseAppearance::Suggested);
dialog.set_default_response(Some("confirm"));
dialog.set_close_response("cancel");
let code_tx = RefCell::new(Some(code_tx));
dialog.connect_response(None, move |_, resp| {
// On cancel/close we take + drop `tx` without sending, so the worker's
// recv errors out and the login is reported as cancelled.
if let Some(tx) = code_tx.borrow_mut().take()
&& resp == "confirm"
{
let _ = tx.send(entry.text().trim().to_string());
}
});
let parent = ui.login_button.root().and_downcast::<gtk4::Window>();
dialog.present(parent.as_ref());
}
/// Connect the sign-out button: disable+stop the mount service (so the daemon
/// isn't left running without credentials), forget the stored session, and drop
/// back to the login page.
fn wire_logout(ui: &Rc<Ui>, button: &gtk4::Button) {
let ui = ui.clone();
button.connect_clicked(move |_| {
if *ui.mounted.borrow() {
unmount(&ui.dirs.default_mountpoint());
}
service::disable_stop();
if let Err(e) = auth::logout() {
tracing::error!("logout failed: {e}");
}
*ui.session.borrow_mut() = None;
refresh(&ui);
});
}
/// Connect the mount switch. The handler id is stored so the refresh loop can
/// block it while syncing the switch to the real mount state (otherwise setting
/// `active` would re-trigger a mount/unmount).
fn wire_mount_switch(ui: &Rc<Ui>) {
let ui_for_handler = ui.clone();
let handler = ui.mount_row.connect_active_notify(move |row| {
let mountpoint = ui_for_handler.dirs.default_mountpoint();
if row.is_active() {
spawn_mount();
} else {
unmount(&mountpoint);
}
/// Connect the Files/Photos "Retry" buttons (shown by [`browser_unreachable`] /
/// [`gallery_unreachable`] when the mount is down): restart the systemd unit and
/// reload the page.
fn wire_retry(ui: &Rc<Ui>) {
let ui_browser = ui.clone();
ui.browser_retry.clone().connect_clicked(move |_| {
service::restart();
load_browser(&ui_browser);
});
let ui_gallery = ui.clone();
ui.gallery_retry.clone().connect_clicked(move |_| {
service::restart();
load_gallery(&ui_gallery, false);
});
*ui.mount_handler.borrow_mut() = Some(handler);
}
/// Re-read mount status, login identity, cache usage and the pin list, and
/// repaint the window. Cheap, synchronous local reads — safe on the main loop.
/// Repaint the window from the cached login identity, then kick an async mount-
/// status fetch. Runs on the 2s tick: the identity check is instant (no keyring),
/// and the status round-trip is offloaded to a worker so the main loop never
/// blocks on a slow or wedged daemon.
fn refresh(ui: &Rc<Ui>) {
// Login identity decides which page is shown.
let session = auth::load().ok();
match &session {
Some(s) => {
// Only pull the user onto "main" when they're sitting on the login
// page; otherwise leave whichever signed-in page they navigated to.
if ui.stack.visible_child_name().as_deref() == Some("login") {
ui.stack.set_visible_child_name("main");
// Login identity decides which page is shown. Read the cached session — set
// at startup and on login/logout — never the keyring.
{
let session = ui.session.borrow();
match session.as_ref() {
Some(s) => {
// Only pull the user onto "main" when they're sitting on the login
// page; otherwise leave whichever signed-in page they navigated to.
if ui.stack.visible_child_name().as_deref() == Some("login") {
ui.stack.set_visible_child_name("main");
}
ui.switcher.set_reveal(true);
ui.account_row.set_title(&s.username);
ui.account_row.set_subtitle("Proton account");
}
None => {
ui.stack.set_visible_child_name("login");
ui.switcher.set_reveal(false);
return;
}
ui.switcher.set_reveal(true);
ui.account_row.set_title(&s.username);
ui.account_row.set_subtitle("Proton account");
}
None => {
ui.stack.set_visible_child_name("login");
ui.switcher.set_reveal(false);
return;
}
}
// Live mount status from the daemon, falling back to "not mounted".
let socket = ui.dirs.control_socket();
let mount = match send(&socket, &Request::Status) {
Ok(Response::Status { mountpoint, .. }) => Some(mountpoint),
_ => None,
};
*ui.mounted.borrow_mut() = mount.is_some();
// Sync the switch without re-triggering its handler.
if let Some(id) = ui.mount_handler.borrow().as_ref() {
ui.mount_row.block_signal(id);
ui.mount_row.set_active(mount.is_some());
ui.mount_row.unblock_signal(id);
}
ui.mount_row.set_subtitle(&match &mount {
Some(mp) => format!("Mounted at {mp}"),
None => "Not mounted".into(),
});
// Cache usage + pins, read straight off disk so they work offline too.
let cache = ContentCache::open(
ui.dirs.content_cache_dir(),
ui.dirs.pins_path(),
DEFAULT_CACHE_BUDGET_BYTES,
);
match cache {
Ok(cache) => {
let used = cache.usage();
let budget = cache.budget();
let fraction = if budget == 0 {
0.0
} else {
(used as f64 / budget as f64).min(1.0)
};
ui.cache_bar.set_fraction(fraction);
ui.cache_label.set_text(&format!(
"{} of {} used",
human_bytes(used),
human_bytes(budget)
));
repaint_pins(ui, &cache.list_pins());
}
Err(e) => ui.cache_label.set_text(&format!("Cache unavailable: {e}")),
}
refresh_status(ui);
}
/// Replace the rows under the pins group with one [`adw::ActionRow`] per pin,
/// each with an unpin button enabled only while a mount daemon is running.
fn repaint_pins(ui: &Rc<Ui>, pins: &[pdfs_core::cache::Pin]) {
for row in ui.pin_rows.borrow_mut().drain(..) {
ui.pins_group.remove(&row);
/// Fetch mount status + cache stats from the daemon on a worker thread and repaint
/// the mount line, cache bar and pin list on the reply. The daemon owns the cache
/// stats now (`used`/`budget`/`pins` ride along on [`Response::Status`]), so the
/// GUI never opens the on-disk cache itself. Skipped while a fetch is in flight so
/// the tick can't stack threads on a stalled daemon.
fn refresh_status(ui: &Rc<Ui>) {
if ui.status_inflight.get() {
return;
}
ui.status_inflight.set(true);
let rx = spawn_request(ui.dirs.control_socket(), Request::Status);
let ui = ui.clone();
glib::spawn_future_local(async move {
let result = rx.recv().await;
ui.status_inflight.set(false);
match result {
Ok(Ok(Response::Status {
mountpoint,
used,
budget,
pins,
..
})) => {
*ui.mounted.borrow_mut() = true;
ui.mount_row
.set_subtitle(&format!("Mounted at {mountpoint}"));
let fraction = if budget == 0 {
0.0
} else {
(used as f64 / budget as f64).min(1.0)
};
ui.cache_bar.set_fraction(fraction);
ui.cache_label.set_text(&format!(
"{} of {} used",
human_bytes(used),
human_bytes(budget)
));
repaint_pins(&ui, &pins, true);
}
// Daemon unreachable (still starting, or down): report not-mounted and
// grey out the unpin buttons in place, but leave the last-known pin
// rows and cache read-out so the page doesn't flicker on a blip.
_ => {
*ui.mounted.borrow_mut() = false;
ui.mount_row.set_subtitle("Not mounted");
for r in ui.pin_rows.borrow().iter() {
if let Some(b) = &r.unpin {
b.set_sensitive(false);
}
}
}
}
});
}
/// Render the pins group from `pins`, with the unpin buttons enabled only while a
/// mount daemon is running (`mounted`). Diffs against the last batch by path: when
/// the set is unchanged (the common case on the 2s tick) it only flips the unpin
/// buttons' `sensitive` flag, avoiding the rebuild that used to flicker the list
/// and drop scroll/selection every tick.
fn repaint_pins(ui: &Rc<Ui>, pins: &[pdfs_core::cache::Pin], mounted: bool) {
let desired: Vec<String> = pins.iter().map(|p| p.path.clone()).collect();
if ui.pins_state.borrow().as_ref() == Some(&desired) {
for r in ui.pin_rows.borrow().iter() {
if let Some(b) = &r.unpin {
b.set_sensitive(mounted);
}
}
return;
}
for pr in ui.pin_rows.borrow_mut().drain(..) {
ui.pins_group.remove(&pr.row);
}
*ui.pins_state.borrow_mut() = Some(desired);
if pins.is_empty() {
let row = adw::ActionRow::builder()
.title("No pinned files")
.subtitle("Right-click a file in the mount to keep it offline.")
.build();
ui.pins_group.add(&row);
ui.pin_rows.borrow_mut().push(row);
ui.pin_rows.borrow_mut().push(PinRow { row, unpin: None });
return;
}
let mounted = *ui.mounted.borrow();
for pin in pins {
let name = Path::new(&pin.path)
.file_name()
@@ -635,7 +787,10 @@ fn repaint_pins(ui: &Rc<Ui>, pins: &[pdfs_core::cache::Pin]) {
row.add_suffix(&unpin);
ui.pins_group.add(&row);
ui.pin_rows.borrow_mut().push(row);
ui.pin_rows.borrow_mut().push(PinRow {
row,
unpin: Some(unpin),
});
}
}
@@ -665,6 +820,7 @@ fn spawn_request(
/// The factories that render entries — and the columns — need the [`Ui`] handle
/// for activation and the right-click menu, so they're installed later in
/// [`wire_browser`]; this builder only assembles the empty widgets.
#[allow(clippy::type_complexity)]
fn build_browser_page() -> (
gtk4::Widget,
(
@@ -674,6 +830,7 @@ fn build_browser_page() -> (
gtk4::Label,
gtk4::GridView,
gtk4::ColumnView,
gtk4::Button,
),
) {
let model = gio::ListStore::new::<BoxedAnyObject>();
@@ -718,6 +875,18 @@ fn build_browser_page() -> (
status.add_css_class("dim-label");
status.set_visible(false);
// Shown only when a load failed because the mount is down; restarts it.
let retry = gtk4::Button::builder()
.label("Retry")
.halign(gtk4::Align::Start)
.build();
retry.add_css_class("pill");
retry.set_visible(false);
let status_box = gtk4::Box::new(gtk4::Orientation::Vertical, 8);
status_box.append(&status);
status_box.append(&retry);
// Icon grid.
let grid = gtk4::GridView::builder()
.model(&gtk4::SingleSelection::new(Some(model.clone())))
@@ -764,12 +933,12 @@ fn build_browser_page() -> (
inner.set_margin_start(12);
inner.set_margin_end(12);
inner.append(&header);
inner.append(&status);
inner.append(&status_box);
inner.append(&view_stack);
(
inner.upcast(),
(model, back, crumb, status, grid, column_view),
(model, back, crumb, status, grid, column_view, retry),
)
}
@@ -999,9 +1168,22 @@ fn activate_entry(ui: &Rc<Ui>, entry: &DirEntry) {
*ui.browser_path.borrow_mut() = rel;
load_browser(ui);
} else {
let rx = spawn_request(ui.dirs.control_socket(), Request::OpenFile { path: rel });
// Ignore a repeat activation of a file already downloading, so an
// impatient double-click doesn't kick off a second round-trip.
if !ui.opening.borrow_mut().insert(rel.clone()) {
return;
}
ui.busy_begin();
let rx = spawn_request(
ui.dirs.control_socket(),
Request::OpenFile { path: rel.clone() },
);
let ui = ui.clone();
glib::spawn_future_local(async move {
match rx.recv().await {
let result = rx.recv().await;
ui.busy_end();
ui.opening.borrow_mut().remove(&rel);
match result {
Ok(Ok(Response::FilePath { path })) => open_path(&path),
Ok(Ok(Response::Error { message })) => {
tracing::error!("open file failed: {message}")
@@ -1099,27 +1281,61 @@ fn load_browser(ui: &Rc<Ui>) {
&path
});
ui.browser_back.set_sensitive(!path.is_empty());
ui.browser_status.set_visible(false);
// Drop the previous folder's rows up front: a slow reply must not leave stale
// entries visible, where clicking one would open with a wrong relative path.
ui.browser_model.remove_all();
ui.browser_retry.set_visible(false);
ui.browser_status.set_label("Loading…");
ui.browser_status.set_visible(true);
ui.busy_begin();
let rx = spawn_request(ui.dirs.control_socket(), Request::ListDir { path });
let ui = ui.clone();
glib::spawn_future_local(async move {
match rx.recv().await {
let result = rx.recv().await;
ui.busy_end();
match result {
Ok(Ok(Response::Entries { entries })) => repaint_browser(&ui, &entries),
Ok(Ok(Response::Error { message })) => browser_message(&ui, &message),
Ok(Ok(_)) => browser_message(&ui, "Unexpected reply from daemon."),
Ok(Err(_)) | Err(_) => browser_message(&ui, "Mount Proton Drive to browse your files."),
Ok(Err(_)) | Err(_) => browser_unreachable(&ui),
}
});
}
/// Clear the model and show a single status line instead.
/// Clear the model and show a single status line instead. Hides the Retry button
/// (used for "in-band" outcomes: empty folder, or an error the daemon returned).
fn browser_message(ui: &Rc<Ui>, message: &str) {
ui.browser_model.remove_all();
ui.browser_retry.set_visible(false);
ui.browser_status.set_label(message);
ui.browser_status.set_visible(true);
}
/// The daemon didn't answer. Distinguish *still starting* (auto-retry, no
/// button) from *down* (actionable error + Retry), so a cold start self-heals
/// once the systemd mount comes up but a real failure stays visible.
fn browser_unreachable(ui: &Rc<Ui>) {
if service::is_failed() || !service::is_active() {
ui.browser_status
.set_label("Couldn't reach Proton Drive. The mount service isn't running.");
ui.browser_status.set_visible(true);
ui.browser_retry.set_visible(true);
return;
}
ui.browser_status.set_label("Connecting to Proton Drive…");
ui.browser_status.set_visible(true);
ui.browser_retry.set_visible(false);
let ui = ui.clone();
glib::timeout_add_local_once(CONNECT_RETRY_INTERVAL, move || {
// Only keep polling while the Files page is the one on screen.
if ui.stack.visible_child_name().as_deref() == Some("browser") {
load_browser(&ui);
}
});
}
/// Repopulate the shared model — folders first, then case-insensitive by name —
/// which refreshes both the grid and the column list.
fn repaint_browser(ui: &Rc<Ui>, entries: &[DirEntry]) {
@@ -1146,7 +1362,13 @@ fn repaint_browser(ui: &Rc<Ui>, entries: &[DirEntry]) {
/// "Load more" button and a status label.
fn build_gallery_page() -> (
gtk4::Widget,
(gio::ListStore, gtk4::Label, gtk4::Button, gtk4::GridView),
(
gio::ListStore,
gtk4::Label,
gtk4::Button,
gtk4::GridView,
gtk4::Button,
),
) {
let model = gio::ListStore::new::<BoxedAnyObject>();
@@ -1187,6 +1409,14 @@ fn build_gallery_page() -> (
status.add_css_class("dim-label");
status.set_visible(false);
// Shown only when a load failed because the mount is down; restarts it.
let retry = gtk4::Button::builder()
.label("Retry")
.halign(gtk4::Align::Center)
.build();
retry.add_css_class("pill");
retry.set_visible(false);
let more = gtk4::Button::builder()
.label("Load more")
.halign(gtk4::Align::Center)
@@ -1205,10 +1435,11 @@ fn build_gallery_page() -> (
inner.set_margin_start(12);
inner.set_margin_end(12);
inner.append(&status);
inner.append(&retry);
inner.append(&scroll);
inner.append(&more);
(inner.upcast(), (model, status, more, grid))
(inner.upcast(), (model, status, more, grid, retry))
}
/// Wire the gallery: activating a tile downloads the photo and opens it; the
@@ -1223,9 +1454,21 @@ fn wire_gallery(ui: &Rc<Ui>, grid: &gtk4::GridView) {
.downcast_ref::<BoxedAnyObject>()
.map(|o| o.borrow::<PhotoItem>().uid.clone());
let Some(uid) = uid else { return };
let rx = spawn_request(ui_open.dirs.control_socket(), Request::OpenPhoto { uid });
// Ignore a repeat activation of a photo already downloading.
if !ui_open.opening.borrow_mut().insert(uid.clone()) {
return;
}
ui_open.busy_begin();
let rx = spawn_request(
ui_open.dirs.control_socket(),
Request::OpenPhoto { uid: uid.clone() },
);
let ui = ui_open.clone();
glib::spawn_future_local(async move {
match rx.recv().await {
let result = rx.recv().await;
ui.busy_end();
ui.opening.borrow_mut().remove(&uid);
match result {
Ok(Ok(Response::FilePath { path })) => open_path(&path),
Ok(Ok(Response::Error { message })) => {
tracing::error!("open photo failed: {message}")
@@ -1244,13 +1487,19 @@ fn wire_gallery(ui: &Rc<Ui>, grid: &gtk4::GridView) {
/// Fetch a timeline page from the daemon. When `append` is false the model is
/// cleared first (fresh load); otherwise the next page is tacked on.
fn load_gallery(ui: &Rc<Ui>, append: bool) {
if !append {
if append {
ui.gallery_status.set_visible(false);
} else {
// Fresh load: clear the grid and show Loading until the first page lands.
ui.gallery_model.remove_all();
ui.gallery_retry.set_visible(false);
ui.gallery_status.set_label("Loading…");
ui.gallery_status.set_visible(true);
}
let offset = ui.gallery_model.n_items() as usize;
ui.gallery_status.set_visible(false);
ui.gallery_more.set_sensitive(false);
ui.busy_begin();
let rx = spawn_request(
ui.dirs.control_socket(),
Request::PhotosTimeline {
@@ -1260,8 +1509,10 @@ fn load_gallery(ui: &Rc<Ui>, append: bool) {
);
let ui = ui.clone();
glib::spawn_future_local(async move {
let result = rx.recv().await;
ui.busy_end();
ui.gallery_more.set_sensitive(true);
match rx.recv().await {
match result {
Ok(Ok(Response::Photos { available, items })) => {
if !available {
gallery_message(&ui, "This account has no photos.");
@@ -1280,18 +1531,41 @@ fn load_gallery(ui: &Rc<Ui>, append: bool) {
}
Ok(Ok(Response::Error { message })) => gallery_message(&ui, &message),
Ok(Ok(_)) => gallery_message(&ui, "Unexpected reply from daemon."),
Ok(Err(_)) | Err(_) => gallery_message(&ui, "Mount Proton Drive to see your photos."),
Ok(Err(_)) | Err(_) => gallery_unreachable(&ui),
}
});
}
/// Show a gallery status line and hide the pager.
/// Show a gallery status line and hide the pager + Retry button.
fn gallery_message(ui: &Rc<Ui>, message: &str) {
ui.gallery_retry.set_visible(false);
ui.gallery_status.set_label(message);
ui.gallery_status.set_visible(true);
ui.gallery_more.set_visible(false);
}
/// Photos counterpart of [`browser_unreachable`]: auto-retry while the mount is
/// still starting, surface an actionable error + Retry once it's actually down.
fn gallery_unreachable(ui: &Rc<Ui>) {
ui.gallery_more.set_visible(false);
if service::is_failed() || !service::is_active() {
ui.gallery_status
.set_label("Couldn't reach Proton Drive. The mount service isn't running.");
ui.gallery_status.set_visible(true);
ui.gallery_retry.set_visible(true);
return;
}
ui.gallery_status.set_label("Connecting to Proton Drive…");
ui.gallery_status.set_visible(true);
ui.gallery_retry.set_visible(false);
let ui = ui.clone();
glib::timeout_add_local_once(CONNECT_RETRY_INTERVAL, move || {
if ui.stack.visible_child_name().as_deref() == Some("gallery") {
load_gallery(&ui, false);
}
});
}
/// Open a local path with the user's default handler.
fn open_path(path: &str) {
if let Err(e) = Command::new("xdg-open").arg(path).spawn() {
@@ -1299,25 +1573,6 @@ fn open_path(path: &str) {
}
}
/// Launch `pdfs mount` detached so the window keeps running independently.
fn spawn_mount() {
match Command::new("pdfs").arg("mount").spawn() {
Ok(_) => tracing::info!("spawned `pdfs mount`"),
Err(e) => tracing::error!("failed to spawn `pdfs mount`: {e}"),
}
}
/// Tear down the FUSE mount. `fusermount3 -u` is the native path; fall back to
/// `fusermount` on older systems.
fn unmount(mountpoint: &Path) {
let try_unmount = |bin: &str| Command::new(bin).arg("-u").arg(mountpoint).status();
match try_unmount("fusermount3").or_else(|_| try_unmount("fusermount")) {
Ok(s) if s.success() => tracing::info!("unmounted {}", mountpoint.display()),
Ok(s) => tracing::error!("unmount exited with {s}"),
Err(e) => tracing::error!("failed to run fusermount: {e}"),
}
}
/// Format a byte count as a short binary-unit string (e.g. `1.2 GiB`).
fn human_bytes(bytes: u64) -> String {
const UNITS: [&str; 5] = ["B", "KiB", "MiB", "GiB", "TiB"];
+12 -29
View File
@@ -19,6 +19,7 @@ use ksni::{MenuItem, Tray, TrayService};
use pdfs_core::auth;
use pdfs_core::config::AppDirs;
use pdfs_core::control::{Request, Response, send};
use pdfs_core::service;
/// How often the tray re-polls the daemon to refresh its menu.
const POLL_INTERVAL: Duration = Duration::from_secs(3);
@@ -71,26 +72,6 @@ fn poll_state(socket: &Path, default_mountpoint: &Path) -> DriveState {
}
}
/// Launch `pdfs mount` detached so the tray keeps running independently.
fn spawn_mount() {
match Command::new("pdfs").arg("mount").spawn() {
Ok(_) => tracing::info!("spawned `pdfs mount`"),
Err(e) => tracing::error!("failed to spawn `pdfs mount`: {e}"),
}
}
/// Unmount the FUSE filesystem. `fusermount3 -u` is the FUSE-native teardown;
/// fall back to `fusermount` on older systems.
fn unmount(mountpoint: &Path) {
let try_unmount = |bin: &str| Command::new(bin).arg("-u").arg(mountpoint).status();
let result = try_unmount("fusermount3").or_else(|_| try_unmount("fusermount"));
match result {
Ok(s) if s.success() => tracing::info!("unmounted {}", mountpoint.display()),
Ok(s) => tracing::error!("unmount exited with {s}"),
Err(e) => tracing::error!("failed to run fusermount: {e}"),
}
}
fn open_folder(mountpoint: &Path) {
if let Err(e) = Command::new("xdg-open").arg(mountpoint).spawn() {
tracing::error!("failed to xdg-open {}: {e}", mountpoint.display());
@@ -111,9 +92,8 @@ impl Tray for DriveTray {
}
fn menu(&self) -> Vec<MenuItem<Self>> {
// Values captured into the action closures (which only get `&mut Self`).
let mountpoint = self.state.mountpoint.clone();
let open_mp = mountpoint.clone();
// Value captured into the action closure (which only gets `&mut Self`).
let open_mp = self.state.mountpoint.clone();
let mut items: Vec<MenuItem<Self>> = vec![
StandardItem {
@@ -134,25 +114,28 @@ impl Tray for DriveTray {
}
.into(),
);
// Stop the systemd service (clean SIGTERM → lazy unmount). The next
// login or reboot brings it back; this is a deliberate disconnect.
items.push(
StandardItem {
label: "Unmount".into(),
label: "Disconnect".into(),
activate: Box::new(move |this: &mut Self| {
unmount(&mountpoint);
service::stop();
this.state.mounted = false;
this.state.line = "Unmounting…".into();
this.state.line = "Disconnecting…".into();
}),
..Default::default()
}
.into(),
);
} else {
// Enable+start the service so it mounts now and on future logins.
items.push(
StandardItem {
label: "Mount".into(),
label: "Connect".into(),
activate: Box::new(|this: &mut Self| {
spawn_mount();
this.state.line = "Mounting…".into();
service::enable_start();
this.state.line = "Connecting…".into();
}),
..Default::default()
}
+6 -1
View File
@@ -8,7 +8,7 @@
# Build + install: cd packaging && makepkg -fi
pkgname=proton-drive-linux
pkgver=0.1.0
pkgver=0.1.2
pkgrel=1
pkgdesc="Proton Drive client for Linux: FUSE files-on-demand mount, CLI, GTK4 app + tray"
arch=('x86_64')
@@ -61,5 +61,10 @@ package() {
install -Dm644 "packaging/io.narl.proton-drive-linux.svg" \
"$pkgdir/usr/share/icons/hicolor/scalable/apps/io.narl.proton-drive-linux.svg"
# systemd user service that runs the auto-mount daemon. Enabled per-user by
# the GUI on login (`systemctl --user enable --now proton-drive.service`).
install -Dm644 "packaging/proton-drive.service" \
"$pkgdir/usr/lib/systemd/user/proton-drive.service"
install -Dm644 "LICENSE" "$pkgdir/usr/share/licenses/$pkgname/LICENSE"
}
+19
View File
@@ -0,0 +1,19 @@
[Unit]
Description=Proton Drive Files-On-Demand mount
Documentation=https://github.com/narl/proton-drive-linux
# Tie the mount to the desktop session: start after it is up and stop with it.
After=graphical-session.target
PartOf=graphical-session.target
[Service]
Type=simple
# The daemon waits for a stored login, mounts ~/ProtonDrive, and keeps it
# mounted. SIGTERM (from `systemctl --user stop`) triggers a clean lazy unmount.
ExecStart=/usr/bin/pdfs daemon
Restart=on-failure
RestartSec=3
# Give the lazy unmount time to finish before SIGKILL.
TimeoutStopSec=15
[Install]
WantedBy=default.target