transfer progress api

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."),

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<SearchHit> },
+    /// A snapshot of in-flight transfers (reply to [`Request::GetQueueStatus`]).
+    Transfers { items: Vec<TransferItem> },
     /// The request failed.
     Error { message: String },
 }

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<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.
@@ -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<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.
@@ -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<usize, String> {
-        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();

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<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)
+    }
+}