Compare commits

...
11 Commits
Author SHA1 Message Date
nvrl d131d3dca8 implemented proton drive prompt 2026-07-01 00:21:42 +02:00
nvrl 5766bc4ed1 transfer ui 2026-07-01 00:02:35 +02:00
nvrl 35bc957cd5 transfer progress api 2026-06-30 23:52:41 +02:00
nvrl f9000f8540 version bump 2026-06-30 21:42:09 +02:00
nvrl d77d7ed817 version bump for sdk 2026-06-30 21:11:21 +02:00
nvrl 56f2b35aca version bump to 0.2.2 2026-06-30 18:39:11 +02:00
nvrl 56abe3fe14 added settings + non blocking ui actions 2026-06-30 18:38:38 +02:00
nvrl 5e71148fec version bump 0.2.1 2026-06-30 17:42:38 +02:00
nvrl 2812e0c893 clippy + pkgbuild 2026-06-30 17:41:52 +02:00
nvrl 03dd8d8496 implemented database for search 2026-06-30 17:35:18 +02:00
nvrl 5e151bbfc5 updated readme 2026-06-30 06:31:51 +02:00
19 changed files with 5052 additions and 271 deletions

No files matched your search

Generated
+83 -9
View File
@@ -53,6 +53,18 @@ dependencies = [
"aes", "aes",
] ]
[[package]]
name = "ahash"
version = "0.8.12"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "5a15f179cd60c4584b8a8c596927aadc462e27f2ca70c04e0071964a73ba7a75"
dependencies = [
"cfg-if",
"once_cell",
"version_check",
"zerocopy",
]
[[package]] [[package]]
name = "aho-corasick" name = "aho-corasick"
version = "1.1.4" version = "1.1.4"
@@ -1278,6 +1290,18 @@ dependencies = [
"pin-project-lite", "pin-project-lite",
] ]
[[package]]
name = "fallible-iterator"
version = "0.3.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "2acce4a10f12dc2fb14a218589d4f1f62ef011b2d0cc4b3cb1bba8e94da14649"
[[package]]
name = "fallible-streaming-iterator"
version = "0.1.9"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "7360491ce676a36bf9bb3c56c1aa791658183a54d2744120f27285738d90465a"
[[package]] [[package]]
name = "fastrand" name = "fastrand"
version = "2.4.1" version = "2.4.1"
@@ -1793,12 +1817,30 @@ dependencies = [
"system-deps", "system-deps",
] ]
[[package]]
name = "hashbrown"
version = "0.14.5"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "e5274423e17b7c9fc20b6e7e208532f9b19825d82dfd615708b70edd83df41f1"
dependencies = [
"ahash",
]
[[package]] [[package]]
name = "hashbrown" name = "hashbrown"
version = "0.17.1" version = "0.17.1"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "ed5909b6e89a2db4456e54cd5f673791d7eca6732202bbf2a9cc504fe2f9b84a" checksum = "ed5909b6e89a2db4456e54cd5f673791d7eca6732202bbf2a9cc504fe2f9b84a"
[[package]]
name = "hashlink"
version = "0.9.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "6ba4ff7128dee98c7dc9794b6a411377e1404dba1c97deb8d1a55297bd25d8af"
dependencies = [
"hashbrown 0.14.5",
]
[[package]] [[package]]
name = "heck" name = "heck"
version = "0.5.0" version = "0.5.0"
@@ -2117,7 +2159,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "d466e9454f08e4a911e14806c24e16fba1b4c121d1ea474396f396069cf949d9" checksum = "d466e9454f08e4a911e14806c24e16fba1b4c121d1ea474396f396069cf949d9"
dependencies = [ dependencies = [
"equivalent", "equivalent",
"hashbrown", "hashbrown 0.17.1",
] ]
[[package]] [[package]]
@@ -2345,6 +2387,17 @@ dependencies = [
"libc", "libc",
] ]
[[package]]
name = "libsqlite3-sys"
version = "0.30.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "2e99fb7a497b1e3339bc746195567ed8d3e24945ecd636e3619d20b9de9e9149"
dependencies = [
"cc",
"pkg-config",
"vcpkg",
]
[[package]] [[package]]
name = "linux-raw-sys" name = "linux-raw-sys"
version = "0.12.1" version = "0.12.1"
@@ -2772,7 +2825,7 @@ dependencies = [
[[package]] [[package]]
name = "pdfs-cli" name = "pdfs-cli"
version = "0.2.0" version = "0.2.3"
dependencies = [ dependencies = [
"anyhow", "anyhow",
"clap 4.6.1", "clap 4.6.1",
@@ -2787,12 +2840,13 @@ dependencies = [
[[package]] [[package]]
name = "pdfs-core" name = "pdfs-core"
version = "0.2.0" version = "0.2.3"
dependencies = [ dependencies = [
"directories", "directories",
"keyring", "keyring",
"proton-drive-rs", "proton-drive-rs",
"proton-sdk", "proton-sdk",
"rusqlite",
"serde", "serde",
"serde_json", "serde_json",
"sha2 0.10.9", "sha2 0.10.9",
@@ -2803,7 +2857,7 @@ dependencies = [
[[package]] [[package]]
name = "pdfs-fuse" name = "pdfs-fuse"
version = "0.2.0" version = "0.2.3"
dependencies = [ dependencies = [
"fuser", "fuser",
"libc", "libc",
@@ -2816,7 +2870,7 @@ dependencies = [
[[package]] [[package]]
name = "pdfs-gui" name = "pdfs-gui"
version = "0.2.0" version = "0.2.3"
dependencies = [ dependencies = [
"anyhow", "anyhow",
"async-channel", "async-channel",
@@ -3031,9 +3085,9 @@ dependencies = [
[[package]] [[package]]
name = "proton-drive-rs" name = "proton-drive-rs"
version = "0.1.2" version = "0.1.3"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "c9380c88a032e2ff079b2d5a6221ccc40ebeee3ebbb4ea60d4d65ef770e4ca8d" checksum = "76e4399b4d9b076c5886c7db87ae7f70322dd7127e6c5a4e7eff8719434d2434"
dependencies = [ dependencies = [
"async-trait", "async-trait",
"base64", "base64",
@@ -3056,9 +3110,9 @@ dependencies = [
[[package]] [[package]]
name = "proton-sdk" name = "proton-sdk"
version = "0.1.2" version = "0.1.3"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "1da22604d2568af2d15be6a45f23f8c5b95fa775bdac1540e833c2eafb587888" checksum = "099f6f20709452f2cdb0cee4db0a82e61a29bec261295812908bf956ffcdaa00"
dependencies = [ dependencies = [
"aes-gcm", "aes-gcm",
"async-trait", "async-trait",
@@ -3406,6 +3460,20 @@ dependencies = [
"windows-sys 0.59.0", "windows-sys 0.59.0",
] ]
[[package]]
name = "rusqlite"
version = "0.32.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "7753b721174eb8ff87a9a0e799e2d7bc3749323e773db92e0984debb00019d6e"
dependencies = [
"bitflags 2.13.0",
"fallible-iterator",
"fallible-streaming-iterator",
"hashlink",
"libsqlite3-sys",
"smallvec",
]
[[package]] [[package]]
name = "rustc-hash" name = "rustc-hash"
version = "2.1.2" version = "2.1.2"
@@ -4347,6 +4415,12 @@ version = "0.1.1"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "ba73ea9cf16a25df0c8caa16c51acb937d5712a8429db78a3ee29d5dcacd3a65" checksum = "ba73ea9cf16a25df0c8caa16c51acb937d5712a8429db78a3ee29d5dcacd3a65"
[[package]]
name = "vcpkg"
version = "0.2.15"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "accd4ea62f7bb7a82fe23066fb0957d48ef677f6eeb8215f372f52e48bb32426"
[[package]] [[package]]
name = "vec_map" name = "vec_map"
version = "0.8.2" version = "0.8.2"
+5 -3
View File
@@ -3,7 +3,7 @@ resolver = "2"
members = ["crates/pdfs-core", "crates/pdfs-fuse", "crates/pdfs-cli", "crates/pdfs-gui"] members = ["crates/pdfs-core", "crates/pdfs-fuse", "crates/pdfs-cli", "crates/pdfs-gui"]
[workspace.package] [workspace.package]
version = "0.2.0" version = "0.2.3"
edition = "2024" edition = "2024"
license = "MIT" license = "MIT"
authors = ["Nils Pukropp <contact@narl.io>"] authors = ["Nils Pukropp <contact@narl.io>"]
@@ -11,8 +11,8 @@ rust-version = "1.96"
[workspace.dependencies] [workspace.dependencies]
# Local Proton SDK crates (path deps until published; switch to versioned later). # Local Proton SDK crates (path deps until published; switch to versioned later).
proton-sdk = "0.1" proton-sdk = "0.1.3"
proton-drive-rs = "0.1" proton-drive-rs = "0.1.3"
# Internal crates. # Internal crates.
pdfs-core = { path = "crates/pdfs-core" } pdfs-core = { path = "crates/pdfs-core" }
@@ -27,8 +27,10 @@ tracing = "0.1"
tracing-subscriber = { version = "0.3", features = ["env-filter"] } tracing-subscriber = { version = "0.3", features = ["env-filter"] }
directories = "5" directories = "5"
sha2 = "0.10" sha2 = "0.10"
rusqlite = { version = "0.32", features = ["bundled"] }
keyring = { version = "3", features = ["sync-secret-service", "crypto-rust"] } keyring = { version = "3", features = ["sync-secret-service", "crypto-rust"] }
clap = { version = "4", features = ["derive"] } clap = { version = "4", features = ["derive"] }
rpassword = "7" rpassword = "7"
fuser = "0.17" fuser = "0.17"
libc = "0.2" libc = "0.2"
+14 -4
View File
@@ -1,16 +1,26 @@
# Proton Drive client for Linux (unofficial) # Proton Drive client for Linux (unofficial)
A fast, unofficial Proton Drive client for Linux. This client provides a FUSE files-on-demand virtual mount, a command-line interface (CLI), and a GTK4 desktop application with system tray integration. A fast, unofficial Proton Drive client for Linux. This client features an advanced files-on-demand FUSE virtual mount with block-level caching, a command-line interface (CLI), and a fully non-blocking GTK4 desktop application with system tray integration.
## Features ## Features
- **Files-on-demand (FUSE)**: Mount your Proton Drive as a virtual filesystem where files are downloaded only when opened. - **Files-on-Demand (FUSE)**: Mount your Proton Drive as a virtual filesystem where files are downloaded only when opened, utilizing smart block-level caching and disk-backed writes.
- **Command-Line Interface (CLI)**: Manage your drive, authenticate, and monitor sync status directly from the terminal. - **Command-Line Interface (CLI)**: Manage your drive, authenticate, and monitor sync status directly from the terminal.
- **GTK4 Desktop App**: Log in, browse files, and configure sync options through a modern, native Linux GUI. - **Non-Blocking GTK4 Desktop App**: Browse files, manage pins, and configure options through a modern, native GUI with a fully non-blocking asynchronous main loop.
- **System Tray Integration**: Background indicator for status monitoring, quick actions, and fast sync controls. - **System Tray Integration**: Background indicator for status monitoring, quick actions, and fast sync controls.
- **Secure Credential Storage**: Integrates with the system Secret Service (GNOME Keyring, KWallet, etc.) for safe token storage. - **Secure Credential Storage**: Integrates with the system Secret Service (GNOME Keyring, KWallet, etc.) with smart in-memory credential caching to avoid UI thread blockages.
- **Proton Photos Support**: Access your Proton Photos timeline, view thumbnails, and download backed-up media natively (available in the GUI as a navigation tab and via the CLI). - **Proton Photos Support**: Access your Proton Photos timeline, view thumbnails, and download backed-up media natively (available in the GUI as a navigation tab and via the CLI).
## Performance & Caching
The client includes several optimizations designed for high efficiency, a low memory footprint, and a responsive user experience:
- **On-Demand Block Cache**: Files are read in fixed 4 MiB blocks. For unpinned files, the client fetches and caches only the blocks containing the requested byte ranges. This enables fast sequential and sparse reads (e.g. media streaming or metadata scanning) without downloading entire files.
- **Disk-Backed Writes**: Large file writes are staged on disk in temporary scratch files (rather than fully buffered in RAM) and track modified byte intervals. Only the unedited remote segments are fetched at commit time, keeping memory usage minimal.
- **Non-Blocking GTK4 Loop**: To prevent UI freezes, all synchronous D-Bus credential checks, control socket requests, and cache usage calculations are offloaded to background worker threads or fetched asynchronously.
- **Flicker-Free UI Rendering**: The GTK4 application performs differential rendering of the pins list, only modifying list rows when the list content changes, preserving the user's scroll position and widget focus.
## Screenshots ## Screenshots
### GUI Application ### GUI Application
+176 -2
View File
@@ -2,6 +2,7 @@
use std::io::Write; use std::io::Write;
use std::path::{Path, PathBuf}; use std::path::{Path, PathBuf};
use std::sync::Arc;
use anyhow::{Context, Result, anyhow, bail}; use anyhow::{Context, Result, anyhow, bail};
use clap::{Parser, Subcommand}; use clap::{Parser, Subcommand};
@@ -9,6 +10,7 @@ use pdfs_core::auth;
use pdfs_core::cache::ContentCache; use pdfs_core::cache::ContentCache;
use pdfs_core::config::AppDirs; use pdfs_core::config::AppDirs;
use pdfs_core::control::{Request as CtlRequest, Response as CtlResponse}; use pdfs_core::control::{Request as CtlRequest, Response as CtlResponse};
use pdfs_core::db::Db;
#[derive(Parser)] #[derive(Parser)]
#[command( #[command(
@@ -76,6 +78,49 @@ enum Command {
/// Photo node uid in `volume~link` form (from `pdfs photos`). /// Photo node uid in `volume~link` form (from `pdfs photos`).
uid: String, uid: String,
}, },
/// Search file/folder names via the running daemon's local index.
Search {
/// Query string (substring match).
query: String,
/// Max hits to return.
#[arg(long, default_value_t = 50)]
limit: usize,
},
/// Rename a file or folder via the running daemon.
Rename {
/// File/folder path, inside the mountpoint or relative to it.
path: PathBuf,
/// New name (a single path component).
new_name: String,
},
/// Move a file or folder into another folder via the running daemon.
Move {
/// File/folder path, inside the mountpoint or relative to it.
path: PathBuf,
/// Destination folder path, inside the mountpoint or relative to it.
new_parent: PathBuf,
},
/// Trash a file or folder via the running daemon.
Rm {
/// File/folder path, inside the mountpoint or relative to it.
path: PathBuf,
},
/// Create a folder via the running daemon.
Mkdir {
/// Parent folder path, inside the mountpoint or relative to it.
parent: PathBuf,
/// New folder name (a single path component).
name: String,
},
/// Upload a local file into a Drive folder via the running daemon.
Upload {
/// Local file to upload.
file: PathBuf,
/// Destination folder path, inside the mountpoint or relative to it.
parent: PathBuf,
},
/// Show the daemon's in-flight transfers (active uploads/downloads).
Transfers,
} }
fn main() -> Result<()> { fn main() -> Result<()> {
@@ -102,6 +147,13 @@ fn main() -> Result<()> {
Command::Ls { path } => cmd_ls(path), Command::Ls { path } => cmd_ls(path),
Command::Photos { limit, offset } => cmd_photos(limit, offset), Command::Photos { limit, offset } => cmd_photos(limit, offset),
Command::OpenPhoto { uid } => cmd_open_photo(uid), Command::OpenPhoto { uid } => cmd_open_photo(uid),
Command::Search { query, limit } => cmd_search(query, limit),
Command::Rename { path, new_name } => cmd_rename(path, new_name),
Command::Move { path, new_parent } => cmd_move(path, new_parent),
Command::Rm { path } => cmd_rm(path),
Command::Mkdir { parent, name } => cmd_mkdir(parent, name),
Command::Upload { file, parent } => cmd_upload(file, parent),
Command::Transfers => cmd_transfers(),
} }
} }
@@ -212,15 +264,23 @@ fn cmd_daemon(mountpoint: Option<PathBuf>) -> Result<()> {
fn mount_once(mountpoint: Option<PathBuf>) -> Result<pdfs_fuse::MountOutcome> { fn mount_once(mountpoint: Option<PathBuf>) -> Result<pdfs_fuse::MountOutcome> {
let dirs = AppDirs::new()?; let dirs = AppDirs::new()?;
dirs.ensure()?; dirs.ensure()?;
let mountpoint = mountpoint.unwrap_or_else(|| dirs.default_mountpoint()); let config = dirs.load_config();
// An explicit `--mountpoint` wins; otherwise honor the Settings-page choice
// (config), falling back to the default location.
let mountpoint = mountpoint.unwrap_or_else(|| dirs.resolved_mountpoint(&config));
std::fs::create_dir_all(&mountpoint) std::fs::create_dir_all(&mountpoint)
.with_context(|| format!("create mountpoint {}", mountpoint.display()))?; .with_context(|| format!("create mountpoint {}", mountpoint.display()))?;
let username = auth::load().map(|s| s.username).unwrap_or_default(); let username = auth::load().map(|s| s.username).unwrap_or_default();
// The daemon owns the unified DB (single writer). Open it here so the content
// cache and the mount share one handle: the cache uses its `cache_entries`
// table as the LRU index, the mount uses it for nodes/search/event cursor.
let db = Arc::new(Db::open(&dirs.db_path()).context("open cache db")?);
let cache = ContentCache::open( let cache = ContentCache::open(
dirs.content_cache_dir(), dirs.content_cache_dir(),
dirs.pins_path(), dirs.pins_path(),
pdfs_core::config::DEFAULT_CACHE_BUDGET_BYTES, config.resolved_cache_budget(),
db.clone(),
) )
.context("open content cache")?; .context("open content cache")?;
let control_socket = dirs.control_socket(); let control_socket = dirs.control_socket();
@@ -272,6 +332,7 @@ fn mount_once(mountpoint: Option<PathBuf>) -> Result<pdfs_fuse::MountOutcome> {
&mountpoint, &mountpoint,
cache, cache,
&control_socket, &control_socket,
db,
username, username,
); );
@@ -308,6 +369,37 @@ fn cmd_unpin(path: PathBuf) -> Result<()> {
Ok(()) Ok(())
} }
fn cmd_transfers() -> Result<()> {
use pdfs_core::control::TransferDirection;
match control_request(CtlRequest::GetQueueStatus)? {
CtlResponse::Transfers { items } if items.is_empty() => println!("No active transfers."),
CtlResponse::Transfers { items } => {
for t in items {
let arrow = match t.direction {
TransferDirection::Download => "↓",
TransferDirection::Upload => "↑",
};
let pct = if t.bytes_total > 0 {
format!(
"{:.0}%",
100.0 * t.bytes_completed as f64 / t.bytes_total as f64
)
} else {
"—".to_string()
};
let speed = t.speed_bytes_sec as f64 / 1_048_576.0;
println!(
"{arrow} {:>4} {:>8} / {:>8} {speed:>6.2} MiB/s {}",
pct, t.bytes_completed, t.bytes_total, t.name
);
}
}
CtlResponse::Error { message } => bail!("{message}"),
other => bail!("unexpected response: {other:?}"),
}
Ok(())
}
fn cmd_pins() -> Result<()> { fn cmd_pins() -> Result<()> {
match control_request(CtlRequest::ListPins)? { match control_request(CtlRequest::ListPins)? {
CtlResponse::Pins { pins } if pins.is_empty() => println!("No pinned files."), CtlResponse::Pins { pins } if pins.is_empty() => println!("No pinned files."),
@@ -371,6 +463,88 @@ fn cmd_open_photo(uid: String) -> Result<()> {
Ok(()) Ok(())
} }
fn cmd_search(query: String, limit: usize) -> Result<()> {
match control_request(CtlRequest::Search { query, limit })? {
CtlResponse::SearchResults { hits } if hits.is_empty() => println!("(no matches)"),
CtlResponse::SearchResults { hits } => {
for h in hits {
let kind = if h.is_dir { "d" } else { "-" };
let pin = if h.pinned { "*" } else { " " };
println!("{kind}{pin} {:>12} {}", h.size, h.path);
}
}
CtlResponse::Error { message } => bail!("{message}"),
other => bail!("unexpected response: {other:?}"),
}
Ok(())
}
fn cmd_rename(path: PathBuf, new_name: String) -> Result<()> {
match control_request(CtlRequest::Rename {
path: path_arg(&path)?,
new_name,
})? {
CtlResponse::Ok { message } => println!("{message}"),
CtlResponse::Error { message } => bail!("{message}"),
other => bail!("unexpected response: {other:?}"),
}
Ok(())
}
fn cmd_move(path: PathBuf, new_parent: PathBuf) -> Result<()> {
match control_request(CtlRequest::Move {
path: path_arg(&path)?,
new_parent: path_arg(&new_parent)?,
})? {
CtlResponse::Ok { message } => println!("{message}"),
CtlResponse::Error { message } => bail!("{message}"),
other => bail!("unexpected response: {other:?}"),
}
Ok(())
}
fn cmd_rm(path: PathBuf) -> Result<()> {
match control_request(CtlRequest::Delete {
path: path_arg(&path)?,
})? {
CtlResponse::Ok { message } => println!("{message}"),
CtlResponse::Error { message } => bail!("{message}"),
other => bail!("unexpected response: {other:?}"),
}
Ok(())
}
fn cmd_mkdir(parent: PathBuf, name: String) -> Result<()> {
match control_request(CtlRequest::CreateFolder {
parent: path_arg(&parent)?,
name,
})? {
CtlResponse::Ok { message } => println!("{message}"),
CtlResponse::Error { message } => bail!("{message}"),
other => bail!("unexpected response: {other:?}"),
}
Ok(())
}
fn cmd_upload(file: PathBuf, parent: PathBuf) -> Result<()> {
let bytes = std::fs::read(&file).with_context(|| format!("read {}", file.display()))?;
let name = file
.file_name()
.and_then(|n| n.to_str())
.ok_or_else(|| anyhow!("source file has no valid name"))?
.to_owned();
match control_request(CtlRequest::UploadFile {
parent: path_arg(&parent)?,
name,
bytes,
})? {
CtlResponse::Ok { message } => println!("{message}"),
CtlResponse::Error { message } => bail!("{message}"),
other => bail!("unexpected response: {other:?}"),
}
Ok(())
}
/// Normalize a user-supplied path argument for sending to the daemon. Absolute /// Normalize a user-supplied path argument for sending to the daemon. Absolute
/// paths are canonicalized (so the daemon can strip its mountpoint prefix); /// paths are canonicalized (so the daemon can strip its mountpoint prefix);
/// relative paths are passed through and resolved against the mount root. /// relative paths are passed through and resolved against the mount root.
+1
View File
@@ -17,4 +17,5 @@ thiserror.workspace = true
tracing.workspace = true tracing.workspace = true
directories.workspace = true directories.workspace = true
sha2.workspace = true sha2.workspace = true
rusqlite.workspace = true
keyring.workspace = true keyring.workspace = true
+355 -123
View File
@@ -19,14 +19,33 @@
use std::collections::{BTreeMap, HashSet}; use std::collections::{BTreeMap, HashSet};
use std::io::{Read, Seek, SeekFrom}; use std::io::{Read, Seek, SeekFrom};
use std::path::{Path, PathBuf}; use std::path::{Path, PathBuf};
use std::time::SystemTime; use std::sync::Arc;
use std::sync::atomic::{AtomicU64, Ordering};
use std::time::{SystemTime, UNIX_EPOCH};
use proton_drive_rs::proton_sdk::ids::NodeUid; use proton_drive_rs::proton_sdk::ids::NodeUid;
use serde::{Deserialize, Serialize}; use serde::{Deserialize, Serialize};
use sha2::{Digest, Sha256}; use sha2::{Digest, Sha256};
use crate::db::Db;
use crate::error::Result; use crate::error::Result;
/// `cache_entries.kind` tag for whole-file content blobs.
const KIND_BLOB: &str = "blob";
/// `cache_entries.kind` tag for on-demand block-cache chunks.
const KIND_BLOCK: &str = "block";
/// Current wall-clock time in unix *milliseconds*, for the LRU `last_accessed`
/// column. Milliseconds (not seconds) so two cache events in the same second
/// still order correctly — a coarse second-granularity clock would make LRU
/// eviction order indeterminate under bursty access.
fn now_millis() -> i64 {
SystemTime::now()
.duration_since(UNIX_EPOCH)
.map(|d| d.as_millis() as i64)
.unwrap_or(0)
}
/// Block size for the on-demand block cache. Matches the SDK content block size /// 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 /// (`DEFAULT_BLOCK_SIZE`, 4 MiB) so each cached block maps to exactly one
/// `download_range` fetch with no straddling. /// `download_range` fetch with no straddling.
@@ -42,18 +61,22 @@ struct Meta {
size: u64, size: u64,
} }
/// One pinned file, as persisted to the pin registry. /// One pin, as carried over the control socket and listed in `status`.
#[derive(Serialize, Deserialize, Clone, Debug)] #[derive(Serialize, Deserialize, Clone, Debug)]
pub struct Pin { pub struct Pin {
/// Node uid in `volume~link` display form. /// Node uid in `volume~link` display form.
pub uid: String, pub uid: String,
/// Last path the file was pinned under, for display in `status`. Advisory /// Last path the node was pinned under, for display in `status`. Advisory
/// only — the uid is the identity. /// only — the uid is the identity.
pub path: String, pub path: String,
/// A folder pin whose whole subtree is kept on disk. `false` for a single
/// file. Defaulted so a legacy `pins.json` (file pins only) still parses.
#[serde(default)]
pub recursive: bool,
} }
/// The pin registry, persisted as JSON. Keyed by uid display string so a pin /// The legacy JSON pin registry, kept only to import a pre-P5 `pins.json` into
/// survives renames/moves of the node. /// the DB once on open. Live pins are owned by the `pins` table.
#[derive(Serialize, Deserialize, Default)] #[derive(Serialize, Deserialize, Default)]
struct PinFile { struct PinFile {
pins: BTreeMap<String, Pin>, pins: BTreeMap<String, Pin>,
@@ -78,15 +101,28 @@ pub struct ContentCache {
/// Soft cap on total blob bytes. Exceeded only transiently: a `store` /// Soft cap on total blob bytes. Exceeded only transiently: a `store`
/// evicts least-recently-used *unpinned* blobs back under the cap. `0` /// evicts least-recently-used *unpinned* blobs back under the cap. `0`
/// disables the cap. Pinned blobs are never evicted, so pins alone may push /// disables the cap. Pinned blobs are never evicted, so pins alone may push
/// the cache over budget. /// the cache over budget. Atomic so the daemon can retune it at runtime
max_bytes: u64, /// (a Settings-page change) via [`set_budget`](Self::set_budget) without
/// taking a lock on every cache read.
max_bytes: AtomicU64,
/// Unified metadata DB. Its `cache_entries` table is the LRU index: every
/// store/read/evict updates it, and the budget enforcers query it instead of
/// scanning the cache directories (plan.md P4).
db: Arc<Db>,
} }
impl ContentCache { impl ContentCache {
/// Open (and create) a cache under `content_dir`, with the pin registry at /// Open (and create) a cache under `content_dir`, with the pin registry at
/// `pins_path` and a `max_bytes` size cap (`0` = unlimited). Both parent /// `pins_path` and a `max_bytes` size cap (`0` = unlimited). Both parent
/// directories are created if missing. /// directories are created if missing. `db` is the shared metadata database
pub fn open(content_dir: PathBuf, pins_path: PathBuf, max_bytes: u64) -> Result<Self> { /// whose `cache_entries` table backs LRU eviction; the on-disk cache is
/// reconciled into it on open.
pub fn open(
content_dir: PathBuf,
pins_path: PathBuf,
max_bytes: u64,
db: Arc<Db>,
) -> Result<Self> {
std::fs::create_dir_all(&content_dir)?; std::fs::create_dir_all(&content_dir)?;
let thumb_dir = content_dir.join("thumbs"); let thumb_dir = content_dir.join("thumbs");
std::fs::create_dir_all(&thumb_dir)?; std::fs::create_dir_all(&thumb_dir)?;
@@ -100,14 +136,71 @@ impl ContentCache {
if let Some(parent) = pins_path.parent() { if let Some(parent) = pins_path.parent() {
std::fs::create_dir_all(parent)?; std::fs::create_dir_all(parent)?;
} }
Ok(Self { let cache = Self {
content_dir, content_dir,
thumb_dir, thumb_dir,
block_dir, block_dir,
scratch_dir, scratch_dir,
pins_path, pins_path,
max_bytes, max_bytes: AtomicU64::new(max_bytes),
}) db,
};
cache.reconcile()?;
cache.import_legacy_pins()?;
Ok(cache)
}
/// One-time migration of a pre-P5 `pins.json` into the DB `pins` table, then
/// delete the file so it never re-imports. Legacy pins were all whole-file
/// (non-recursive). Absent file → nothing to do.
fn import_legacy_pins(&self) -> Result<()> {
let Ok(bytes) = std::fs::read(&self.pins_path) else {
return Ok(());
};
if let Ok(file) = serde_json::from_slice::<PinFile>(&bytes) {
for (uid, pin) in file.pins {
self.db.pin_add(&uid, &pin.path, false)?;
}
}
let _ = std::fs::remove_file(&self.pins_path);
Ok(())
}
/// Rebuild the `cache_entries` LRU index from the on-disk cache. Called once
/// on open: the index is cleared, then every blob (`content_dir`) and block
/// (`block_dir`) file is re-registered with its on-disk size and last-modified
/// time as the initial LRU key. This makes the DB authoritative even after a
/// crash or an external file deletion, and picks up caches written by builds
/// predating the index. In-run accesses then refine the ordering.
fn reconcile(&self) -> Result<()> {
self.db.cache_clear()?;
for (dir, kind) in [
(&self.content_dir, KIND_BLOB),
(&self.block_dir, KIND_BLOCK),
] {
let Ok(rd) = std::fs::read_dir(dir) else {
continue;
};
for entry in rd.flatten() {
let name = entry.file_name();
let Some(name) = name.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;
}
let at = meta
.modified()
.ok()
.and_then(|t| t.duration_since(UNIX_EPOCH).ok())
.map(|d| d.as_millis() as i64)
.unwrap_or(0);
self.db.cache_touch(name, kind, meta.len(), at)?;
}
}
Ok(())
} }
/// Filesystem-safe, fixed-length cache key for a uid display string: the /// Filesystem-safe, fixed-length cache key for a uid display string: the
@@ -171,10 +264,10 @@ impl ContentCache {
len: u64, len: u64,
) -> Option<Vec<u8>> { ) -> Option<Vec<u8>> {
let blob = self.valid_blob(uid, mtime, size)?; let blob = self.valid_blob(uid, mtime, size)?;
// Record the access for LRU: bump the blob's mtime to now. Best effort — // Record the access for LRU in the cache index. Best effort — a failed
// a failed touch only makes eviction order slightly less accurate. // update only makes eviction order slightly less accurate.
let _ = self.db.cache_accessed(&Self::key(uid), now_millis());
let mut f = std::fs::File::open(&blob).ok()?; let mut f = std::fs::File::open(&blob).ok()?;
let _ = f.set_times(std::fs::FileTimes::new().set_modified(SystemTime::now()));
if offset >= size { if offset >= size {
return Some(Vec::new()); return Some(Vec::new());
} }
@@ -197,12 +290,22 @@ impl ContentCache {
std::fs::rename(&tmp, &blob)?; std::fs::rename(&tmp, &blob)?;
let meta = serde_json::to_vec(&Meta { mtime, size })?; let meta = serde_json::to_vec(&Meta { mtime, size })?;
std::fs::write(self.meta_path(uid), meta)?; std::fs::write(self.meta_path(uid), meta)?;
// Register the blob in the LRU index before enforcing the budget so the
// newest entry is weighed (and ordered most-recent) like any other.
self.db
.cache_touch(&Self::key(uid), KIND_BLOB, bytes.len() as u64, now_millis())?;
self.enforce_budget(); self.enforce_budget();
Ok(()) Ok(())
} }
/// Cache-index key for block `idx` of `uid` — the block file's name, which
/// is also its `cache_entries.cache_key`.
fn block_key(&self, uid: &NodeUid, idx: u64) -> String {
format!("{}.b{idx}", Self::key(uid))
}
fn block_blob(&self, uid: &NodeUid, idx: u64) -> PathBuf { fn block_blob(&self, uid: &NodeUid, idx: u64) -> PathBuf {
self.block_dir.join(format!("{}.b{idx}", Self::key(uid))) self.block_dir.join(self.block_key(uid, idx))
} }
fn block_meta(&self, uid: &NodeUid, idx: u64) -> PathBuf { fn block_meta(&self, uid: &NodeUid, idx: u64) -> PathBuf {
@@ -222,7 +325,10 @@ impl ContentCache {
return None; return None;
} }
let mut f = std::fs::File::open(self.block_blob(uid, idx)).ok()?; 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())); // LRU touch in the cache index (best effort).
let _ = self
.db
.cache_accessed(&self.block_key(uid, idx), now_millis());
let mut buf = Vec::new(); let mut buf = Vec::new();
f.read_to_end(&mut buf).ok()?; f.read_to_end(&mut buf).ok()?;
Some(buf) Some(buf)
@@ -249,6 +355,12 @@ impl ContentCache {
self.block_meta(uid, idx), self.block_meta(uid, idx),
serde_json::to_vec(&Meta { mtime, size })?, serde_json::to_vec(&Meta { mtime, size })?,
)?; )?;
self.db.cache_touch(
&self.block_key(uid, idx),
KIND_BLOCK,
bytes.len() as u64,
now_millis(),
)?;
self.enforce_block_budget(); self.enforce_block_budget();
Ok(()) Ok(())
} }
@@ -257,45 +369,29 @@ impl ContentCache {
/// `max_bytes`. No-op when the cap is disabled (`0`). All blocks are /// `max_bytes`. No-op when the cap is disabled (`0`). All blocks are
/// evictable — pinned files are served from whole-file blobs, never blocks. /// evictable — pinned files are served from whole-file blobs, never blocks.
fn enforce_block_budget(&self) { fn enforce_block_budget(&self) {
if self.max_bytes == 0 { let cap = self.cap();
if cap == 0 {
return; return;
} }
let mut total: u64 = 0; // LRU-ordered (oldest first) block entries straight from the index — no
let mut blobs: Vec<(PathBuf, u64, SystemTime)> = Vec::new(); // directory scan. All blocks are evictable; pinned files are served from
let Ok(rd) = std::fs::read_dir(&self.block_dir) else { // whole-file blobs, never blocks.
let Ok(entries) = self.db.cache_entries_by_kind(KIND_BLOCK) else {
return; return;
}; };
for entry in rd.flatten() { let mut total: u64 = entries.iter().map(|(_, size)| *size).sum();
let path = entry.path(); if total <= cap {
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; return;
} }
blobs.sort_by_key(|(_, _, atime)| *atime); for (key, size) in entries {
for (path, len, _) in blobs { if total <= cap {
if total <= self.max_bytes {
break; break;
} }
let meta = path.with_file_name(format!( // `key` is the block file's name (and its `.meta` sibling's stem).
"{}.meta", let _ = std::fs::remove_file(self.block_dir.join(&key));
path.file_name().unwrap_or_default().to_string_lossy() let _ = std::fs::remove_file(self.block_dir.join(format!("{key}.meta")));
)); let _ = self.db.cache_remove(&key);
let _ = std::fs::remove_file(&path); total = total.saturating_sub(size);
let _ = std::fs::remove_file(meta);
total = total.saturating_sub(len);
} }
} }
@@ -328,7 +424,10 @@ impl ContentCache {
let _ = std::fs::remove_file(self.thumb_blob(uid, ttype)); let _ = std::fs::remove_file(self.thumb_blob(uid, ttype));
let _ = std::fs::remove_file(self.thumb_meta(uid, ttype)); let _ = std::fs::remove_file(self.thumb_meta(uid, ttype));
} }
// Drop every cached block (and its meta/tmp) for this uid. // Drop every cached block (and its meta/tmp) for this uid. The block
// dir is still scanned here because eviction targets one uid's files by
// name prefix, not by LRU order; the index rows are removed in one shot
// by `cache_remove_all` below.
let prefix = format!("{}.b", Self::key(uid)); let prefix = format!("{}.b", Self::key(uid));
if let Ok(rd) = std::fs::read_dir(&self.block_dir) { if let Ok(rd) = std::fs::read_dir(&self.block_dir) {
for entry in rd.flatten() { for entry in rd.flatten() {
@@ -341,6 +440,8 @@ impl ContentCache {
} }
} }
} }
// Forget the blob and all of this uid's block rows in the LRU index.
let _ = self.db.cache_remove_all(&Self::key(uid));
} }
fn thumb_blob(&self, uid: &NodeUid, ttype: i32) -> PathBuf { fn thumb_blob(&self, uid: &NodeUid, ttype: i32) -> PathBuf {
@@ -420,104 +521,84 @@ impl ContentCache {
/// cache already fits. Pinned blobs are skipped, so a cache held entirely by /// cache already fits. Pinned blobs are skipped, so a cache held entirely by
/// pins can legitimately stay over budget. /// pins can legitimately stay over budget.
fn enforce_budget(&self) { fn enforce_budget(&self) {
if self.max_bytes == 0 { let cap = self.cap();
if cap == 0 {
return; return;
} }
// Pinned blobs (by cache key) are exempt from eviction. // Pinned blobs (by cache key) are exempt from eviction. Resolves direct
// pins plus every descendant of a recursively-pinned folder, hashed into
// cache keys to match on-disk filenames.
let pinned: HashSet<String> = self let pinned: HashSet<String> = self
.load_pins() .db
.pins .pinned_uids()
.keys() .unwrap_or_default()
.iter()
.map(|uid| Self::key_str(uid)) .map(|uid| Self::key_str(uid))
.collect(); .collect();
// Scan blobs (the `<key>` files; skip `.meta`/`.tmp` siblings), record // LRU-ordered (oldest first) blob entries from the index — no directory
// size + last-access (mtime) for each. // scan. Pinned blobs still count toward the total (so pins alone can hold
let mut total: u64 = 0; // the cache over budget) but are never chosen as victims.
let mut blobs: Vec<(PathBuf, u64, SystemTime)> = Vec::new(); let Ok(entries) = self.db.cache_entries_by_kind(KIND_BLOB) else {
let Ok(rd) = std::fs::read_dir(&self.content_dir) else {
return; return;
}; };
for entry in rd.flatten() { let mut total: u64 = entries.iter().map(|(_, size)| *size).sum();
let path = entry.path(); if total <= cap {
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();
if pinned.contains(name) {
continue; // counts toward total but is never a victim
}
let atime = meta.modified().unwrap_or(SystemTime::UNIX_EPOCH);
blobs.push((path, meta.len(), atime));
}
if total <= self.max_bytes {
return; return;
} }
// Oldest access first. for (key, size) in entries {
blobs.sort_by_key(|(_, _, atime)| *atime); if total <= cap {
for (path, len, _) in blobs {
if total <= self.max_bytes {
break; break;
} }
let _ = std::fs::remove_file(&path); if pinned.contains(&key) {
let _ = std::fs::remove_file(path.with_extension("meta")); continue; // counts toward total but is never a victim
total = total.saturating_sub(len); }
// `key` is the blob file's name (and its `.meta` sibling's stem).
let _ = std::fs::remove_file(self.content_dir.join(&key));
let _ = std::fs::remove_file(self.content_dir.join(format!("{key}.meta")));
let _ = self.db.cache_remove(&key);
total = total.saturating_sub(size);
} }
} }
fn load_pins(&self) -> PinFile { /// Whether `uid` is pinned — directly, or because an ancestor folder is
std::fs::read(&self.pins_path) /// pinned recursively (resolved in the DB against the node tree). Independent
.ok() /// of whether its blob is currently cached. Best effort: a DB error reads as
.and_then(|b| serde_json::from_slice(&b).ok()) /// "not pinned" rather than failing the calling FUSE op.
.unwrap_or_default()
}
fn save_pins(&self, pins: &PinFile) -> Result<()> {
std::fs::write(&self.pins_path, serde_json::to_vec_pretty(pins)?)?;
Ok(())
}
/// Whether `uid` is pinned (independent of whether its blob is currently
/// cached).
pub fn is_pinned(&self, uid: &NodeUid) -> bool { pub fn is_pinned(&self, uid: &NodeUid) -> bool {
self.load_pins().pins.contains_key(&uid.to_string()) self.db.is_pinned(&uid.to_string()).unwrap_or(false)
} }
/// Record `uid` as pinned under `path`. The caller is responsible for having /// Record `uid` as pinned under `path`. `recursive` marks a folder pin whose
/// already cached the content via [`store`](Self::store). /// whole subtree is kept. The caller is responsible for having cached the
pub fn add_pin(&self, uid: &NodeUid, path: &Path) -> Result<()> { /// content (a file via [`store`](Self::store); a folder's descendants by the
let mut file = self.load_pins(); /// daemon walking the subtree).
let key = uid.to_string(); pub fn add_pin(&self, uid: &NodeUid, path: &Path, recursive: bool) -> Result<()> {
file.pins.insert( self.db
key.clone(), .pin_add(&uid.to_string(), &path.display().to_string(), recursive)
Pin {
uid: key,
path: path.display().to_string(),
},
);
self.save_pins(&file)
} }
/// Drop `uid` from the pin registry and evict its blob. No-op if not pinned. /// Drop `uid` from the pin registry and evict its blob. No-op if not pinned.
/// A recursively-pinned folder's descendants are evicted by the daemon
/// (`Core::unpin`), which knows the node tree.
pub fn remove_pin(&self, uid: &NodeUid) -> Result<()> { pub fn remove_pin(&self, uid: &NodeUid) -> Result<()> {
let mut file = self.load_pins(); self.db.pin_remove(&uid.to_string())?;
if file.pins.remove(&uid.to_string()).is_some() {
self.save_pins(&file)?;
}
self.evict(uid); self.evict(uid);
Ok(()) Ok(())
} }
/// All pinned files, ordered by uid. /// All pins (files and recursive folders), ordered by uid.
pub fn list_pins(&self) -> Vec<Pin> { pub fn list_pins(&self) -> Vec<Pin> {
self.load_pins().pins.into_values().collect() self.db
.pin_list()
.unwrap_or_default()
.into_iter()
.map(|(uid, path, recursive)| Pin {
uid,
path,
recursive,
})
.collect()
} }
/// Total bytes of cached content blobs (pinned + unpinned), matching what /// Total bytes of cached content blobs (pinned + unpinned), matching what
@@ -548,7 +629,61 @@ impl ContentCache {
/// Configured soft byte cap (`0` = unlimited), for display alongside /// Configured soft byte cap (`0` = unlimited), for display alongside
/// [`usage`](Self::usage). /// [`usage`](Self::usage).
pub fn budget(&self) -> u64 { pub fn budget(&self) -> u64 {
self.max_bytes self.cap()
}
/// The current soft byte cap. A plain atomic load; the value can change
/// under [`set_budget`](Self::set_budget) while the daemon runs.
fn cap(&self) -> u64 {
self.max_bytes.load(Ordering::Relaxed)
}
/// Retune the soft byte cap (`0` = unlimited) at runtime and immediately
/// enforce it: blobs and blocks are LRU-evicted back under the new cap (a
/// lower cap frees space now; a higher one is a no-op until the next store).
/// Called by the daemon when the Settings page changes the cache budget.
pub fn set_budget(&self, bytes: u64) {
self.max_bytes.store(bytes, Ordering::Relaxed);
self.enforce_budget();
self.enforce_block_budget();
}
/// Delete every *unpinned* cached blob plus all on-demand block chunks,
/// keeping pinned files intact. Pinned files keep their whole-file blobs
/// (and are never served from blocks), so dropping all blocks is safe.
/// Returns the number of bytes freed, for a user-facing confirmation.
pub fn clear_unpinned(&self) -> u64 {
// Pinned blobs are exempt; resolve them to cache keys to match filenames.
let pinned: HashSet<String> = self
.db
.pinned_uids()
.unwrap_or_default()
.iter()
.map(|uid| Self::key_str(uid))
.collect();
let mut freed = 0u64;
// Unpinned whole-file blobs.
if let Ok(entries) = self.db.cache_entries_by_kind(KIND_BLOB) {
for (key, size) in entries {
if pinned.contains(&key) {
continue;
}
let _ = std::fs::remove_file(self.content_dir.join(&key));
let _ = std::fs::remove_file(self.content_dir.join(format!("{key}.meta")));
let _ = self.db.cache_remove(&key);
freed = freed.saturating_add(size);
}
}
// Every on-demand block (transient partial reads, re-fetched on demand).
if let Ok(entries) = self.db.cache_entries_by_kind(KIND_BLOCK) {
for (key, size) in entries {
let _ = std::fs::remove_file(self.block_dir.join(&key));
let _ = std::fs::remove_file(self.block_dir.join(format!("{key}.meta")));
let _ = self.db.cache_remove(&key);
freed = freed.saturating_add(size);
}
}
freed
} }
} }
@@ -591,10 +726,12 @@ mod tests {
fn cache_capped(max_bytes: u64) -> (ContentCache, TempDir) { fn cache_capped(max_bytes: u64) -> (ContentCache, TempDir) {
let dir = TempDir::new(); let dir = TempDir::new();
let db = Arc::new(Db::open_in_memory().unwrap());
let c = ContentCache::open( let c = ContentCache::open(
dir.path().join("content"), dir.path().join("content"),
dir.path().join("pins.json"), dir.path().join("pins.json"),
max_bytes, max_bytes,
db,
) )
.unwrap(); .unwrap();
(c, dir) (c, dir)
@@ -618,6 +755,40 @@ mod tests {
); );
} }
#[test]
fn reconcile_rebuilds_index_from_disk() {
let dir = TempDir::new();
let content = dir.path().join("content");
let pins = dir.path().join("pins.json");
// First session: unlimited cap, store three blobs onto disk.
let db1 = Arc::new(Db::open_in_memory().unwrap());
{
let c = ContentCache::open(content.clone(), pins.clone(), 0, db1).unwrap();
c.store(&uid("a"), 1, 4, b"aaaa").unwrap();
c.store(&uid("b"), 1, 4, b"bbbb").unwrap();
c.store(&uid("c"), 1, 4, b"cccc").unwrap();
}
// Second session with a *fresh* index (as if predating P4, or after a
// crash) but the same on-disk cache: reconcile repopulates from disk.
let db2 = Arc::new(Db::open_in_memory().unwrap());
assert!(db2.cache_entries_by_kind("blob").unwrap().is_empty());
let c = ContentCache::open(content, pins, 8, db2.clone()).unwrap();
assert_eq!(db2.cache_entries_by_kind("blob").unwrap().len(), 3);
// With the index rebuilt, a further store enforces the budget against the
// reconciled entries instead of leaking them.
c.store(&uid("d"), 1, 4, b"dddd").unwrap();
let total: u64 = db2
.cache_entries_by_kind("blob")
.unwrap()
.iter()
.map(|(_, s)| s)
.sum();
assert!(total <= 8, "budget enforced after reconcile, total={total}");
}
#[test] #[test]
fn stale_metadata_is_a_miss() { fn stale_metadata_is_a_miss() {
let (c, _d) = cache(); let (c, _d) = cache();
@@ -630,12 +801,36 @@ mod tests {
assert!(c.read_range(&u, 100, 4, 0, 3).is_none()); assert!(c.read_range(&u, 100, 4, 0, 3).is_none());
} }
#[test]
fn legacy_pins_json_imported_then_removed() {
let dir = TempDir::new();
let pins = dir.path().join("pins.json");
// Hand-write a pre-P5 pins.json (file pins only, no `recursive` field).
std::fs::write(
&pins,
r#"{"pins":{"vol~a":{"uid":"vol~a","path":"docs/a.txt"}}}"#,
)
.unwrap();
let db = Arc::new(Db::open_in_memory().unwrap());
let c = ContentCache::open(dir.path().join("content"), pins.clone(), 0, db).unwrap();
// Imported into the DB as a non-recursive pin...
assert!(c.is_pinned(&uid("a")));
let listed = c.list_pins();
assert_eq!(listed.len(), 1);
assert_eq!(listed[0].path, "docs/a.txt");
assert!(!listed[0].recursive);
// ...and the legacy file is gone so it never re-imports.
assert!(!pins.exists());
}
#[test] #[test]
fn pin_lifecycle_evicts_on_unpin() { fn pin_lifecycle_evicts_on_unpin() {
let (c, _d) = cache(); let (c, _d) = cache();
let u = uid("a"); let u = uid("a");
c.store(&u, 100, 3, b"abc").unwrap(); c.store(&u, 100, 3, b"abc").unwrap();
c.add_pin(&u, Path::new("docs/a.txt")).unwrap(); c.add_pin(&u, Path::new("docs/a.txt"), false).unwrap();
assert!(c.is_pinned(&u)); assert!(c.is_pinned(&u));
assert_eq!(c.list_pins().len(), 1); assert_eq!(c.list_pins().len(), 1);
assert_eq!(c.list_pins()[0].path, "docs/a.txt"); assert_eq!(c.list_pins()[0].path, "docs/a.txt");
@@ -675,7 +870,7 @@ mod tests {
let (a, b, d) = (uid("a"), uid("b"), uid("d")); let (a, b, d) = (uid("a"), uid("b"), uid("d"));
c.store(&a, 1, 4, b"aaaa").unwrap(); c.store(&a, 1, 4, b"aaaa").unwrap();
c.add_pin(&a, Path::new("a")).unwrap(); c.add_pin(&a, Path::new("a"), false).unwrap();
std::thread::sleep(std::time::Duration::from_millis(10)); std::thread::sleep(std::time::Duration::from_millis(10));
c.store(&b, 1, 4, b"bbbb").unwrap(); c.store(&b, 1, 4, b"bbbb").unwrap();
// Over budget: `a` is pinned (oldest) so `b` must be the victim. // Over budget: `a` is pinned (oldest) so `b` must be the victim.
@@ -686,6 +881,43 @@ mod tests {
assert!(c.is_cached(&d, 1, 4)); assert!(c.is_cached(&d, 1, 4));
} }
#[test]
fn clear_unpinned_keeps_pinned() {
let (c, _d) = cache();
let (a, b) = (uid("a"), uid("b"));
c.store(&a, 1, 4, b"aaaa").unwrap();
c.add_pin(&a, Path::new("a"), false).unwrap();
c.store(&b, 1, 4, b"bbbb").unwrap();
// A stray on-demand block of the unpinned file is purged too.
c.store_block(&b, 1, 8, 0, b"bbbb").unwrap();
let freed = c.clear_unpinned();
assert_eq!(freed, 8, "one 4-byte blob + one 4-byte block freed");
assert!(c.is_cached(&a, 1, 4), "pinned blob survives purge");
assert!(!c.is_cached(&b, 1, 4), "unpinned blob purged");
assert!(c.cached_block(&b, 1, 8, 0).is_none(), "block purged");
}
#[test]
fn set_budget_evicts_immediately() {
// Start unlimited so three blobs all fit, then tighten the cap.
let (c, _d) = cache();
let (a, b, d) = (uid("a"), uid("b"), uid("d"));
c.store(&a, 1, 4, b"aaaa").unwrap();
std::thread::sleep(std::time::Duration::from_millis(10));
c.store(&b, 1, 4, b"bbbb").unwrap();
std::thread::sleep(std::time::Duration::from_millis(10));
c.store(&d, 1, 4, b"dddd").unwrap();
// Tightening to 8 bytes evicts the LRU unpinned blob(s) now, not on the
// next store.
c.set_budget(8);
assert!(!c.is_cached(&a, 1, 4), "oldest blob evicted on tighten");
assert!(c.is_cached(&b, 1, 4));
assert!(c.is_cached(&d, 1, 4));
assert_eq!(c.budget(), 8);
}
#[test] #[test]
fn thumbnail_store_read_and_evict() { fn thumbnail_store_read_and_evict() {
let (c, _d) = cache(); let (c, _d) = cache();
+50
View File
@@ -24,6 +24,16 @@ pub const DEFAULT_CACHE_BUDGET_BYTES: u64 = 5 * 1024 * 1024 * 1024;
pub struct AppConfig { pub struct AppConfig {
pub app_version: String, pub app_version: String,
pub user_agent: String, pub user_agent: String,
/// Soft cap on the on-disk content cache, in bytes (`0` = unlimited).
/// `None` means "use [`DEFAULT_CACHE_BUDGET_BYTES`]"; the Settings page
/// writes an explicit value here. Defaulted for configs predating the field.
#[serde(default)]
pub cache_budget: Option<u64>,
/// Mountpoint the daemon mounts at. `None` means
/// [`AppDirs::default_mountpoint`]; the Settings page writes an explicit
/// path here. Defaulted for configs predating the field.
#[serde(default)]
pub mountpoint: Option<String>,
} }
impl Default for AppConfig { impl Default for AppConfig {
@@ -31,10 +41,20 @@ impl Default for AppConfig {
Self { Self {
app_version: APP_VERSION.to_string(), app_version: APP_VERSION.to_string(),
user_agent: USER_AGENT.to_string(), user_agent: USER_AGENT.to_string(),
cache_budget: None,
mountpoint: None,
} }
} }
} }
impl AppConfig {
/// The effective cache budget in bytes: the user's explicit choice, or
/// [`DEFAULT_CACHE_BUDGET_BYTES`] when unset.
pub fn resolved_cache_budget(&self) -> u64 {
self.cache_budget.unwrap_or(DEFAULT_CACHE_BUDGET_BYTES)
}
}
/// Resolved XDG locations for state, cache, and the default mountpoint. /// Resolved XDG locations for state, cache, and the default mountpoint.
pub struct AppDirs { pub struct AppDirs {
dirs: ProjectDirs, dirs: ProjectDirs,
@@ -70,6 +90,18 @@ impl AppDirs {
default_config default_config
} }
/// Persist `config` to [`config_path`](Self::config_path), creating the
/// config dir if missing. The Settings page calls this after editing the
/// cache budget or mountpoint; the daemon re-reads the file on its next mount.
pub fn save_config(&self, config: &AppConfig) -> Result<()> {
let path = self.config_path();
if let Some(parent) = path.parent() {
std::fs::create_dir_all(parent)?;
}
std::fs::write(&path, serde_json::to_string_pretty(config)?)?;
Ok(())
}
/// Persistent state: inode map / cache index DB lives here. /// Persistent state: inode map / cache index DB lives here.
pub fn state_dir(&self) -> PathBuf { pub fn state_dir(&self) -> PathBuf {
// `state_dir` is Linux-only in `directories`; fall back to data dir. // `state_dir` is Linux-only in `directories`; fall back to data dir.
@@ -94,6 +126,12 @@ impl AppDirs {
self.state_dir().join("pins.json") self.state_dir().join("pins.json")
} }
/// Unified SQLite metadata cache (inodes, FTS, cache LRU, pins). Lives in
/// persistent state next to `control.sock`; only the daemon writes it.
pub fn db_path(&self) -> PathBuf {
self.state_dir().join("cache.db")
}
/// Unix domain socket the mount daemon listens on for CLI control commands. /// Unix domain socket the mount daemon listens on for CLI control commands.
pub fn control_socket(&self) -> PathBuf { pub fn control_socket(&self) -> PathBuf {
self.state_dir().join("control.sock") self.state_dir().join("control.sock")
@@ -111,6 +149,16 @@ impl AppDirs {
.unwrap_or_else(|| PathBuf::from("/tmp/ProtonDrive")) .unwrap_or_else(|| PathBuf::from("/tmp/ProtonDrive"))
} }
/// The effective mountpoint: the config's explicit path, or
/// [`default_mountpoint`](Self::default_mountpoint) when unset.
pub fn resolved_mountpoint(&self, config: &AppConfig) -> PathBuf {
config
.mountpoint
.as_ref()
.map(PathBuf::from)
.unwrap_or_else(|| self.default_mountpoint())
}
/// Create state + cache dirs if missing. /// Create state + cache dirs if missing.
pub fn ensure(&self) -> Result<()> { pub fn ensure(&self) -> Result<()> {
std::fs::create_dir_all(self.state_dir())?; std::fs::create_dir_all(self.state_dir())?;
@@ -129,6 +177,8 @@ mod tests {
let config = AppConfig { let config = AppConfig {
app_version: "external-drive-test-client@1.0.0".to_string(), app_version: "external-drive-test-client@1.0.0".to_string(),
user_agent: "test-agent/1.0".to_string(), user_agent: "test-agent/1.0".to_string(),
cache_budget: Some(1234),
mountpoint: Some("/tmp/x".to_string()),
}; };
let json = serde_json::to_string(&config).unwrap(); let json = serde_json::to_string(&config).unwrap();
let decoded: AppConfig = serde_json::from_str(&json).unwrap(); let decoded: AppConfig = serde_json::from_str(&json).unwrap();
+135
View File
@@ -49,10 +49,74 @@ pub enum Request {
PhotosTimeline { offset: usize, limit: usize }, PhotosTimeline { offset: usize, limit: usize },
/// Download a photo's full content into the cache; replies with its path. /// Download a photo's full content into the cache; replies with its path.
OpenPhoto { uid: String }, OpenPhoto { uid: String },
/// Upload a photo with the given name, media type, and content bytes.
UploadPhoto {
name: String,
media_type: String,
bytes: Vec<u8>,
capture_time: Option<i64>,
},
/// Download a Drive file's full content into the cache; replies with the /// Download a Drive file's full content into the cache; replies with the
/// on-disk path so the front-end can open it with the default app. `path` /// on-disk path so the front-end can open it with the default app. `path`
/// is mountpoint-relative. /// is mountpoint-relative.
OpenFile { path: String }, OpenFile { path: String },
/// Full-text search node names against the daemon's local metadata index.
/// `limit` caps the number of hits returned. Replies with [`Response::SearchResults`].
Search { query: String, limit: usize },
/// Rename a file or folder. `path` is mountpoint-relative; `new_name` is a
/// single path component (no separators). Replies with [`Response::Ok`].
Rename { path: String, new_name: String },
/// Move a file or folder into a new parent folder. Both `path` and
/// `new_parent` are mountpoint-relative. Replies with [`Response::Ok`].
Move { path: String, new_parent: String },
/// Trash a file or folder. `path` is mountpoint-relative. Replies with
/// [`Response::Ok`].
Delete { path: String },
/// Create a new folder named `name` under the mountpoint-relative `parent`.
/// Replies with [`Response::Ok`].
CreateFolder { parent: String, name: String },
/// Upload a file named `name` with content `bytes` into the
/// mountpoint-relative `parent` folder. Replies with [`Response::Ok`].
UploadFile {
parent: String,
name: String,
bytes: Vec<u8>,
},
/// Delete all unpinned cached blobs and on-demand blocks, keeping pinned
/// files intact. Replies with [`Response::Ok`] reporting the bytes freed.
PurgeCache,
/// Retune the on-disk cache's soft byte cap at runtime (`0` = unlimited) and
/// persist it to config so the next mount keeps it. Replies with
/// [`Response::Ok`].
SetCacheBudget { bytes: u64 },
/// Snapshot the daemon's in-flight transfers (active uploads/downloads).
/// Replies with [`Response::Transfers`]. Cheap to poll: the daemon keeps the
/// registry in memory, so a front-end can render a live progress widget.
GetQueueStatus,
}
/// Which way an active transfer is moving, in a [`TransferItem`].
#[derive(Serialize, Deserialize, Debug, Clone, Copy, PartialEq, Eq)]
pub enum TransferDirection {
Download,
Upload,
}
/// One in-flight transfer in a [`Response::Transfers`] snapshot.
#[derive(Serialize, Deserialize, Debug, Clone)]
pub struct TransferItem {
/// Node uid in `volume~link` form (empty for an upload whose uid isn't known
/// until the draft is sealed).
pub uid: String,
/// File name being transferred.
pub name: String,
pub direction: TransferDirection,
/// Bytes moved so far.
pub bytes_completed: u64,
/// Total bytes expected, or `0` when unknown (indeterminate progress).
pub bytes_total: u64,
/// Average throughput since the transfer began, bytes per second.
pub speed_bytes_sec: u64,
} }
/// One entry in a [`Request::ListDir`] listing. /// One entry in a [`Request::ListDir`] listing.
@@ -67,8 +131,35 @@ pub struct DirEntry {
pub modified: i64, pub modified: i64,
/// Whether the file is pinned to this device. /// Whether the file is pinned to this device.
pub pinned: bool, pub pinned: bool,
/// Whether the file's full content is present in the local cache (a current,
/// non-stale blob). Always false for folders. Defaulted for wire-compat with
/// clients/daemons predating the field.
#[serde(default)]
pub cached: bool,
/// Node uid in `volume~link` form, for follow-up requests. /// Node uid in `volume~link` form, for follow-up requests.
pub uid: String, pub uid: String,
/// Full mountpoint-relative path. Empty for a [`Request::ListDir`] listing
/// (the entry lives in the requested directory, so the caller derives the
/// path by joining its name); populated when an entry can live anywhere in
/// the tree, as for search hits rendered through the browser.
#[serde(default)]
pub path: String,
}
/// One hit in a [`Request::Search`] result. Like [`DirEntry`] but carries the
/// full mountpoint-relative `path` (a hit can live anywhere in the tree), so the
/// front-end can navigate to or open it directly.
#[derive(Serialize, Deserialize, Debug, Clone)]
pub struct SearchHit {
pub name: String,
/// Mountpoint-relative path (`/`-joined, no leading slash).
pub path: String,
pub is_dir: bool,
pub size: u64,
pub modified: i64,
pub pinned: bool,
/// Node uid in `volume~link` form.
pub uid: String,
} }
/// One photo in a [`Request::PhotosTimeline`] page. /// One photo in a [`Request::PhotosTimeline`] page.
@@ -113,6 +204,10 @@ pub enum Response {
}, },
/// An on-disk path the front-end can open (e.g. a downloaded photo). /// An on-disk path the front-end can open (e.g. a downloaded photo).
FilePath { path: String }, FilePath { path: String },
/// Full-text search results (reply to [`Request::Search`]).
SearchResults { hits: Vec<SearchHit> },
/// A snapshot of in-flight transfers (reply to [`Request::GetQueueStatus`]).
Transfers { items: Vec<TransferItem> },
/// The request failed. /// The request failed.
Error { message: String }, Error { message: String },
} }
@@ -134,3 +229,43 @@ pub fn send(socket: &Path, req: &Request) -> Result<Response> {
reader.read_line(&mut resp)?; reader.read_line(&mut resp)?;
Ok(serde_json::from_str(resp.trim())?) Ok(serde_json::from_str(resp.trim())?)
} }
#[cfg(test)]
mod tests {
use super::*;
/// The mutation requests must survive a line-delimited JSON round-trip, since
/// that is exactly how they cross the control socket.
#[test]
fn mutation_requests_roundtrip() {
let reqs = [
Request::Rename {
path: "a/b.txt".into(),
new_name: "c.txt".into(),
},
Request::Move {
path: "a/b.txt".into(),
new_parent: "d".into(),
},
Request::Delete {
path: "a/b.txt".into(),
},
Request::CreateFolder {
parent: "a".into(),
name: "new".into(),
},
Request::UploadFile {
parent: "a".into(),
name: "f.bin".into(),
bytes: vec![0, 1, 2, 255],
},
];
for req in reqs {
let line = serde_json::to_string(&req).unwrap();
assert!(!line.contains('\n'), "wire form must be a single line");
let back: Request = serde_json::from_str(&line).unwrap();
// Round-trip is lossless: re-serializing yields the same bytes.
assert_eq!(line, serde_json::to_string(&back).unwrap());
}
}
}
File diff suppressed because it is too large. Load diff
+3
View File
@@ -16,6 +16,9 @@ pub enum Error {
#[error("io: {0}")] #[error("io: {0}")]
Io(#[from] std::io::Error), Io(#[from] std::io::Error),
#[error("database: {0}")]
Db(#[from] rusqlite::Error),
/// No persisted session found — caller must run an interactive login. /// No persisted session found — caller must run an interactive login.
#[error("not logged in (no saved session)")] #[error("not logged in (no saved session)")]
NotLoggedIn, NotLoggedIn,
+1
View File
@@ -6,6 +6,7 @@ pub mod auth;
pub mod cache; pub mod cache;
pub mod config; pub mod config;
pub mod control; pub mod control;
pub mod db;
pub mod error; pub mod error;
pub mod service; pub mod service;
+20
View File
@@ -79,6 +79,26 @@ pub fn is_failed() -> bool {
systemctl_query(&["is-failed"]) systemctl_query(&["is-failed"])
} }
/// Whether the unit is enabled to start on login/boot. `systemctl is-enabled`
/// exits 0 only for an enabled unit, so the Settings page can render the
/// "Start on login" switch in its true state.
pub fn is_enabled() -> bool {
systemctl_query(&["is-enabled"])
}
/// Enable the service so it auto-starts on future logins, without necessarily
/// starting it now beyond what `enable` does. Used by the Settings "Start on
/// login" switch; pairs with [`disable`](disable).
pub fn enable() -> bool {
systemctl(&["enable"])
}
/// Disable auto-start without stopping the running mount. Used by the Settings
/// "Start on login" switch turning off; the current session stays mounted.
pub fn disable() -> bool {
systemctl(&["disable"])
}
/// Disable the service so it no longer auto-starts, and stop it now. Called on /// Disable the service so it no longer auto-starts, and stop it now. Called on
/// logout. /// logout.
pub fn disable_stop() -> bool { pub fn disable_stop() -> bool {
+679 -48
View File
@@ -49,13 +49,22 @@ use fuser::{
ReplyWrite, Request, Session, TimeOrNow, WriteFlags, ReplyWrite, Request, Session, TimeOrNow, WriteFlags,
}; };
use pdfs_core::cache::{BLOCK_SIZE, ContentCache}; use pdfs_core::cache::{BLOCK_SIZE, ContentCache};
use pdfs_core::control::{DirEntry, PhotoItem, Request as CtlRequest, Response as CtlResponse}; use pdfs_core::config::AppDirs;
use pdfs_core::control::{
DirEntry, PhotoItem, Request as CtlRequest, Response as CtlResponse, SearchHit,
TransferDirection,
};
use pdfs_core::db::{Db, StoredNode};
use proton_drive_rs::proton_sdk::error::ProtonError;
use proton_drive_rs::proton_sdk::ids::{DriveEventId, LinkId, NodeUid, VolumeId}; use proton_drive_rs::proton_sdk::ids::{DriveEventId, LinkId, NodeUid, VolumeId};
use proton_drive_rs::{ use proton_drive_rs::{
DriveEvent, DriveEventScopeId, Node, NodeKind, PhotosTimelineItem, ProtonDriveClient, DriveEvent, DriveEventScopeId, Node, NodeKind, PhotosTimelineItem, ProtonDriveClient,
ProtonPhotosClient, ThumbnailType, ProtonPhotosClient, ThumbnailType,
}; };
mod transfers;
use tracing::{debug, error, info, warn}; use tracing::{debug, error, info, warn};
use transfers::{CountingReader, CountingWriter, TransferRegistry};
/// Attribute/entry cache lifetime handed back to the kernel. Long because the /// Attribute/entry cache lifetime handed back to the kernel. Long because the
/// Phase 2 event poller actively invalidates changed inodes; without a remote /// Phase 2 event poller actively invalidates changed inodes; without a remote
@@ -186,11 +195,19 @@ struct State {
/// have no entry here. /// have no entry here.
handles: HashMap<u64, WriteHandle>, handles: HashMap<u64, WriteHandle>,
next_fh: u64, next_fh: u64,
/// Unified SQLite metadata cache. Every map mutation below writes through to
/// it inside the `State` lock so the DB stays the authoritative copy across
/// restarts (see plan.md P1).
db: Arc<Db>,
} }
impl State { impl State {
/// Allocate (or reuse) a stable inode for a node and store its metadata. /// Allocate (or reuse) a stable inode for a node and store its metadata,
/// writing the node through to the DB.
fn intern(&mut self, parent: u64, node: Node) -> u64 { fn intern(&mut self, parent: u64, node: Node) -> u64 {
if let Err(e) = self.db.upsert_node(&node) {
warn!(uid = %node.uid, error = %e, "db upsert_node failed");
}
if let Some(&ino) = self.by_uid.get(&node.uid) { if let Some(&ino) = self.by_uid.get(&node.uid) {
if let Some(e) = self.entries.get_mut(&ino) { if let Some(e) = self.entries.get_mut(&ino) {
e.node = node; e.node = node;
@@ -213,10 +230,14 @@ impl State {
} }
/// Forget a node entirely: drop its inode, its uid mapping, its own cached /// Forget a node entirely: drop its inode, its uid mapping, its own cached
/// listing, and its slot in its parent's listing. Returns `(parent_ino, /// listing, its slot in its parent's listing, and its DB row. Returns
/// name)` when the node was known, so the caller can notify the kernel. /// `(parent_ino, name)` when the node was known, so the caller can notify
/// the kernel.
fn forget(&mut self, uid: &NodeUid) -> Option<(u64, String)> { fn forget(&mut self, uid: &NodeUid) -> Option<(u64, String)> {
let ino = self.by_uid.remove(uid)?; let ino = self.by_uid.remove(uid)?;
if let Err(e) = self.db.delete_node(uid) {
warn!(%uid, error = %e, "db delete_node failed");
}
let entry = self.entries.remove(&ino)?; let entry = self.entries.remove(&ino)?;
self.children.remove(&ino); self.children.remove(&ino);
if let Some(kids) = self.children.get_mut(&entry.parent) { if let Some(kids) = self.children.get_mut(&entry.parent) {
@@ -225,6 +246,20 @@ impl State {
Some((entry.parent, entry.node.name)) Some((entry.parent, entry.node.name))
} }
/// Drop a directory's cached child listing and mark it unlisted in the DB,
/// so the next access re-enumerates instead of trusting a stale listing.
fn invalidate_listing(&mut self, ino: u64) {
if self.children.remove(&ino).is_none() {
return;
}
if let Some(e) = self.entries.get(&ino) {
let uid = e.uid.clone();
if let Err(err) = self.db.set_listed(&uid, false) {
warn!(%uid, error = %err, "db set_listed(false) failed");
}
}
}
/// Update a file entry's recorded plaintext size so `getattr` reflects an /// Update a file entry's recorded plaintext size so `getattr` reflects an
/// in-progress write before the new revision is sealed. /// in-progress write before the new revision is sealed.
fn set_size(&mut self, ino: u64, size: u64) { fn set_size(&mut self, ino: u64, size: u64) {
@@ -255,10 +290,17 @@ struct Core {
rt: tokio::runtime::Handle, rt: tokio::runtime::Handle,
state: Arc<Mutex<State>>, state: Arc<Mutex<State>>,
cache: Arc<ContentCache>, cache: Arc<ContentCache>,
/// Unified SQLite metadata cache: the persistence layer behind the in-memory
/// `State` maps. Every mutation writes through here, and the maps rehydrate
/// from it on mount (plan.md P1).
db: Arc<Db>,
/// Cached full photos timeline (newest first), so paged `PhotosTimeline` /// Cached full photos timeline (newest first), so paged `PhotosTimeline`
/// requests slice memory instead of re-fetching the whole timeline per page. /// requests slice memory instead of re-fetching the whole timeline per page.
/// `None` until first fetched; refreshed once older than [`TIMELINE_TTL`]. /// `None` until first fetched; refreshed once older than [`TIMELINE_TTL`].
timeline: Arc<Mutex<Option<TimelineCache>>>, timeline: Arc<Mutex<Option<TimelineCache>>>,
/// In-flight upload/download progress, served to `GetQueueStatus`. Shared
/// across the FUSE session and the control-socket task.
transfers: Arc<TransferRegistry>,
} }
/// The whole photos timeline plus when it was fetched, for TTL freshness. /// The whole photos timeline plus when it was fetched, for TTL freshness.
@@ -268,6 +310,71 @@ struct TimelineCache {
} }
impl Core { impl Core {
/// Rehydrate the in-memory `State` maps from the DB on mount, so a cold
/// start serves previously-seen metadata (stable inodes, instant listings)
/// without re-hitting the API. The root inode is already installed by
/// [`ProtonFs::new`]; this fills in every other persisted node and rebuilds
/// the child listings of folders the DB records as fully enumerated.
fn hydrate(&self) {
let stored = match self.db.load_all() {
Ok(s) => s,
Err(e) => {
warn!(error = %e, "db load_all failed; mounting cold");
return;
}
};
if stored.is_empty() {
return;
}
let mut st = self.state.lock().unwrap();
// Pass 1: assign a stable inode to every uid (root is already mapped).
for sn in &stored {
if st.by_uid.contains_key(&sn.node.uid) {
continue;
}
let ino = st.next_ino;
st.next_ino += 1;
st.by_uid.insert(sn.node.uid.clone(), ino);
}
// Pass 2: materialize entries, resolving each parent's inode by uid.
// Track folders flagged complete so their listings rebuild in pass 3.
let mut listed_dirs: Vec<u64> = Vec::new();
for sn in stored {
let StoredNode { node, listed } = sn;
let Some(&ino) = st.by_uid.get(&node.uid) else {
continue;
};
if listed && node.is_folder() {
listed_dirs.push(ino);
}
// The root entry is owned by `ProtonFs::new`; don't overwrite it.
if ino == ROOT_INO {
continue;
}
let parent = node
.parent_uid
.as_ref()
.and_then(|p| st.by_uid.get(p).copied())
.unwrap_or(ROOT_INO);
let uid = node.uid.clone();
st.entries.insert(ino, Entry { uid, parent, node });
}
// Pass 3: rebuild child listings for fully-enumerated folders.
for dir_ino in listed_dirs {
let kids: Vec<u64> = st
.entries
.iter()
.filter(|(_, e)| e.parent == dir_ino && !e.node.trashed)
.map(|(&ino, _)| ino)
.collect();
st.children.insert(dir_ino, kids);
}
info!(nodes = st.entries.len(), "hydrated metadata cache from db");
}
/// Enumerate `ino`'s children from the remote and cache them. No-op if the /// Enumerate `ino`'s children from the remote and cache them. No-op if the
/// directory has already been listed. Network I/O happens without the lock /// directory has already been listed. Network I/O happens without the lock
/// held so concurrent metadata reads aren't blocked behind a fetch. /// held so concurrent metadata reads aren't blocked behind a fetch.
@@ -283,6 +390,29 @@ impl Core {
} }
}; };
// Offline fast path: a folder the DB still records as fully enumerated
// can be rebuilt from disk without hitting the API, even if its listing
// was trimmed from the hot cache mid-run.
match self.db.children_if_listed(&folder_uid) {
Ok(Some(nodes)) => {
let mut st = self.state.lock().unwrap();
if st.children.contains_key(&ino) {
return Ok(());
}
let mut child_inos = Vec::with_capacity(nodes.len());
for node in nodes {
if node.trashed {
continue;
}
child_inos.push(st.intern(ino, node));
}
st.children.insert(ino, child_inos);
return Ok(());
}
Ok(None) => {}
Err(e) => warn!(%folder_uid, error = %e, "db children_if_listed failed"),
}
let uids = self let uids = self
.rt .rt
.block_on(self.client.enumerate_folder_children_node_uids(&folder_uid)) .block_on(self.client.enumerate_folder_children_node_uids(&folder_uid))
@@ -311,6 +441,11 @@ impl Core {
child_inos.push(st.intern(ino, node)); child_inos.push(st.intern(ino, node));
} }
st.children.insert(ino, child_inos); st.children.insert(ino, child_inos);
// Record the listing as complete so a later restart (or a trimmed hot
// cache) can rebuild it from the DB without the API.
if let Err(e) = self.db.set_listed(&folder_uid, true) {
warn!(%folder_uid, error = %e, "db set_listed(true) failed");
}
Ok(()) Ok(())
} }
@@ -393,23 +528,62 @@ impl Core {
let mut out = Vec::with_capacity((end - offset) as usize); let mut out = Vec::with_capacity((end - offset) as usize);
let first = offset / BLOCK_SIZE; let first = offset / BLOCK_SIZE;
let last = (end - 1) / BLOCK_SIZE; let last = (end - 1) / BLOCK_SIZE;
// Collect the blocks overlapping the request, serving any already cached
// and fetching the rest concurrently. A multi-block read (e.g. a media
// player buffering, or a large sequential read split into one FUSE call)
// would otherwise stall on each block round-trip in turn; downloading the
// misses in parallel saturates the connection and bounds latency at the
// slowest single block instead of their sum.
let mut blocks: Vec<Option<Vec<u8>>> = Vec::with_capacity((last - first + 1) as usize);
let mut misses: Vec<u64> = Vec::new();
for bidx in first..=last { for bidx in first..=last {
let bstart = bidx * BLOCK_SIZE; match self.cache.cached_block(uid, mtime, fsize, bidx) {
let block = match self.cache.cached_block(uid, mtime, fsize, bidx) { Some(b) => blocks.push(Some(b)),
Some(b) => b,
None => { None => {
let blen = BLOCK_SIZE.min(fsize - bstart); blocks.push(None);
let bytes = self misses.push(bidx);
.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
} }
}; }
}
if !misses.is_empty() {
let fetched = self.rt.block_on(async {
let mut set = tokio::task::JoinSet::new();
for &bidx in &misses {
let client = self.client.clone();
let uid = uid.clone();
let bstart = bidx * BLOCK_SIZE;
let blen = BLOCK_SIZE.min(fsize - bstart);
set.spawn(async move {
client
.download_range(&uid, bstart, blen)
.await
.map(|bytes| (bidx, bytes))
.map_err(|e| {
warn!(%uid, bstart, blen, error = %e, "download_range failed");
Errno::EIO
})
});
}
let mut out = Vec::with_capacity(misses.len());
while let Some(joined) = set.join_next().await {
// A join error means the task panicked; surface it as EIO.
out.push(joined.map_err(|_| Errno::EIO)??);
}
Ok::<_, Errno>(out)
})?;
for (bidx, bytes) in fetched {
let _ = self.cache.store_block(uid, mtime, fsize, bidx, &bytes);
blocks[(bidx - first) as usize] = Some(bytes);
}
}
for (i, block) in blocks.into_iter().enumerate() {
let bidx = first + i as u64;
let bstart = bidx * BLOCK_SIZE;
// Every slot is populated: cache hits up front, misses by the fetch above.
let block = block.expect("block fetched or cached");
let s = (offset.max(bstart) - bstart) as usize; let s = (offset.max(bstart) - bstart) as usize;
let e = (end.min(bstart + block.len() as u64) - bstart) as usize; let e = (end.min(bstart + block.len() as u64) - bstart) as usize;
if s < e { if s < e {
@@ -515,6 +689,17 @@ impl Core {
error!(%uid, error = %e, "reopen scratch file failed"); error!(%uid, error = %e, "reopen scratch file failed");
Errno::EIO Errno::EIO
})?; })?;
let name = {
let st = self.state.lock().unwrap();
st.entries
.get(&ino)
.map(|e| e.node.name.clone())
.unwrap_or_default()
};
let guard = self
.transfers
.begin(name, uid.to_string(), TransferDirection::Upload, len);
let reader = CountingReader::new(reader, &guard);
self.rt self.rt
.block_on(self.client.upload_new_revision_from( .block_on(self.client.upload_new_revision_from(
&uid, &uid,
@@ -527,6 +712,7 @@ impl Core {
error!(%uid, error = %e, "upload new revision failed"); error!(%uid, error = %e, "upload new revision failed");
Errno::EIO Errno::EIO
})?; })?;
drop(guard);
let now = now_secs(); let now = now_secs();
{ {
let mut st = self.state.lock().unwrap(); let mut st = self.state.lock().unwrap();
@@ -553,38 +739,131 @@ impl Core {
Ok(()) Ok(())
} }
/// Pin the file at mountpoint-relative `rel`: download its full plaintext /// Download a whole file's plaintext, registering the transfer so
/// into the content cache and record it in the pin registry. Returns the /// `GetQueueStatus` can report its progress. `total` is the expected size
/// resolved node name on success. /// (`0` if unknown). Streams through [`download_file_to`] so each block ticks
/// the progress counter.
///
/// [`download_file_to`]: ProtonDriveClient::download_file_to
fn download_file_tracked(
&self,
uid: &NodeUid,
name: &str,
total: u64,
) -> std::result::Result<Vec<u8>, ProtonError> {
let guard = self
.transfers
.begin(name, uid.to_string(), TransferDirection::Download, total);
let mut out = CountingWriter::new(Vec::with_capacity(total as usize), &guard);
self.rt
.block_on(self.client.download_file_to(uid, &mut out))?;
Ok(out.into_inner())
}
/// Like [`download_file_tracked`] for a photo, streaming through the photos
/// client's [`download_photo_to`].
///
/// [`download_file_tracked`]: Core::download_file_tracked
/// [`download_photo_to`]: ProtonPhotosClient::download_photo_to
fn download_photo_tracked(
&self,
photos: &ProtonPhotosClient,
uid: &NodeUid,
name: &str,
total: u64,
) -> std::result::Result<Vec<u8>, ProtonError> {
let guard = self
.transfers
.begin(name, uid.to_string(), TransferDirection::Download, total);
let mut out = CountingWriter::new(Vec::with_capacity(total as usize), &guard);
self.rt.block_on(photos.download_photo_to(uid, &mut out))?;
Ok(out.into_inner())
}
/// Pin the node at mountpoint-relative `rel`. A file downloads its full
/// plaintext into the content cache; a folder records a recursive pin and
/// downloads every descendant file (selective sync). Returns a human message.
fn pin(&self, rel: &Path) -> Result<String, String> { fn pin(&self, rel: &Path) -> Result<String, String> {
let (ino, uid) = self let (ino, uid) = self
.resolve_path(rel) .resolve_path(rel)
.map_err(|e| format!("resolve path: {e:?}"))?; .map_err(|e| format!("resolve path: {e:?}"))?;
let (name, mtime, size) = { let (name, is_folder, mtime, size) = {
let st = self.state.lock().unwrap(); let st = self.state.lock().unwrap();
let e = st.entries.get(&ino).ok_or("node vanished")?; let e = st.entries.get(&ino).ok_or("node vanished")?;
if !e.node.is_file() {
return Err("not a regular file".into());
}
( (
e.node.name.clone(), e.node.name.clone(),
e.node.is_folder(),
e.node.modification_time, e.node.modification_time,
node_size(&e.node), node_size(&e.node),
) )
}; };
if is_folder {
// Record the recursive pin first so every descendant is eviction-
// exempt before we start filling the cache with the subtree.
self.cache
.add_pin(&uid, rel, true)
.map_err(|e| format!("pin: {e}"))?;
let n = self.pin_subtree(ino)?;
return Ok(format!("{name} ({n} files)"));
}
let bytes = self let bytes = self
.rt .download_file_tracked(&uid, &name, size)
.block_on(self.client.download_file(&uid))
.map_err(|e| format!("download: {e}"))?; .map_err(|e| format!("download: {e}"))?;
self.cache self.cache
.store(&uid, mtime, size, &bytes) .store(&uid, mtime, size, &bytes)
.map_err(|e| format!("cache store: {e}"))?; .map_err(|e| format!("cache store: {e}"))?;
self.cache self.cache
.add_pin(&uid, rel) .add_pin(&uid, rel, false)
.map_err(|e| format!("pin: {e}"))?; .map_err(|e| format!("pin: {e}"))?;
Ok(name) Ok(name)
} }
/// Download and cache every file in the subtree rooted at folder `ino`,
/// returning the count cached (already-fresh blobs counted, not re-fetched).
/// Walks the tree depth-first, enumerating each folder so a cold subtree is
/// fully discovered; the lock is dropped before each network download.
fn pin_subtree(&self, ino: u64) -> Result<usize, String> {
let mut files: Vec<(NodeUid, String, i64, u64)> = Vec::new();
let mut stack = vec![ino];
while let Some(dir) = stack.pop() {
self.ensure_children(dir)
.map_err(|e| format!("enumerate: {e:?}"))?;
let st = self.state.lock().unwrap();
if let Some(kids) = st.children.get(&dir) {
for &k in kids {
if let Some(e) = st.entries.get(&k) {
if e.node.is_folder() {
stack.push(k);
} else {
files.push((
e.uid.clone(),
e.node.name.clone(),
e.node.modification_time,
node_size(&e.node),
));
}
}
}
}
}
let mut count = 0;
for (uid, name, mtime, size) in files {
if self.cache.is_cached(&uid, mtime, size) {
count += 1;
continue;
}
match self.download_file_tracked(&uid, &name, size) {
Ok(bytes) => {
if self.cache.store(&uid, mtime, size, &bytes).is_ok() {
count += 1;
}
}
Err(e) => warn!(%uid, error = %e, "pin subtree: download failed"),
}
}
Ok(count)
}
/// Fetch a thumbnail of `ttype` for the file at `ino`, served from the cache /// Fetch a thumbnail of `ttype` for the file at `ino`, served from the cache
/// when fresh and otherwise downloaded from the remote and cached. Returns /// when fresh and otherwise downloaded from the remote and cached. Returns
/// `Ok(None)` when the node is not a file or has no thumbnail of that type. /// `Ok(None)` when the node is not a file or has no thumbnail of that type.
@@ -615,21 +894,32 @@ impl Core {
Ok(bytes) Ok(bytes)
} }
/// Unpin the file at `rel`, evicting its cached content. /// Unpin the node at `rel`, evicting its cached content. For a folder, also
/// evicts every descendant's cached blob (the subtree is no longer kept).
fn unpin(&self, rel: &Path) -> Result<String, String> { fn unpin(&self, rel: &Path) -> Result<String, String> {
let (ino, uid) = self let (ino, uid) = self
.resolve_path(rel) .resolve_path(rel)
.map_err(|e| format!("resolve path: {e:?}"))?; .map_err(|e| format!("resolve path: {e:?}"))?;
let name = { let (name, is_folder) = {
let st = self.state.lock().unwrap(); let st = self.state.lock().unwrap();
st.entries st.entries
.get(&ino) .get(&ino)
.map(|e| e.node.name.clone()) .map(|e| (e.node.name.clone(), e.node.is_folder()))
.unwrap_or_default() .unwrap_or_default()
}; };
self.cache self.cache
.remove_pin(&uid) .remove_pin(&uid)
.map_err(|e| format!("unpin: {e}"))?; .map_err(|e| format!("unpin: {e}"))?;
// A recursively-pinned folder's descendants were eviction-exempt; now
// that the pin is gone, reclaim their blobs eagerly instead of waiting
// for budget pressure. Descendants come from the DB node tree.
if is_folder && let Ok(uids) = self.db.descendants(&uid.to_string()) {
for s in uids {
if let Some(u) = parse_uid(&s) {
self.cache.evict(&u);
}
}
}
Ok(name) Ok(name)
} }
@@ -680,7 +970,33 @@ impl Core {
size, size,
modified, modified,
pinned: self.cache.is_pinned(&uid), pinned: self.cache.is_pinned(&uid),
cached: !is_dir && self.cache.is_cached(&uid, modified, size),
uid: uid.to_string(), uid: uid.to_string(),
// Listing entries live in the requested dir; the caller derives
// the path from its name. Left empty.
path: String::new(),
})
.collect())
}
/// Full-text search node names against the local SQLite index, mapping each
/// DB hit to the wire [`SearchHit`] (resolving live pin state from the cache,
/// which the DB doesn't track). Pure local lookup — never hits the network.
fn search(&self, query: &str, limit: usize) -> Result<Vec<SearchHit>, String> {
let hits = self
.db
.search(query, limit)
.map_err(|e| format!("search: {e}"))?;
Ok(hits
.into_iter()
.map(|h| SearchHit {
name: h.node.name.clone(),
path: h.path,
is_dir: h.node.is_folder(),
size: node_size(&h.node),
modified: h.node.modification_time,
pinned: self.cache.is_pinned(&h.node.uid),
uid: h.node.uid.to_string(),
}) })
.collect()) .collect())
} }
@@ -795,8 +1111,7 @@ impl Core {
return Ok(p); return Ok(p);
} }
let bytes = self let bytes = self
.rt .download_photo_tracked(&photos, uid, &node.name, size)
.block_on(photos.download_photo(uid))
.map_err(|e| format!("download photo: {e}"))?; .map_err(|e| format!("download photo: {e}"))?;
self.cache self.cache
.store(uid, mtime, size, &bytes) .store(uid, mtime, size, &bytes)
@@ -812,26 +1127,178 @@ impl Core {
let (ino, uid) = self let (ino, uid) = self
.resolve_path(rel) .resolve_path(rel)
.map_err(|e| format!("resolve path: {e:?}"))?; .map_err(|e| format!("resolve path: {e:?}"))?;
let (mtime, size) = { let (name, mtime, size) = {
let st = self.state.lock().unwrap(); let st = self.state.lock().unwrap();
let e = st.entries.get(&ino).ok_or("node vanished")?; let e = st.entries.get(&ino).ok_or("node vanished")?;
if !e.node.is_file() { if !e.node.is_file() {
return Err("not a regular file".into()); return Err("not a regular file".into());
} }
(e.node.modification_time, node_size(&e.node)) (
e.node.name.clone(),
e.node.modification_time,
node_size(&e.node),
)
}; };
if let Some(p) = self.cache.cached_content_path(&uid, mtime, size) { if let Some(p) = self.cache.cached_content_path(&uid, mtime, size) {
return Ok(p); return Ok(p);
} }
let bytes = self let bytes = self
.rt .download_file_tracked(&uid, &name, size)
.block_on(self.client.download_file(&uid))
.map_err(|e| format!("download: {e}"))?; .map_err(|e| format!("download: {e}"))?;
self.cache self.cache
.store(&uid, mtime, size, &bytes) .store(&uid, mtime, size, &bytes)
.map_err(|e| format!("cache store: {e}"))?; .map_err(|e| format!("cache store: {e}"))?;
Ok(self.cache.content_path(&uid)) Ok(self.cache.content_path(&uid))
} }
/// Drop the cached child listing of `rel`'s parent directory so the next
/// `ListDir` re-enumerates it from the server. No-op when the parent can't be
/// resolved (e.g. `rel` is the root). Resolves the parent without holding the
/// state lock, then invalidates under it.
fn invalidate_parent_listing(&self, rel: &Path) {
let parent = rel.parent().unwrap_or_else(|| Path::new(""));
if let Ok((pino, _)) = self.resolve_path(parent) {
self.state.lock().unwrap().invalidate_listing(pino);
}
}
/// Rename a file or folder to `new_name`. `rel` is mountpoint-relative.
/// Mirrors the FUSE `rename` write path: rename on the remote, forget the
/// node so it re-interns under its new name, and drop the parent listing so
/// the next `ListDir` re-enumerates.
fn rename(&self, rel: &Path, new_name: &str) -> Result<String, String> {
if new_name.is_empty() || new_name.contains('/') {
return Err(format!("invalid name: {new_name:?}"));
}
let (_ino, uid) = self
.resolve_path(rel)
.map_err(|e| format!("resolve path: {e:?}"))?;
self.rt
.block_on(self.client.rename_node(&uid, new_name, None))
.map_err(|e| format!("rename: {e}"))?;
self.state.lock().unwrap().forget(&uid);
self.invalidate_parent_listing(rel);
Ok(new_name.to_string())
}
/// Move a file or folder into the folder at `new_parent_rel`. Both paths are
/// mountpoint-relative. Forgets the node and invalidates both the source and
/// destination listings so each re-enumerates on next access.
fn move_to(&self, rel: &Path, new_parent_rel: &Path) -> Result<String, String> {
let (_ino, uid) = self
.resolve_path(rel)
.map_err(|e| format!("resolve path: {e:?}"))?;
let (pino, new_parent_uid) = self
.resolve_path(new_parent_rel)
.map_err(|e| format!("resolve new parent: {e:?}"))?;
self.rt
.block_on(self.client.move_node(&uid, &new_parent_uid))
.map_err(|e| format!("move: {e}"))?;
let name = self
.state
.lock()
.unwrap()
.forget(&uid)
.map(|(_, n)| n)
.unwrap_or_default();
self.invalidate_parent_listing(rel);
self.state.lock().unwrap().invalidate_listing(pino);
Ok(name)
}
/// Trash a file or folder. `rel` is mountpoint-relative. Forgets the node,
/// evicts any cached content, and invalidates the parent listing.
fn delete(&self, rel: &Path) -> Result<String, String> {
let (_ino, uid) = self
.resolve_path(rel)
.map_err(|e| format!("resolve path: {e:?}"))?;
self.rt
.block_on(self.client.trash_nodes(std::slice::from_ref(&uid)))
.map_err(|e| format!("trash: {e}"))?;
let name = self
.state
.lock()
.unwrap()
.forget(&uid)
.map(|(_, n)| n)
.unwrap_or_default();
self.cache.evict(&uid);
self.invalidate_parent_listing(rel);
Ok(name)
}
/// Create a folder named `name` under the mountpoint-relative `parent_rel`.
/// Interns the new node directly so it shows up without a re-enumeration.
fn create_folder(&self, parent_rel: &Path, name: &str) -> Result<String, String> {
if name.is_empty() || name.contains('/') {
return Err(format!("invalid name: {name:?}"));
}
let (pino, parent_uid) = self
.resolve_path(parent_rel)
.map_err(|e| format!("resolve path: {e:?}"))?;
self.ensure_children(pino)
.map_err(|e| format!("enumerate: {e:?}"))?;
let new_uid = self
.rt
.block_on(
self.client
.create_folder(&parent_uid, name, Some(now_secs())),
)
.map_err(|e| format!("create folder: {e}"))?;
let node = self
.fetch_node(&new_uid)
.map_err(|e| format!("fetch node: {e:?}"))?;
let mut st = self.state.lock().unwrap();
let ino = st.intern(pino, node);
if let Some(kids) = st.children.get_mut(&pino)
&& !kids.contains(&ino)
{
kids.push(ino);
}
Ok(name.to_string())
}
/// Upload a file named `name` with content `bytes` into the
/// mountpoint-relative `parent_rel` folder. Interns the new node directly.
fn upload(&self, parent_rel: &Path, name: &str, bytes: &[u8]) -> Result<String, String> {
if name.is_empty() || name.contains('/') {
return Err(format!("invalid name: {name:?}"));
}
let (pino, parent_uid) = self
.resolve_path(parent_rel)
.map_err(|e| format!("resolve path: {e:?}"))?;
self.ensure_children(pino)
.map_err(|e| format!("enumerate: {e:?}"))?;
let guard = self
.transfers
.begin(name, "", TransferDirection::Upload, bytes.len() as u64);
let reader = CountingReader::new(std::io::Cursor::new(bytes), &guard);
let new_uid = self
.rt
.block_on(self.client.upload_file_from(
&parent_uid,
name,
media_type_for(name),
reader,
bytes.len() as i64,
Vec::new(),
None,
false,
))
.map_err(|e| format!("upload: {e}"))?;
drop(guard);
let node = self
.fetch_node(&new_uid)
.map_err(|e| format!("fetch node: {e:?}"))?;
let mut st = self.state.lock().unwrap();
let ino = st.intern(pino, node);
if let Some(kids) = st.children.get_mut(&pino)
&& !kids.contains(&ino)
{
kids.push(ino);
}
Ok(name.to_string())
}
} }
/// Parse a `volume~link` uid display string back into a [`NodeUid`]. Front-ends /// Parse a `volume~link` uid display string back into a [`NodeUid`]. Front-ends
@@ -855,6 +1322,9 @@ impl ProtonFs {
fn new(core: Core, root: Node) -> Self { fn new(core: Core, root: Node) -> Self {
{ {
let mut st = core.state.lock().unwrap(); let mut st = core.state.lock().unwrap();
if let Err(e) = st.db.upsert_node(&root) {
warn!(uid = %root.uid, error = %e, "db upsert root failed");
}
st.by_uid.insert(root.uid.clone(), ROOT_INO); st.by_uid.insert(root.uid.clone(), ROOT_INO);
st.entries.insert( st.entries.insert(
ROOT_INO, ROOT_INO,
@@ -1710,7 +2180,7 @@ fn apply_event(
// Known node changed: drop its cached attrs/data (and listing if // Known node changed: drop its cached attrs/data (and listing if
// it is a directory) so the next access re-fetches. Its content // it is a directory) so the next access re-fetches. Its content
// blob may now be stale, so evict it too. // blob may now be stale, so evict it too.
st.children.remove(&ino); st.invalidate_listing(ino);
content.evict(node_uid); content.evict(node_uid);
let _ = notifier.inval_inode(INodeNo(ino), 0, 0); let _ = notifier.inval_inode(INodeNo(ino), 0, 0);
} }
@@ -1718,8 +2188,8 @@ fn apply_event(
// drop it so the new child is picked up on the next readdir. // drop it so the new child is picked up on the next readdir.
if let Some(parent_uid) = parent_node_uid if let Some(parent_uid) = parent_node_uid
&& let Some(&parent) = st.by_uid.get(parent_uid) && let Some(&parent) = st.by_uid.get(parent_uid)
&& st.children.remove(&parent).is_some()
{ {
st.invalidate_listing(parent);
let _ = notifier.inval_inode(INodeNo(parent), 0, 0); let _ = notifier.inval_inode(INodeNo(parent), 0, 0);
} }
} }
@@ -1746,8 +2216,8 @@ fn apply_event(
warn!("event continuity lost; dropping all cached listings, resyncing lazily"); warn!("event continuity lost; dropping all cached listings, resyncing lazily");
let mut st = state.lock().unwrap(); let mut st = state.lock().unwrap();
let dirs: Vec<u64> = st.children.keys().copied().collect(); let dirs: Vec<u64> = st.children.keys().copied().collect();
st.children.clear(); for &ino in &dirs {
for ino in dirs { st.invalidate_listing(ino);
let _ = notifier.inval_inode(INodeNo(ino), 0, 0); let _ = notifier.inval_inode(INodeNo(ino), 0, 0);
} }
} }
@@ -1758,20 +2228,41 @@ fn apply_event(
} }
/// Poll the remote event cursor forever, applying each batch to the shared /// Poll the remote event cursor forever, applying each batch to the shared
/// state. Seeds the cursor from the current server head so we only react to /// state. Resumes from the cursor persisted in the DB so changes made while
/// changes made after mount. Runs as a Tokio task; returns only on fatal error. /// unmounted are applied; only a first-ever mount seeds from the server head.
/// The cursor is persisted after every batch. Runs as a Tokio task; returns
/// only on fatal error.
async fn run_event_sync( async fn run_event_sync(
client: ProtonDriveClient, client: ProtonDriveClient,
scope: DriveEventScopeId, scope: DriveEventScopeId,
state: Arc<Mutex<State>>, state: Arc<Mutex<State>>,
content: Arc<ContentCache>, content: Arc<ContentCache>,
db: Arc<Db>,
notifier: Notifier, notifier: Notifier,
) { ) {
// Seed: a `None` cursor yields a single `CursorAdvanced` at the server head. let mut cursor: Option<DriveEventId> = match db.get_event_cursor() {
let mut cursor: Option<DriveEventId> = match client.enumerate_events(&scope, None).await { // Resume: pick up exactly where the last run left off.
Ok(events) => events.last().map(|e| e.id().clone()), Ok(Some(saved)) => Some(DriveEventId::from(saved)),
// First mount: a `None` cursor yields a single `CursorAdvanced` at the
// server head; persist it so the next restart resumes instead of
// reseeding (which would skip everything that changed offline).
Ok(None) => match client.enumerate_events(&scope, None).await {
Ok(events) => {
let head = events.last().map(|e| e.id().clone());
if let Some(c) = &head
&& let Err(e) = db.set_event_cursor(c.as_str())
{
warn!(error = %e, "persist seed cursor failed");
}
head
}
Err(e) => {
error!(error = %e, "failed to seed event cursor; live sync disabled");
return;
}
},
Err(e) => { Err(e) => {
error!(error = %e, "failed to seed event cursor; live sync disabled"); error!(error = %e, "read persisted cursor failed; live sync disabled");
return; return;
} }
}; };
@@ -1794,6 +2285,11 @@ async fn run_event_sync(
apply_event(&state, &content, &notifier, event); apply_event(&state, &content, &notifier, event);
} }
cursor = events.last().map(|e| e.id().clone()); cursor = events.last().map(|e| e.id().clone());
if let Some(c) = &cursor
&& let Err(e) = db.set_event_cursor(c.as_str())
{
warn!(error = %e, "persist event cursor failed");
}
} }
} }
@@ -1889,6 +2385,44 @@ fn handle_control_conn(core: &Core, username: &str, mountpoint: &Path, stream: U
message: format!("bad uid: {uid}"), message: format!("bad uid: {uid}"),
}, },
}, },
Ok(CtlRequest::UploadPhoto {
name,
media_type,
bytes,
capture_time,
}) => {
let photos = core.photos();
let metadata = proton_drive_rs::PhotoUploadMetadata {
capture_time,
..Default::default()
};
let guard = core.transfers.begin(
name.clone(),
"",
TransferDirection::Upload,
bytes.len() as u64,
);
let reader = CountingReader::new(std::io::Cursor::new(&bytes), &guard);
match core.rt.block_on(photos.upload_photo_from(
&name,
&media_type,
reader,
bytes.len() as i64,
Vec::new(),
metadata,
false,
)) {
Ok(uid) => {
*core.timeline.lock().unwrap() = None;
CtlResponse::Ok {
message: format!("uploaded photo with uid {uid}"),
}
}
Err(e) => CtlResponse::Error {
message: format!("upload photo failed: {e}"),
},
}
}
Ok(CtlRequest::OpenFile { path }) => match rel_to_mount(mountpoint, &path) { Ok(CtlRequest::OpenFile { path }) => match rel_to_mount(mountpoint, &path) {
Ok(rel) => match core.open_file(&rel) { Ok(rel) => match core.open_file(&rel) {
Ok(p) => CtlResponse::FilePath { Ok(p) => CtlResponse::FilePath {
@@ -1898,6 +2432,96 @@ fn handle_control_conn(core: &Core, username: &str, mountpoint: &Path, stream: U
}, },
Err(e) => CtlResponse::Error { message: e }, Err(e) => CtlResponse::Error { message: e },
}, },
Ok(CtlRequest::Search { query, limit }) => match core.search(&query, limit) {
Ok(hits) => CtlResponse::SearchResults { hits },
Err(e) => CtlResponse::Error { message: e },
},
Ok(CtlRequest::Rename { path, new_name }) => match rel_to_mount(mountpoint, &path) {
Ok(rel) => match core.rename(&rel, &new_name) {
Ok(name) => CtlResponse::Ok {
message: format!("renamed to {name}"),
},
Err(e) => CtlResponse::Error { message: e },
},
Err(e) => CtlResponse::Error { message: e },
},
Ok(CtlRequest::Move { path, new_parent }) => {
match (
rel_to_mount(mountpoint, &path),
rel_to_mount(mountpoint, &new_parent),
) {
(Ok(rel), Ok(parent_rel)) => match core.move_to(&rel, &parent_rel) {
Ok(name) => CtlResponse::Ok {
message: format!("moved {name}"),
},
Err(e) => CtlResponse::Error { message: e },
},
(Err(e), _) | (_, Err(e)) => CtlResponse::Error { message: e },
}
}
Ok(CtlRequest::Delete { path }) => match rel_to_mount(mountpoint, &path) {
Ok(rel) => match core.delete(&rel) {
Ok(name) => CtlResponse::Ok {
message: format!("trashed {name}"),
},
Err(e) => CtlResponse::Error { message: e },
},
Err(e) => CtlResponse::Error { message: e },
},
Ok(CtlRequest::CreateFolder { parent, name }) => match rel_to_mount(mountpoint, &parent) {
Ok(parent_rel) => match core.create_folder(&parent_rel, &name) {
Ok(name) => CtlResponse::Ok {
message: format!("created folder {name}"),
},
Err(e) => CtlResponse::Error { message: e },
},
Err(e) => CtlResponse::Error { message: e },
},
Ok(CtlRequest::UploadFile {
parent,
name,
bytes,
}) => match rel_to_mount(mountpoint, &parent) {
Ok(parent_rel) => match core.upload(&parent_rel, &name, &bytes) {
Ok(name) => CtlResponse::Ok {
message: format!("uploaded {name}"),
},
Err(e) => CtlResponse::Error { message: e },
},
Err(e) => CtlResponse::Error { message: e },
},
Ok(CtlRequest::PurgeCache) => {
let freed = core.cache.clear_unpinned();
CtlResponse::Ok {
message: format!(
"purged {:.1} MiB of unpinned cache",
freed as f64 / 1_048_576.0
),
}
}
Ok(CtlRequest::GetQueueStatus) => CtlResponse::Transfers {
items: core.transfers.snapshot(),
},
Ok(CtlRequest::SetCacheBudget { bytes }) => {
core.cache.set_budget(bytes);
// Persist so the next mount keeps the new cap. Best-effort: a config
// write failure is reported but the live cap is already applied.
match AppDirs::new().map(|dirs| {
let mut cfg = dirs.load_config();
cfg.cache_budget = Some(bytes);
dirs.save_config(&cfg)
}) {
Ok(Ok(())) => CtlResponse::Ok {
message: format!("cache budget set to {bytes} bytes"),
},
Ok(Err(e)) => CtlResponse::Error {
message: format!("budget applied but config write failed: {e}"),
},
Err(e) => CtlResponse::Error {
message: format!("budget applied but config unavailable: {e}"),
},
}
}
Err(e) => CtlResponse::Error { Err(e) => CtlResponse::Error {
message: format!("bad request: {e}"), message: format!("bad request: {e}"),
}, },
@@ -1981,6 +2605,7 @@ pub fn mount(
mountpoint: &Path, mountpoint: &Path,
cache: ContentCache, cache: ContentCache,
control_socket: &Path, control_socket: &Path,
db: Arc<Db>,
username: String, username: String,
) -> std::io::Result<MountOutcome> { ) -> std::io::Result<MountOutcome> {
let root = rt let root = rt
@@ -1998,9 +2623,12 @@ pub fn mount(
next_ino: 2, next_ino: 2,
handles: HashMap::new(), handles: HashMap::new(),
next_fh: 1, next_fh: 1,
db: db.clone(),
})), })),
cache: Arc::new(cache), cache: Arc::new(cache),
db,
timeline: Arc::new(Mutex::new(None)), timeline: Arc::new(Mutex::new(None)),
transfers: TransferRegistry::new(),
}; };
let mut config = Config::default(); let mut config = Config::default();
@@ -2029,6 +2657,9 @@ pub fn mount(
} }
let fs = ProtonFs::new(core.clone(), root); let fs = ProtonFs::new(core.clone(), root);
// Warm the in-memory maps from the DB so a cold start serves previously
// seen metadata without re-hitting the API (plan.md P1).
core.hydrate();
// Build the session explicitly (not `mount2`) so we can grab a `Notifier` // Build the session explicitly (not `mount2`) so we can grab a `Notifier`
// for the event task. `spawn` runs the session loop on its own background // for the event task. `spawn` runs the session loop on its own background
@@ -2036,7 +2667,7 @@ pub fn mount(
let bg = Session::new(fs, mountpoint, &config)?.spawn()?; let bg = Session::new(fs, mountpoint, &config)?.spawn()?;
let notifier = bg.notifier(); let notifier = bg.notifier();
rt.spawn(run_event_sync( rt.spawn(run_event_sync(
client, scope, core.state, core.cache, notifier, client, scope, core.state, core.cache, core.db, notifier,
)); ));
// Stop signals (SIGTERM from `systemctl --user stop`, SIGINT from Ctrl-C) // Stop signals (SIGTERM from `systemctl --user stop`, SIGINT from Ctrl-C)
+169
View File
@@ -0,0 +1,169 @@
//! In-memory registry of in-flight transfers, behind [`Request::GetQueueStatus`].
//!
//! The daemon downloads and uploads whole files through the SDK's streaming
//! [`Read`]/[`Write`] variants. Wrapping that reader/writer in a
//! [`CountingReader`]/[`CountingWriter`] lets each block of bytes tick a
//! per-transfer counter without the SDK knowing anything about progress. A
//! [`TransferGuard`] registers a transfer on creation and deregisters it on
//! drop, so the registry always reflects exactly what is in flight — even if a
//! transfer fails partway and unwinds.
//!
//! [`Request::GetQueueStatus`]: pdfs_core::control::Request::GetQueueStatus
use std::collections::HashMap;
use std::io::{Read, Write};
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, Mutex};
use std::time::Instant;
use pdfs_core::control::{TransferDirection, TransferItem};
/// One registered transfer. `done` is bumped from the I/O wrapper without the
/// registry lock; the rest is immutable for the transfer's lifetime.
struct Entry {
name: String,
uid: String,
direction: TransferDirection,
total: u64,
done: AtomicU64,
started: Instant,
}
/// The set of transfers currently in flight. Cloned `Arc`-style across the FUSE
/// session and the control-socket task (both share one registry via [`Core`]).
#[derive(Default)]
pub struct TransferRegistry {
inner: Mutex<HashMap<u64, Arc<Entry>>>,
next: AtomicU64,
}
impl TransferRegistry {
pub fn new() -> Arc<Self> {
Arc::new(Self::default())
}
/// Register a new transfer of `total` bytes (`0` = unknown), returning a
/// guard that tracks progress and deregisters the transfer when dropped.
pub fn begin(
self: &Arc<Self>,
name: impl Into<String>,
uid: impl Into<String>,
direction: TransferDirection,
total: u64,
) -> TransferGuard {
let id = self.next.fetch_add(1, Ordering::Relaxed);
let entry = Arc::new(Entry {
name: name.into(),
uid: uid.into(),
direction,
total,
done: AtomicU64::new(0),
started: Instant::now(),
});
self.inner.lock().unwrap().insert(id, entry.clone());
TransferGuard {
reg: self.clone(),
id,
entry,
}
}
/// Snapshot every in-flight transfer for [`Response::Transfers`]. Speed is the
/// running average since the transfer began — simple, and stable enough for a
/// progress widget without per-tick sampling state.
///
/// [`Response::Transfers`]: pdfs_core::control::Response::Transfers
pub fn snapshot(&self) -> Vec<TransferItem> {
let map = self.inner.lock().unwrap();
map.values()
.map(|e| {
let done = e.done.load(Ordering::Relaxed);
let secs = e.started.elapsed().as_secs_f64();
let speed = if secs > 0.0 {
(done as f64 / secs) as u64
} else {
0
};
TransferItem {
uid: e.uid.clone(),
name: e.name.clone(),
direction: e.direction,
bytes_completed: done,
bytes_total: e.total,
speed_bytes_sec: speed,
}
})
.collect()
}
}
/// Lifetime handle for a registered transfer: tick progress with [`add`], and
/// the transfer leaves the registry when this drops.
///
/// [`add`]: TransferGuard::add
pub struct TransferGuard {
reg: Arc<TransferRegistry>,
id: u64,
entry: Arc<Entry>,
}
impl TransferGuard {
/// Record `n` more bytes moved.
pub fn add(&self, n: u64) {
self.entry.done.fetch_add(n, Ordering::Relaxed);
}
}
impl Drop for TransferGuard {
fn drop(&mut self) {
self.reg.inner.lock().unwrap().remove(&self.id);
}
}
/// A [`Write`] that tallies bytes written to a [`TransferGuard`] (download path).
pub struct CountingWriter<'a, W> {
inner: W,
guard: &'a TransferGuard,
}
impl<'a, W: Write> CountingWriter<'a, W> {
pub fn new(inner: W, guard: &'a TransferGuard) -> Self {
Self { inner, guard }
}
pub fn into_inner(self) -> W {
self.inner
}
}
impl<W: Write> Write for CountingWriter<'_, W> {
fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
let n = self.inner.write(buf)?;
self.guard.add(n as u64);
Ok(n)
}
fn flush(&mut self) -> std::io::Result<()> {
self.inner.flush()
}
}
/// A [`Read`] that tallies bytes read through a [`TransferGuard`] (upload path).
pub struct CountingReader<'a, R> {
inner: R,
guard: &'a TransferGuard,
}
impl<'a, R: Read> CountingReader<'a, R> {
pub fn new(inner: R, guard: &'a TransferGuard) -> Self {
Self { inner, guard }
}
}
impl<R: Read> Read for CountingReader<'_, R> {
fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
let n = self.inner.read(buf)?;
self.guard.add(n as u64);
Ok(n)
}
}
+4
View File
@@ -14,6 +14,10 @@ path = "src/main.rs"
name = "pdfs-app" name = "pdfs-app"
path = "src/app.rs" path = "src/app.rs"
[[bin]]
name = "pdfs-prompt"
path = "src/prompt.rs"
[dependencies] [dependencies]
pdfs-core.workspace = true pdfs-core.workspace = true
tracing.workspace = true tracing.workspace = true
+1509 -78
View File
File diff suppressed because it is too large. Load diff
+80 -2
View File
@@ -18,7 +18,7 @@ use ksni::menu::StandardItem;
use ksni::{MenuItem, Tray, TrayService}; use ksni::{MenuItem, Tray, TrayService};
use pdfs_core::auth; use pdfs_core::auth;
use pdfs_core::config::AppDirs; use pdfs_core::config::AppDirs;
use pdfs_core::control::{Request, Response, send}; use pdfs_core::control::{Request, Response, TransferDirection, TransferItem, send};
use pdfs_core::service; use pdfs_core::service;
/// How often the tray re-polls the daemon to refresh its menu. /// How often the tray re-polls the daemon to refresh its menu.
@@ -33,6 +33,42 @@ struct DriveState {
mounted: bool, mounted: bool,
/// Mountpoint to act on (from the daemon when mounted, else the default). /// Mountpoint to act on (from the daemon when mounted, else the default).
mountpoint: PathBuf, mountpoint: PathBuf,
/// One-line sync summary of in-flight transfers, empty when idle. Shown as a
/// disabled menu line under the status line.
sync: String,
}
/// Summarise a transfer snapshot into one menu line, or empty when idle. A single
/// transfer names the file and its percentage; several collapse to counts so the
/// menu stays one line regardless of queue depth.
fn sync_line(items: &[TransferItem]) -> String {
match items {
[] => String::new(),
[t] => {
let arrow = match t.direction {
TransferDirection::Download => "↓",
TransferDirection::Upload => "↑",
};
if t.bytes_total == 0 {
format!("{arrow} {}…", t.name)
} else {
let pct = (t.bytes_completed * 100 / t.bytes_total.max(1)).min(100);
format!("{arrow} {} ({pct}%)", t.name)
}
}
_ => {
let down = items
.iter()
.filter(|t| t.direction == TransferDirection::Download)
.count();
let up = items.len() - down;
match (down, up) {
(d, 0) => format!("↓ {d} downloading"),
(0, u) => format!("↑ {u} uploading"),
(d, u) => format!("↓ {d} · ↑ {u}"),
}
}
}
} }
struct DriveTray { struct DriveTray {
@@ -49,12 +85,18 @@ fn poll_state(socket: &Path, default_mountpoint: &Path) -> DriveState {
line: format!("Mounted at {mountpoint} ({pinned} pinned)"), line: format!("Mounted at {mountpoint} ({pinned} pinned)"),
mounted: true, mounted: true,
mountpoint: PathBuf::from(mountpoint), mountpoint: PathBuf::from(mountpoint),
// Same daemon is up, so a cheap follow-up poll gives the sync line.
sync: match send(socket, &Request::GetQueueStatus) {
Ok(Response::Transfers { items }) => sync_line(&items),
_ => String::new(),
},
}, },
// Socket answered but with something unexpected — treat as up but odd. // Socket answered but with something unexpected — treat as up but odd.
Ok(_) => DriveState { Ok(_) => DriveState {
line: "Mount: unexpected daemon response".into(), line: "Mount: unexpected daemon response".into(),
mounted: true, mounted: true,
mountpoint: default_mountpoint.to_path_buf(), mountpoint: default_mountpoint.to_path_buf(),
sync: String::new(),
}, },
// No daemon: describe login state instead so the menu is still useful. // No daemon: describe login state instead so the menu is still useful.
Err(_) => { Err(_) => {
@@ -67,6 +109,7 @@ fn poll_state(socket: &Path, default_mountpoint: &Path) -> DriveState {
line, line,
mounted: false, mounted: false,
mountpoint: default_mountpoint.to_path_buf(), mountpoint: default_mountpoint.to_path_buf(),
sync: String::new(),
} }
} }
} }
@@ -78,6 +121,20 @@ fn open_folder(mountpoint: &Path) {
} }
} }
/// Launch (or, since it's `SingleMainWindow`, raise) the settings/management
/// window. The app binary lives next to the tray binary; fall back to the bare
/// name so a PATH install still works.
fn open_manager() {
let exe = std::env::current_exe()
.ok()
.and_then(|p| p.parent().map(|d| d.join("pdfs-app")))
.filter(|p| p.exists())
.unwrap_or_else(|| PathBuf::from("pdfs-app"));
if let Err(e) = Command::new(&exe).spawn() {
tracing::error!("failed to launch {}: {e}", exe.display());
}
}
impl Tray for DriveTray { impl Tray for DriveTray {
fn id(&self) -> String { fn id(&self) -> String {
"io.narl.proton-drive-linux".into() "io.narl.proton-drive-linux".into()
@@ -102,9 +159,30 @@ impl Tray for DriveTray {
..Default::default() ..Default::default()
} }
.into(), .into(),
MenuItem::Separator,
]; ];
// Live sync status, shown only while something is transferring.
if !self.state.sync.is_empty() {
items.push(
StandardItem {
label: self.state.sync.clone(),
enabled: false,
..Default::default()
}
.into(),
);
}
items.extend([
MenuItem::Separator,
StandardItem {
label: "Open Manager".into(),
activate: Box::new(|_: &mut Self| open_manager()),
..Default::default()
}
.into(),
]);
if self.state.mounted { if self.state.mounted {
items.push( items.push(
StandardItem { StandardItem {
+716
View File
@@ -0,0 +1,716 @@
//! `pdfs-prompt` — a fast, Google Drive-style launcher prompt for searching
//! and opening files or folders in Proton Drive.
//!
//! Designed to be bound to a system shortcut (e.g. in Hyprland) for quick HUD search.
use std::cell::{Cell, RefCell};
use std::path::{Path, PathBuf};
use std::process::Command;
use std::rc::Rc;
use std::time::Duration;
use adw::prelude::*;
use gtk4::glib;
use pdfs_core::config::AppDirs;
use pdfs_core::control::{Request, Response, SearchHit, send};
const APP_ID: &str = "io.narl.proton-drive-linux-prompt";
const PROTON_PURPLE: &str = "#6d4aff";
const SEARCH_DEBOUNCE: Duration = Duration::from_millis(150);
/// Run one blocking control-socket round-trip on a worker thread, returning a
/// channel that yields the [`Response`] once.
fn spawn_request(
socket: PathBuf,
req: Request,
) -> async_channel::Receiver<Result<Response, String>> {
let (tx, rx) = async_channel::bounded(1);
std::thread::spawn(move || {
let result = send(&socket, &req).map_err(|e| e.to_string());
let _ = tx.send_blocking(result);
});
rx
}
fn main() -> glib::ExitCode {
tracing_subscriber::fmt()
.with_env_filter(
tracing_subscriber::EnvFilter::try_from_default_env().unwrap_or_else(|_| "info".into()),
)
.init();
let app = adw::Application::builder().application_id(APP_ID).build();
app.connect_startup(|_| {
load_prompt_theme();
});
app.connect_activate(build_window);
app.run()
}
/// Override accent colors and style the launcher for a premium HUD look.
fn load_prompt_theme() {
let css = format!(
"@define-color accent_bg_color {PROTON_PURPLE};\n\
@define-color accent_color {PROTON_PURPLE};\n\
\n\
window.launcher-window {{\n\
background-color: transparent;\n\
}}\n\
\n\
.launcher-container {{\n\
background-color: @window_bg_color;\n\
border: 2px solid {PROTON_PURPLE};\n\
border-radius: 12px;\n\
padding: 12px;\n\
box-shadow: 0 10px 30px rgba(0, 0, 0, 0.5);\n\
}}\n\
\n\
.launcher-header-box {{\n\
padding: 6px 12px;\n\
background-color: alpha(@view_bg_color, 0.4);\n\
border-radius: 8px;\n\
border: 1px solid alpha({PROTON_PURPLE}, 0.2);\n\
}}\n\
\n\
.launcher-entry {{\n\
font-size: 1.15rem;\n\
background: transparent;\n\
border: none;\n\
box-shadow: none;\n\
}}\n\
\n\
.launcher-listbox {{\n\
background: transparent;\n\
}}\n\
\n\
.launcher-row {{\n\
padding: 10px 14px;\n\
border-radius: 6px;\n\
margin: 2px 0px;\n\
transition: background-color 0.15s ease, color 0.15s ease;\n\
}}\n\
\n\
.launcher-row:hover {{\n\
background-color: alpha({PROTON_PURPLE}, 0.1);\n\
}}\n\
\n\
.launcher-row:selected {{\n\
background-color: {PROTON_PURPLE};\n\
color: white;\n\
}}\n\
\n\
.launcher-row:selected .dim-label {{\n\
color: alpha(white, 0.7);\n\
}}\n\
\n\
.launcher-footer {{\n\
padding: 6px 12px 0px 12px;\n\
font-size: 0.85rem;\n\
border-top: 1px solid alpha({PROTON_PURPLE}, 0.1);\n\
}}\n\
\n\
.dim-label {{\n\
color: @view_fg_color;\n\
opacity: 0.6;\n\
font-size: 0.85rem;\n\
}}\n\
\n\
.error-box {{\n\
padding: 24px;\n\
}}\n\
\n\
.error-title {{\n\
font-size: 1.25rem;\n\
font-weight: bold;\n\
margin-bottom: 8px;\n\
}}\n"
);
let provider = gtk4::CssProvider::new();
provider.load_from_string(&css);
if let Some(display) = gtk4::gdk::Display::default() {
gtk4::style_context_add_provider_for_display(
&display,
&provider,
gtk4::STYLE_PROVIDER_PRIORITY_APPLICATION,
);
}
}
fn build_window(app: &adw::Application) {
let dirs = match AppDirs::new() {
Ok(d) => d,
Err(e) => {
tracing::error!("cannot resolve app dirs: {e}");
return;
}
};
let socket = dirs.control_socket();
let default_mountpoint = dirs.default_mountpoint();
let window = adw::ApplicationWindow::builder()
.application(app)
.title("Proton Drive Launcher")
.default_width(650)
.default_height(450)
.decorated(false)
.build();
window.add_css_class("launcher-window");
// Prefer dark mode for high-tech HUD look
let style_manager = adw::StyleManager::default();
style_manager.set_color_scheme(adw::ColorScheme::PreferDark);
let container = gtk4::Box::new(gtk4::Orientation::Vertical, 8);
container.add_css_class("launcher-container");
window.set_content(Some(&container));
// Header box: search icon, entry field, and loading spinner
let header_box = gtk4::Box::new(gtk4::Orientation::Horizontal, 8);
header_box.add_css_class("launcher-header-box");
let search_icon = gtk4::Image::from_icon_name("system-search-symbolic");
header_box.append(&search_icon);
let entry = gtk4::Entry::builder()
.placeholder_text("Search Proton Drive...")
.hexpand(true)
.build();
entry.add_css_class("launcher-entry");
header_box.append(&entry);
let spinner = gtk4::Spinner::new();
spinner.set_visible(false);
header_box.append(&spinner);
container.append(&header_box);
let separator = gtk4::Separator::new(gtk4::Orientation::Horizontal);
container.append(&separator);
// View stack to transition between results list and warning page
let stack = adw::ViewStack::new();
container.append(&stack);
// Page 1: Scrolled window and listbox for results
let list_box = gtk4::ListBox::new();
list_box.add_css_class("launcher-listbox");
let scrolled_window = gtk4::ScrolledWindow::builder()
.hscrollbar_policy(gtk4::PolicyType::Never)
.vscrollbar_policy(gtk4::PolicyType::Automatic)
.vexpand(true)
.child(&list_box)
.build();
stack.add_named(&scrolled_window, Some("search"));
// Page 2: Error screen if daemon is offline
let error_box = gtk4::Box::new(gtk4::Orientation::Vertical, 12);
error_box.add_css_class("error-box");
error_box.set_valign(gtk4::Align::Center);
error_box.set_halign(gtk4::Align::Center);
let error_icon = gtk4::Image::from_icon_name("dialog-warning-symbolic");
error_icon.set_pixel_size(48);
error_icon.add_css_class("brand-icon");
error_box.append(&error_icon);
let error_title = gtk4::Label::builder()
.label("Daemon Offline")
.justify(gtk4::Justification::Center)
.build();
error_title.add_css_class("error-title");
error_box.append(&error_title);
let error_desc = gtk4::Label::builder()
.label("Proton Drive mount daemon is not running.\nStart the daemon or run `pdfs mount` to connect.")
.justify(gtk4::Justification::Center)
.build();
error_desc.add_css_class("dim-label");
error_box.append(&error_desc);
let retry_button = gtk4::Button::builder()
.label("Retry Connection")
.halign(gtk4::Align::Center)
.build();
retry_button.add_css_class("suggested-action");
retry_button.add_css_class("pill");
error_box.append(&retry_button);
stack.add_named(&error_box, Some("error"));
// Footer box: status label
let footer_box = gtk4::Box::new(gtk4::Orientation::Horizontal, 8);
footer_box.add_css_class("launcher-footer");
let status_label = gtk4::Label::builder()
.label("Initialising...")
.halign(gtk4::Align::Start)
.ellipsize(gtk4::pango::EllipsizeMode::End)
.hexpand(true)
.build();
status_label.add_css_class("dim-label");
footer_box.append(&status_label);
container.append(&footer_box);
// Shared state variables
let hits = Rc::new(RefCell::new(Vec::<SearchHit>::new()));
let mountpoint = Rc::new(RefCell::new(default_mountpoint.display().to_string()));
let is_opening = Rc::new(Cell::new(false));
// Handle retry on connection failure
let stack_clone = stack.clone();
let socket_clone = socket.clone();
let mountpoint_clone = mountpoint.clone();
let status_label_clone = status_label.clone();
let spinner_clone = spinner.clone();
let list_box_clone = list_box.clone();
let hits_clone = hits.clone();
let entry_clone = entry.clone();
let check_connection = move || {
status_label_clone.set_text("Connecting to daemon...");
match send(&socket_clone, &Request::Status) {
Ok(Response::Status { mountpoint: mp, .. }) => {
*mountpoint_clone.borrow_mut() = mp;
stack_clone.set_visible_child_name("search");
entry_clone.set_sensitive(true);
load_initial_pins(
&list_box_clone,
&hits_clone,
&socket_clone,
&status_label_clone,
&spinner_clone,
);
true
}
_ => {
stack_clone.set_visible_child_name("error");
entry_clone.set_sensitive(false);
status_label_clone.set_text("Daemon offline.");
false
}
}
};
let check_conn = check_connection.clone();
retry_button.connect_clicked(move |_| {
check_conn();
});
window.connect_is_active_notify(move |w| {
if !w.is_active() {
w.close();
}
});
// Auto-focus search entry on window show
let entry_focus = entry.clone();
window.connect_map(move |_| {
entry_focus.grab_focus();
});
// Event key controller on entry field for keyboard shortcuts
let key_controller = gtk4::EventControllerKey::new();
let list_box_keys = list_box.clone();
let window_keys = window.clone();
let entry_keys = entry.clone();
let hits_keys = hits.clone();
let is_opening_keys = is_opening.clone();
key_controller.connect_key_pressed(move |_, key, _keycode, _state| {
if is_opening_keys.get() {
return glib::Propagation::Stop;
}
match key {
gtk4::gdk::Key::Escape => {
window_keys.close();
glib::Propagation::Stop
}
gtk4::gdk::Key::Down => {
let count = hits_keys.borrow().len();
if count > 0 {
let selected = list_box_keys.selected_row();
let next_idx = if let Some(row) = selected {
(row.index() + 1) % count as i32
} else {
0
};
if let Some(row) = list_box_keys.row_at_index(next_idx) {
list_box_keys.select_row(Some(&row));
row.grab_focus();
entry_keys.grab_focus();
}
}
glib::Propagation::Stop
}
gtk4::gdk::Key::Up => {
let count = hits_keys.borrow().len();
if count > 0 {
let selected = list_box_keys.selected_row();
let next_idx = if let Some(row) = selected {
(row.index() - 1 + count as i32) % count as i32
} else {
(count - 1) as i32
};
if let Some(row) = list_box_keys.row_at_index(next_idx) {
list_box_keys.select_row(Some(&row));
row.grab_focus();
entry_keys.grab_focus();
}
}
glib::Propagation::Stop
}
_ => glib::Propagation::Proceed,
}
});
entry.add_controller(key_controller);
// Entry activation (pressing Return/Enter)
let list_box_activate = list_box.clone();
let hits_activate = hits.clone();
let window_activate = window.clone();
let socket_activate = socket.clone();
let mountpoint_activate = mountpoint.clone();
let status_label_activate = status_label.clone();
let spinner_activate = spinner.clone();
let is_opening_activate = is_opening.clone();
entry.connect_activate(move |_| {
if is_opening_activate.get() {
return;
}
if let Some(row) = list_box_activate.selected_row() {
let idx = row.index() as usize;
let hits_borrow = hits_activate.borrow();
if idx < hits_borrow.len() {
let hit = &hits_borrow[idx];
open_hit(
&window_activate,
hit,
&socket_activate,
&mountpoint_activate,
&status_label_activate,
&spinner_activate,
is_opening_activate.clone(),
);
}
}
});
// List box row activation (mouse clicks / touch taps)
let hits_act = hits.clone();
let window_act = window.clone();
let socket_act = socket.clone();
let mountpoint_act = mountpoint.clone();
let status_label_act = status_label.clone();
let spinner_act = spinner.clone();
let is_opening_act = is_opening.clone();
list_box.connect_row_activated(move |_, row| {
if is_opening_act.get() {
return;
}
let idx = row.index() as usize;
let hits_borrow = hits_act.borrow();
if idx < hits_borrow.len() {
let hit = &hits_borrow[idx];
open_hit(
&window_act,
hit,
&socket_act,
&mountpoint_act,
&status_label_act,
&spinner_act,
is_opening_act.clone(),
);
}
});
// Debounced search text entry listener
let search_source = Rc::new(RefCell::new(None::<glib::SourceId>));
let search_source_changed = search_source.clone();
let list_box_search = list_box.clone();
let hits_search = hits.clone();
let socket_search = socket.clone();
let status_label_search = status_label.clone();
let spinner_search = spinner.clone();
entry.connect_changed(move |ent| {
if let Some(src) = search_source_changed.borrow_mut().take() {
src.remove();
}
let ent_clone = ent.clone();
let list_box_clone = list_box_search.clone();
let hits_clone = hits_search.clone();
let socket_clone = socket_search.clone();
let status_label_clone = status_label_search.clone();
let spinner_clone = spinner_search.clone();
let search_source_cb = search_source_changed.clone();
let src = glib::timeout_add_local_once(SEARCH_DEBOUNCE, move || {
search_source_cb.borrow_mut().take();
let query = ent_clone.text().trim().to_string();
if query.is_empty() {
load_initial_pins(
&list_box_clone,
&hits_clone,
&socket_clone,
&status_label_clone,
&spinner_clone,
);
} else {
run_prompt_search(
&list_box_clone,
&hits_clone,
&socket_clone,
&query,
&status_label_clone,
&spinner_clone,
);
}
});
*search_source_changed.borrow_mut() = Some(src);
});
// Perform the initial connection and load check
check_connection();
window.present();
}
/// Retrieve pinned files from the daemon and render them in the results list.
fn load_initial_pins(
list_box: &gtk4::ListBox,
hits: &Rc<RefCell<Vec<SearchHit>>>,
socket: &Path,
status_label: &gtk4::Label,
spinner: &gtk4::Spinner,
) {
status_label.set_text("Loading pinned files...");
spinner.set_visible(true);
spinner.start();
let rx = spawn_request(socket.to_path_buf(), Request::ListPins);
let list_box = list_box.clone();
let hits = hits.clone();
let status_label = status_label.clone();
let spinner = spinner.clone();
glib::spawn_future_local(async move {
let res = rx.recv().await;
spinner.stop();
spinner.set_visible(false);
match res {
Ok(Ok(Response::Pins { pins })) => {
let mapped_hits: Vec<SearchHit> = pins
.into_iter()
.map(|p| {
let path_obj = Path::new(&p.path);
let name = path_obj
.file_name()
.and_then(|s| s.to_str())
.unwrap_or(&p.path)
.to_string();
SearchHit {
name,
path: p.path,
is_dir: p.recursive,
size: 0,
modified: 0,
pinned: true,
uid: p.uid,
}
})
.collect();
repaint_prompt_results(&list_box, &hits, &mapped_hits);
if mapped_hits.is_empty() {
status_label.set_text("No pinned files. Type to search.");
} else {
status_label.set_text("Showing pinned files. Type to search.");
}
}
_ => {
status_label.set_text("Failed to load pinned files. Type to search.");
}
}
});
}
/// Run search on the daemon and update the results list.
fn run_prompt_search(
list_box: &gtk4::ListBox,
hits: &Rc<RefCell<Vec<SearchHit>>>,
socket: &Path,
query: &str,
status_label: &gtk4::Label,
spinner: &gtk4::Spinner,
) {
status_label.set_text("Searching...");
spinner.set_visible(true);
spinner.start();
let rx = spawn_request(
socket.to_path_buf(),
Request::Search {
query: query.to_string(),
limit: 30,
},
);
let list_box = list_box.clone();
let hits = hits.clone();
let status_label = status_label.clone();
let spinner = spinner.clone();
glib::spawn_future_local(async move {
let res = rx.recv().await;
spinner.stop();
spinner.set_visible(false);
match res {
Ok(Ok(Response::SearchResults { hits: search_hits })) => {
repaint_prompt_results(&list_box, &hits, &search_hits);
if search_hits.is_empty() {
status_label.set_text("No matches found.");
} else {
status_label.set_text(&format!("Found {} matches.", search_hits.len()));
}
}
Ok(Ok(Response::Error { message })) => {
status_label.set_text(&format!("Search error: {message}"));
}
_ => {
status_label.set_text("Search request failed.");
}
}
});
}
/// Render the provided search hits into the ListBox.
fn repaint_prompt_results(
list_box: &gtk4::ListBox,
hits: &Rc<RefCell<Vec<SearchHit>>>,
new_hits: &[SearchHit],
) {
// Clear all rows
while let Some(child) = list_box.first_child() {
list_box.remove(&child);
}
*hits.borrow_mut() = new_hits.to_vec();
for hit in new_hits {
let row = gtk4::ListBoxRow::new();
row.add_css_class("launcher-row");
let row_box = gtk4::Box::new(gtk4::Orientation::Horizontal, 12);
let icon_name = if hit.is_dir {
"folder-symbolic"
} else {
"document-symbolic"
};
let icon = gtk4::Image::from_icon_name(icon_name);
row_box.append(&icon);
let text_box = gtk4::Box::new(gtk4::Orientation::Vertical, 2);
text_box.set_hexpand(true);
let name_label = gtk4::Label::builder()
.label(&hit.name)
.halign(gtk4::Align::Start)
.ellipsize(gtk4::pango::EllipsizeMode::End)
.build();
let path_label = gtk4::Label::builder()
.label(&hit.path)
.halign(gtk4::Align::Start)
.ellipsize(gtk4::pango::EllipsizeMode::End)
.build();
path_label.add_css_class("dim-label");
text_box.append(&name_label);
text_box.append(&path_label);
row_box.append(&text_box);
row.set_child(Some(&row_box));
list_box.append(&row);
}
// Automatically select the first item
if !new_hits.is_empty()
&& let Some(row) = list_box.row_at_index(0) {
list_box.select_row(Some(&row));
}
}
/// Open the selected search hit. Directories open FUSE directly, files trigger cache download.
fn open_hit(
window: &adw::ApplicationWindow,
hit: &SearchHit,
socket: &Path,
mountpoint: &Rc<RefCell<String>>,
status_label: &gtk4::Label,
spinner: &gtk4::Spinner,
is_opening: Rc<Cell<bool>>,
) {
if hit.is_dir {
let mp = mountpoint.borrow();
let full_path = Path::new(&*mp).join(&hit.path);
tracing::info!("Opening directory: {}", full_path.display());
if let Err(e) = Command::new("xdg-open").arg(&full_path).spawn() {
tracing::error!("Failed to xdg-open directory: {e}");
}
window.close();
} else {
is_opening.set(true);
status_label.set_text(&format!("Downloading {}...", hit.name));
spinner.set_visible(true);
spinner.start();
let rx = spawn_request(
socket.to_path_buf(),
Request::OpenFile {
path: hit.path.clone(),
},
);
let window_clone = window.clone();
let status_label_clone = status_label.clone();
let spinner_clone = spinner.clone();
let is_opening_clone = is_opening.clone();
glib::spawn_future_local(async move {
let res = rx.recv().await;
spinner_clone.stop();
spinner_clone.set_visible(false);
is_opening_clone.set(false);
match res {
Ok(Ok(Response::FilePath { path })) => {
tracing::info!("Opening file: {}", path);
if let Err(e) = Command::new("xdg-open").arg(&path).spawn() {
tracing::error!("Failed to xdg-open file: {e}");
}
window_clone.close();
}
Ok(Ok(Response::Error { message })) => {
status_label_clone.set_text(&format!("Error: {message}"));
}
_ => {
status_label_clone.set_text("Error: Failed to communicate with daemon.");
}
}
});
}
}
+4 -2
View File
@@ -8,7 +8,7 @@
# Build + install: cd packaging && makepkg -fi # Build + install: cd packaging && makepkg -fi
pkgname=proton-drive-linux pkgname=proton-drive-linux
pkgver=0.1.2 pkgver=0.2.1
pkgrel=1 pkgrel=1
pkgdesc="Proton Drive client for Linux: FUSE files-on-demand mount, CLI, GTK4 app + tray" pkgdesc="Proton Drive client for Linux: FUSE files-on-demand mount, CLI, GTK4 app + tray"
arch=('x86_64') arch=('x86_64')
@@ -43,7 +43,8 @@ build() {
cargo build --release --frozen \ cargo build --release --frozen \
--bin pdfs \ --bin pdfs \
--bin pdfs-tray \ --bin pdfs-tray \
--bin pdfs-app --bin pdfs-app \
--bin pdfs-prompt
} }
package() { package() {
@@ -53,6 +54,7 @@ package() {
install -Dm755 "$rel/pdfs" "$pkgdir/usr/bin/pdfs" install -Dm755 "$rel/pdfs" "$pkgdir/usr/bin/pdfs"
install -Dm755 "$rel/pdfs-tray" "$pkgdir/usr/bin/pdfs-tray" install -Dm755 "$rel/pdfs-tray" "$pkgdir/usr/bin/pdfs-tray"
install -Dm755 "$rel/pdfs-app" "$pkgdir/usr/bin/pdfs-app" install -Dm755 "$rel/pdfs-app" "$pkgdir/usr/bin/pdfs-app"
install -Dm755 "$rel/pdfs-prompt" "$pkgdir/usr/bin/pdfs-prompt"
install -Dm644 "packaging/io.narl.proton-drive-linux.desktop" \ install -Dm644 "packaging/io.narl.proton-drive-linux.desktop" \
"$pkgdir/usr/share/applications/io.narl.proton-drive-linux.desktop" "$pkgdir/usr/share/applications/io.narl.proton-drive-linux.desktop"