From 35bc957cd58fba16c4127af8f8f7422cbcb7f4d7 Mon Sep 17 00:00:00 2001 From: Nils Pukropp Date: Tue, 30 Jun 2026 23:52:41 +0200 Subject: [PATCH] transfer progress api --- Cargo.lock | 8 +- Cargo.toml | 5 +- crates/pdfs-cli/src/main.rs | 31 ++++++ crates/pdfs-core/src/control.rs | 30 ++++++ crates/pdfs-fuse/src/lib.rs | 122 +++++++++++++++++---- crates/pdfs-fuse/src/transfers.rs | 169 ++++++++++++++++++++++++++++++ 6 files changed, 340 insertions(+), 25 deletions(-) create mode 100644 crates/pdfs-fuse/src/transfers.rs diff --git a/Cargo.lock b/Cargo.lock index 390609a..059cdd7 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3085,9 +3085,9 @@ dependencies = [ [[package]] name = "proton-drive-rs" -version = "0.1.2" +version = "0.1.3" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c9380c88a032e2ff079b2d5a6221ccc40ebeee3ebbb4ea60d4d65ef770e4ca8d" +checksum = "76e4399b4d9b076c5886c7db87ae7f70322dd7127e6c5a4e7eff8719434d2434" dependencies = [ "async-trait", "base64", @@ -3110,9 +3110,9 @@ dependencies = [ [[package]] name = "proton-sdk" -version = "0.1.2" +version = "0.1.3" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1da22604d2568af2d15be6a45f23f8c5b95fa775bdac1540e833c2eafb587888" +checksum = "099f6f20709452f2cdb0cee4db0a82e61a29bec261295812908bf956ffcdaa00" dependencies = [ "aes-gcm", "async-trait", diff --git a/Cargo.toml b/Cargo.toml index c1b0f9f..4eb31f3 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -11,8 +11,8 @@ rust-version = "1.96" [workspace.dependencies] # Local Proton SDK crates (path deps until published; switch to versioned later). -proton-sdk = "0.1.2" -proton-drive-rs = "0.1.2" +proton-sdk = "0.1.3" +proton-drive-rs = "0.1.3" # Internal crates. pdfs-core = { path = "crates/pdfs-core" } @@ -33,3 +33,4 @@ clap = { version = "4", features = ["derive"] } rpassword = "7" fuser = "0.17" libc = "0.2" + diff --git a/crates/pdfs-cli/src/main.rs b/crates/pdfs-cli/src/main.rs index 74cdf03..c30c0cc 100644 --- a/crates/pdfs-cli/src/main.rs +++ b/crates/pdfs-cli/src/main.rs @@ -119,6 +119,8 @@ enum Command { /// 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<()> { @@ -151,6 +153,7 @@ fn main() -> Result<()> { 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(), } } @@ -366,6 +369,34 @@ fn cmd_unpin(path: PathBuf) -> Result<()> { 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<()> { match control_request(CtlRequest::ListPins)? { CtlResponse::Pins { pins } if pins.is_empty() => println!("No pinned files."), diff --git a/crates/pdfs-core/src/control.rs b/crates/pdfs-core/src/control.rs index cf13c9b..d80446c 100644 --- a/crates/pdfs-core/src/control.rs +++ b/crates/pdfs-core/src/control.rs @@ -89,6 +89,34 @@ pub enum Request { /// 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. @@ -178,6 +206,8 @@ pub enum Response { FilePath { path: String }, /// Full-text search results (reply to [`Request::Search`]). SearchResults { hits: Vec }, + /// A snapshot of in-flight transfers (reply to [`Request::GetQueueStatus`]). + Transfers { items: Vec }, /// The request failed. Error { message: String }, } diff --git a/crates/pdfs-fuse/src/lib.rs b/crates/pdfs-fuse/src/lib.rs index e847da1..556650e 100644 --- a/crates/pdfs-fuse/src/lib.rs +++ b/crates/pdfs-fuse/src/lib.rs @@ -52,13 +52,18 @@ use pdfs_core::cache::{BLOCK_SIZE, ContentCache}; 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::{ DriveEvent, DriveEventScopeId, Node, NodeKind, PhotosTimelineItem, ProtonDriveClient, ProtonPhotosClient, ThumbnailType, }; + +mod transfers; +use transfers::{CountingReader, CountingWriter, TransferRegistry}; use tracing::{debug, error, info, warn}; /// Attribute/entry cache lifetime handed back to the kernel. Long because the @@ -293,6 +298,9 @@ struct Core { /// requests slice memory instead of re-fetching the whole timeline per page. /// `None` until first fetched; refreshed once older than [`TIMELINE_TTL`]. timeline: Arc>>, + /// In-flight upload/download progress, served to `GetQueueStatus`. Shared + /// across the FUSE session and the control-socket task. + transfers: Arc, } /// The whole photos timeline plus when it was fetched, for TTL freshness. @@ -681,6 +689,17 @@ impl Core { error!(%uid, error = %e, "reopen scratch file failed"); Errno::EIO })?; + let name = { + let st = self.state.lock().unwrap(); + st.entries + .get(&ino) + .map(|e| e.node.name.clone()) + .unwrap_or_default() + }; + let guard = self + .transfers + .begin(name, uid.to_string(), TransferDirection::Upload, len); + let reader = CountingReader::new(reader, &guard); self.rt .block_on(self.client.upload_new_revision_from( &uid, @@ -693,6 +712,7 @@ impl Core { error!(%uid, error = %e, "upload new revision failed"); Errno::EIO })?; + drop(guard); let now = now_secs(); { let mut st = self.state.lock().unwrap(); @@ -719,6 +739,48 @@ impl Core { Ok(()) } + /// Download a whole file's plaintext, registering the transfer so + /// `GetQueueStatus` can report its progress. `total` is the expected size + /// (`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, 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, 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. @@ -746,8 +808,7 @@ impl Core { return Ok(format!("{name} ({n} files)")); } let bytes = self - .rt - .block_on(self.client.download_file(&uid)) + .download_file_tracked(&uid, &name, size) .map_err(|e| format!("download: {e}"))?; self.cache .store(&uid, mtime, size, &bytes) @@ -763,7 +824,7 @@ impl Core { /// 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 { - let mut files: Vec<(NodeUid, i64, u64)> = Vec::new(); + 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) @@ -777,6 +838,7 @@ impl Core { } else { files.push(( e.uid.clone(), + e.node.name.clone(), e.node.modification_time, node_size(&e.node), )); @@ -786,12 +848,12 @@ impl Core { } } let mut count = 0; - for (uid, mtime, size) in files { + for (uid, name, mtime, size) in files { if self.cache.is_cached(&uid, mtime, size) { count += 1; continue; } - match self.rt.block_on(self.client.download_file(&uid)) { + match self.download_file_tracked(&uid, &name, size) { Ok(bytes) => { if self.cache.store(&uid, mtime, size, &bytes).is_ok() { count += 1; @@ -1050,8 +1112,7 @@ impl Core { return Ok(p); } let bytes = self - .rt - .block_on(photos.download_photo(uid)) + .download_photo_tracked(&photos, uid, &node.name, size) .map_err(|e| format!("download photo: {e}"))?; self.cache .store(uid, mtime, size, &bytes) @@ -1067,20 +1128,19 @@ impl Core { let (ino, uid) = self .resolve_path(rel) .map_err(|e| format!("resolve path: {e:?}"))?; - let (mtime, size) = { + let (name, mtime, size) = { let st = self.state.lock().unwrap(); let e = st.entries.get(&ino).ok_or("node vanished")?; if !e.node.is_file() { 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) { return Ok(p); } let bytes = self - .rt - .block_on(self.client.download_file(&uid)) + .download_file_tracked(&uid, &name, size) .map_err(|e| format!("download: {e}"))?; self.cache .store(&uid, mtime, size, &bytes) @@ -1206,13 +1266,24 @@ impl Core { .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(&parent_uid, name, media_type_for(name), bytes), - ) + .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:?}"))?; @@ -2322,10 +2393,19 @@ fn handle_control_conn(core: &Core, username: &str, mountpoint: &Path, stream: U capture_time, ..Default::default() }; - match core - .rt - .block_on(photos.upload_photo(&name, &media_type, &bytes, metadata)) - { + 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 { @@ -2413,6 +2493,9 @@ fn handle_control_conn(core: &Core, username: &str, mountpoint: &Path, stream: U ), } } + 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 @@ -2539,6 +2622,7 @@ pub fn mount( cache: Arc::new(cache), db, timeline: Arc::new(Mutex::new(None)), + transfers: TransferRegistry::new(), }; let mut config = Config::default(); diff --git a/crates/pdfs-fuse/src/transfers.rs b/crates/pdfs-fuse/src/transfers.rs new file mode 100644 index 0000000..ac58626 --- /dev/null +++ b/crates/pdfs-fuse/src/transfers.rs @@ -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>>, + next: AtomicU64, +} + +impl TransferRegistry { + pub fn new() -> Arc { + 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, + name: impl Into, + uid: impl Into, + 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 { + 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, + id: u64, + entry: Arc, +} + +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 Write for CountingWriter<'_, W> { + fn write(&mut self, buf: &[u8]) -> std::io::Result { + 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 Read for CountingReader<'_, R> { + fn read(&mut self, buf: &mut [u8]) -> std::io::Result { + let n = self.inner.read(buf)?; + self.guard.add(n as u64); + Ok(n) + } +}