MTP: Fix copy progress and mid-file cancellation

- Implement `export_to_local_with_progress()` on `MtpVolume`, mirroring the existing SMB pattern
- Add `download_file_with_progress()` to MTP `file_ops` — calls the progress callback after each USB chunk and breaks on `ControlFlow::Break`
- Thread progress callback through `download_recursive_with_progress()` and `download_entries_recursive_with_progress()` in `bulk_ops`
- On cancel, clean up partial file before returning `MtpConnectionError::Cancelled`
- Add `Cancelled` variant to `MtpConnectionError` and `VolumeError`
- Previously, MTP fell through to the default `export_to_local_with_progress()` which ignored the callback entirely — no progress events, and cancel only took effect between files (not mid-file)

Author: vdavid

apps/desktop/src-tauri/src/file_system/volume/mod.rs

@@ -94,6 +94,8 @@ pub enum VolumeError {
     },
     /// Connection timed out.
     ConnectionTimeout(String),
+    /// Operation was cancelled by the user (progress callback returned Break).
+    Cancelled(String),
     IoError(String),
 }
 
@@ -108,6 +110,7 @@ impl std::fmt::Display for VolumeError {
             Self::ReadOnly(msg) => write!(f, "Read-only: {}", msg),
             Self::StorageFull { message } => write!(f, "Storage full: {}", message),
             Self::ConnectionTimeout(msg) => write!(f, "Connection timed out: {}", msg),
+            Self::Cancelled(msg) => write!(f, "Cancelled: {}", msg),
             Self::IoError(msg) => write!(f, "I/O error: {}", msg),
         }
     }

apps/desktop/src-tauri/src/file_system/volume/mtp.rs

@@ -10,6 +10,24 @@ use log::debug;
 use mtp_rs::FileDownload;
 use std::path::{Path, PathBuf};
 
+/// Wrapper to assert Send + Sync on a progress callback reference.
+///
+/// SAFETY: MtpVolume methods are called from `spawn_blocking` contexts (single OS thread).
+/// The callback never crosses thread boundaries — `block_on` runs the async download on the
+/// same thread. The Volume trait's `Fn` callbacks use atomics for interior mutation, so they
+/// are effectively Send + Sync even though the trait doesn't declare it.
+struct SendSyncProgress<'a>(&'a dyn Fn(u64, u64) -> std::ops::ControlFlow<()>);
+
+// SAFETY: See above — callback is only accessed from the single spawn_blocking thread.
+unsafe impl Send for SendSyncProgress<'_> {}
+unsafe impl Sync for SendSyncProgress<'_> {}
+
+impl SendSyncProgress<'_> {
+    fn call(&self, bytes_done: u64, bytes_total: u64) -> std::ops::ControlFlow<()> {
+        (self.0)(bytes_done, bytes_total)
+    }
+}
+
 /// A volume backed by an MTP device storage.
 ///
 /// This implementation wraps the MTP connection manager to provide file system
@@ -479,6 +497,43 @@ impl Volume for MtpVolume {
             .map_err(map_mtp_error)
     }
 
