Volume: Add progress + cancellation to streaming writes

- `write_from_stream` signature now includes `on_progress` callback that doubles as cancellation (return `ControlFlow::Break`)
- `SmbVolume` uses smb2's `write_file_streamed` with the callback bridging progress/cancel per chunk
- Streaming branch in `copy_single_path` passes progress and calls `on_file_complete`
- Updated smb2 dep to v0.6.0 (streaming write support)

Note: MTP→SMB currently panics due to nested `block_on` (MtpReadStream calls block_on inside SmbVolume's block_on). Next commit fixes this with a channel-based MTP stream.

Author: vdavid

Cargo.lock

@@ -558,12 +558,22 @@ pub trait Volume: Send + Sync {
 
     /// Writes data from a stream to the given path.
     ///
+    /// `on_progress(bytes_written, total_size)` is called after each chunk is
+    /// written. Return `ControlFlow::Break(())` to cancel the transfer.
+    ///
     /// # Arguments
     /// * `dest` - Destination path (file will be created/overwritten)
     /// * `size` - Total size in bytes (required for protocols like MTP)
     /// * `stream` - Source data stream
-    fn write_from_stream(&self, dest: &Path, size: u64, stream: Box<dyn VolumeReadStream>) -> Result<u64, VolumeError> {
-        let _ = (dest, size, stream);
+    /// * `on_progress` - Progress callback; return `ControlFlow::Break(())` to cancel
+    fn write_from_stream(
+        &self,
+        dest: &Path,
+        size: u64,
+        stream: Box<dyn VolumeReadStream>,
+        on_progress: &dyn Fn(u64, u64) -> std::ops::ControlFlow<()>,
+    ) -> Result<u64, VolumeError> {
+        let _ = (dest, size, stream, on_progress);
         Err(VolumeError::NotSupported)
     }
 }

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

@@ -759,6 +759,7 @@ impl Volume for MtpVolume {
         dest: &Path,
         size: u64,
         mut stream: Box<dyn VolumeReadStream>,
+        _on_progress: &dyn Fn(u64, u64) -> std::ops::ControlFlow<()>,
     ) -> Result<u64, VolumeError> {
         let dest_folder = dest.parent().map(|p| self.to_mtp_path(p)).unwrap_or_default();
         let filename = dest

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

@@ -1136,8 +1136,9 @@ impl Volume for SmbVolume {
     fn write_from_stream(
         &self,
         dest: &Path,
-        _size: u64,
+        size: u64,
         mut stream: Box<dyn VolumeReadStream>,
+        on_progress: &dyn Fn(u64, u64) -> std::ops::ControlFlow<()>,
     ) -> Result<u64, VolumeError> {
         let smb_path = self.to_smb_path(dest);
         let handle = self.runtime_handle.clone();
@@ -1147,19 +1148,47 @@ impl Volume for SmbVolume {
             self.share_name, smb_path
         );
 
-        // Collect all chunks into a buffer, then write in one pipelined call
-        let mut data = Vec::new();
-        while let Some(result) = stream.next_chunk() {
-            data.extend_from_slice(&result?);
-        }
-
-        let len = data.len() as u64;
+        // Bridge VolumeReadStream → smb2's write_file_streamed callback.
+        // smb2 pulls chunks on demand, so memory is bounded to the sliding window.
+        let mut bytes_written = 0u64;
+        let mut cancelled = false;
         let sp = smb_path;
-        self.with_smb("write_from_stream", |client, tree| {
-            handle.block_on(client.write_file_pipelined(tree, &sp, &data))
-        })?;
 
-        Ok(len)
+        let result = self.with_smb("write_from_stream", |client, tree| {
+            handle.block_on(client.write_file_streamed(tree, &sp, &mut || {
+                // Check cancellation via the progress callback
+                if cancelled {
+                    return Some(Err(std::io::Error::new(
+                        std::io::ErrorKind::Interrupted,
+                        "Operation cancelled",
+                    )));
+                }
+
+                match stream.next_chunk() {
+                    Some(Ok(chunk)) => {
+                        bytes_written += chunk.len() as u64;
+                        if on_progress(bytes_written, size) == std::ops::ControlFlow::Break(()) {
+                            cancelled = true;
+                            // Don't return error here — let smb2 finish writing this chunk
+                            // so we don't corrupt the connection state. Next call will error.
+                        }
+                        Some(Ok(chunk))
+                    }
+                    Some(Err(e)) => Some(Err(std::io::Error::other(e.to_string()))),
+                    None => None,
+                }
+            }))
+        });
+
+        if cancelled {
+            return Err(VolumeError::IoError {
+                message: "Operation cancelled".to_string(),
+                raw_os_error: None,
+            });
+        }
+
+        result?;
+        Ok(bytes_written)
     }
 
     fn smb_connection_state(&self) -> Option<SmbConnectionState> {

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

@@ -77,7 +77,9 @@ pub(super) fn copy_single_path(
             );
             let stream = source_volume.open_read_stream(source_path)?;
             let size = stream.total_size();
-            return dest_volume.write_from_stream(dest_path, size, stream);
+            let bytes = dest_volume.write_from_stream(dest_path, size, stream, on_file_progress)?;
+            on_file_complete();
+            return Ok(bytes);
         }
 
         // Neither supports streaming — fall back to temp local (export then import)