+    fn export_to_local_with_progress(
+        &self,
+        source: &Path,
+        local_dest: &Path,
+        on_progress: &dyn Fn(u64, u64) -> std::ops::ControlFlow<()>,
+    ) -> Result<u64, VolumeError> {
+        let mtp_path = self.to_mtp_path(source);
+        let device_id = self.device_id.clone();
+        let storage_id = self.storage_id;
+        let local_dest = local_dest.to_path_buf();
+
+        debug!(
+            "MtpVolume::export_to_local_with_progress: device={}, storage={}, source={}, dest={}",
+            device_id,
+            storage_id,
+            mtp_path,
+            local_dest.display()
+        );
+
+        let handle = tokio::runtime::Handle::current();
+        let progress = SendSyncProgress(on_progress);
+
+        handle
+            .block_on(async {
+                connection_manager()
+                    .download_recursive_with_progress(
+                        &device_id,
+                        storage_id,
+                        &mtp_path,
+                        &local_dest,
+                        &|bytes_done, bytes_total| progress.call(bytes_done, bytes_total),
+                    )
+                    .await
+            })
+            .map_err(map_mtp_error)
+    }
+
     fn export_to_local(&self, source: &Path, local_dest: &Path) -> Result<u64, VolumeError> {
         let mtp_path = self.to_mtp_path(source);
         let device_id = self.device_id.clone();
@@ -707,6 +762,7 @@ fn map_mtp_error(e: MtpConnectionError) -> VolumeError {
         MtpConnectionError::ExclusiveAccess { .. } | MtpConnectionError::PermissionDenied { .. } => {
             VolumeError::PermissionDenied(e.to_string())
         }
+        MtpConnectionError::Cancelled { .. } => VolumeError::Cancelled(e.to_string()),
         MtpConnectionError::Disconnected { .. } => VolumeError::DeviceDisconnected(e.to_string()),
         MtpConnectionError::Timeout { .. } => VolumeError::ConnectionTimeout(e.to_string()),
         MtpConnectionError::StorageFull { .. } => VolumeError::StorageFull { message: e.to_string() },

apps/desktop/src-tauri/src/file_system/write_operations/volume_copy.rs

@@ -1337,6 +1337,9 @@ pub(super) fn map_volume_error(context_path: &str, e: VolumeError) -> WriteOpera
         VolumeError::ConnectionTimeout(_) => WriteOperationError::ConnectionInterrupted {
             path: context_path.to_string(),
         },
+        VolumeError::Cancelled(_) => WriteOperationError::Cancelled {
+            message: "Operation cancelled by user".to_string(),
+        },
         VolumeError::IoError(msg) => WriteOperationError::IoError {
             path: context_path.to_string(),
             message: msg,

apps/desktop/src-tauri/src/mtp/connection/bulk_ops.rs

@@ -273,6 +273,144 @@ impl MtpConnectionManager {
         Ok(total_bytes)
     }
 
+    /// Downloads a file or directory recursively with a progress/cancellation callback.
+    ///
+    /// Same as `download_recursive` but calls `on_progress(bytes_done, bytes_total)` per chunk.
+    /// Returns `ControlFlow::Break(())` from the callback to cancel.
+    pub async fn download_recursive_with_progress(
+        &self,
+        device_id: &str,
+        storage_id: u32,
+        object_path: &str,
+        local_dest: &Path,
+        on_progress: &(dyn Fn(u64, u64) -> std::ops::ControlFlow<()> + Send + Sync),
+    ) -> Result<u64, MtpConnectionError> {
+        debug!(
+            "MTP download_recursive_with_progress: device={}, storage={}, path={}, dest={}",
+            device_id,
+            storage_id,
+            object_path,
+            local_dest.display()
+        );
+
+        match self.list_directory(device_id, storage_id, object_path).await {
+            Ok(entries) if !entries.is_empty() => {
+                self.download_entries_recursive_with_progress(device_id, storage_id, &entries, local_dest, on_progress)
+                    .await
+            }
+            Ok(_) => {
+                if self.is_file_by_parent(device_id, storage_id, object_path).await {
+                    let operation_id = format!("download-{}", uuid::Uuid::new_v4());
+                    let result = self
+                        .download_file_with_progress(
+                            device_id,
+                            storage_id,
+                            object_path,
+                            local_dest,
+                            None,
+                            &operation_id,
+                            Some(on_progress),
+                        )
+                        .await?;
+                    Ok(result.bytes_transferred)
+                } else {
+                    tokio::fs::create_dir_all(local_dest)
+                        .await
+                        .map_err(|e| MtpConnectionError::Other {
+                            device_id: device_id.to_string(),
+                            message: format!("Failed to create local directory: {}", e),
+                        })?;
+                    Ok(0)
+                }
+            }
+            Err(e) => {
+                if self.is_file_by_parent(device_id, storage_id, object_path).await {
+                    let operation_id = format!("download-{}", uuid::Uuid::new_v4());
+                    let result = self
+                        .download_file_with_progress(
+                            device_id,
+                            storage_id,
+                            object_path,
+                            local_dest,
+                            None,
+                            &operation_id,
+                            Some(on_progress),
+                        )
+                        .await?;
+                    Ok(result.bytes_transferred)
+                } else {
+                    Err(e)
+                }
+            }
+        }
+    }
+
+    /// Downloads directory entries recursively with a progress/cancellation callback.
+    async fn download_entries_recursive_with_progress(
+        &self,
+        device_id: &str,
+        storage_id: u32,
+        entries: &[FileEntry],
+        local_dest: &Path,
+        on_progress: &(dyn Fn(u64, u64) -> std::ops::ControlFlow<()> + Send + Sync),
+    ) -> Result<u64, MtpConnectionError> {
+        tokio::fs::create_dir_all(local_dest)
+            .await
+            .map_err(|e| MtpConnectionError::Other {
+                device_id: device_id.to_string(),
+                message: format!("Failed to create local directory: {}", e),
+            })?;
+
+        let mut total_bytes = 0u64;
+
+        for entry in entries {
+            let child_dest = local_dest.join(&entry.name);
+
+            if entry.is_directory {
+                let children = self.list_directory(device_id, storage_id, &entry.path).await?;
+                if children.is_empty() {
+                    tokio::fs::create_dir_all(&child_dest)
+                        .await
+                        .map_err(|e| MtpConnectionError::Other {
+                            device_id: device_id.to_string(),
+                            message: format!("Failed to create local directory: {}", e),
+                        })?;
+                } else {
+                    let bytes = Box::pin(self.download_entries_recursive_with_progress(
+                        device_id,
+                        storage_id,
+                        &children,
+                        &child_dest,
+                        on_progress,
+                    ))
+                    .await?;
+                    total_bytes += bytes;
+                }
+            } else {
+                let operation_id = format!("download-{}", uuid::Uuid::new_v4());
+                let result = self
+                    .download_file_with_progress(
+                        device_id,
+                        storage_id,
+                        &entry.path,
+                        &child_dest,
+                        None,
+                        &operation_id,
+                        Some(on_progress),
+                    )
+                    .await?;
+                total_bytes += result.bytes_transferred;
+            }
+        }
+
+        debug!(
+            "MTP download_entries_recursive_with_progress: directory {} complete, {} bytes",
+            local_dest.display(),
+            total_bytes
+        );
+        Ok(total_bytes)
+    }
+
     /// Checks if a path is a file by looking it up in its parent directory listing.
     async fn is_file_by_parent(&self, device_id: &str, storage_id: u32, path: &str) -> bool {
         let path_buf = normalize_mtp_path(path);

apps/desktop/src-tauri/src/mtp/connection/errors.rs

@@ -40,6 +40,10 @@ pub enum MtpConnectionError {
     PermissionDenied {
         device_id: String,
     },
+    Cancelled {
+        device_id: String,
+        message: String,
+    },
     ObjectNotFound {
         device_id: String,
         path: String,
@@ -90,6 +94,9 @@ impl std::fmt::Display for MtpConnectionError {
             Self::PermissionDenied { device_id } => {
                 write!(f, "Permission denied for device: {device_id}")
             }
+            Self::Cancelled { device_id, message } => {
+                write!(f, "Cancelled on {device_id}: {message}")
+            }
             Self::ObjectNotFound { device_id, path } => {
                 write!(f, "Object not found on {device_id}: {path}")
             }
@@ -299,6 +306,10 @@ mod tests {
             MtpConnectionError::PermissionDenied {
                 device_id: "test".to_string(),
             },
+            MtpConnectionError::Cancelled {
+                device_id: "test".to_string(),
+                message: "cancelled".to_string(),
+            },
             MtpConnectionError::ObjectNotFound {
                 device_id: "test".to_string(),
                 path: "/path".to_string(),

apps/desktop/src-tauri/src/mtp/connection/file_ops.rs

@@ -33,6 +33,25 @@ impl MtpConnectionManager {
         local_dest: &Path,
         app: Option<&AppHandle>,
         operation_id: &str,
+    ) -> Result<MtpOperationResult, MtpConnectionError> {
+        self.download_file_with_progress(device_id, storage_id, object_path, local_dest, app, operation_id, None)
+            .await
+    }
+
+    /// Downloads a file from the MTP device to a local path, with optional progress/cancellation callback.
+    ///
+    /// The `on_progress` callback receives `(bytes_done, bytes_total)` and returns `ControlFlow::Break(())`
+    /// to cancel the transfer. On cancellation, the partial file is removed.
+    #[allow(clippy::too_many_arguments, reason = "mirrors download_file with added on_progress callback")]
+    pub async fn download_file_with_progress(
+        &self,
+        device_id: &str,
+        storage_id: u32,
+        object_path: &str,
+        local_dest: &Path,
+        app: Option<&AppHandle>,
+        operation_id: &str,
+        on_progress: Option<&(dyn Fn(u64, u64) -> std::ops::ControlFlow<()> + Send + Sync)>,
     ) -> Result<MtpOperationResult, MtpConnectionError> {
         debug!(
             "MTP download_file: device={}, storage={}, path={}, dest={}",
@@ -117,6 +136,7 @@ impl MtpConnectionManager {
 
         // Write chunks to file (must complete before releasing device lock)
         let mut bytes_written = 0u64;
+        let mut cancelled = false;
         while let Some(chunk_result) = download.next_chunk().await {
             let chunk = chunk_result.map_err(|e| MtpConnectionError::Other {
                 device_id: device_id.to_string(),
@@ -129,12 +149,29 @@ impl MtpConnectionManager {
             })?;
 
             bytes_written += chunk.len() as u64;
+
+            // Report progress and check for cancellation
+            if let Some(ref cb) = on_progress
+                && cb(bytes_written, total_size).is_break() {
+                    cancelled = true;
+                    break;
+                }
         }
 
         // Release device lock after download completes
         drop(storage);
         drop(device);
 
+        // On cancellation, clean up the partial file
+        if cancelled {
+            drop(file);
+            let _ = tokio::fs::remove_file(local_dest).await;
+            return Err(MtpConnectionError::Cancelled {
+                device_id: device_id.to_string(),
+                message: "Download cancelled".to_string(),
+            });
+        }
+
         file.flush().await.map_err(|e| MtpConnectionError::Other {
             device_id: device_id.to_string(),
             message: format!("Failed to flush local file: {}", e),