Volume: Stream open_read_stream and write_from_stream on LocalPosixVolume

Prerequisite for Phase 4.1 trait unification. Adds a `LocalPosixReadStream` that yields 1 MiB chunks on the blocking pool and a chunked `write_from_stream` that pipes incoming stream chunks to disk via `spawn_blocking`. Flips `supports_streaming()` to `true`. Rewrote the old `export_to_local`/`import_from_local` tests onto the streaming equivalents (they were the only callers of those two methods on LocalPosix).

Author: vdavid

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

@@ -1,7 +1,8 @@
 //! Local POSIX file system volume implementation.
 
 use super::{
-    CopyScanResult, ScanConflict, SourceItemInfo, SpaceInfo, Volume, VolumeError, VolumeScanner, VolumeWatcher,
+    CopyScanResult, ScanConflict, SourceItemInfo, SpaceInfo, Volume, VolumeError, VolumeReadStream, VolumeScanner,
+    VolumeWatcher,
 };
 use crate::file_system::listing::{FileEntry, get_single_entry, list_directory_core};
 use crate::indexing::scanner::{self, ScanConfig, ScanError, ScanHandle, ScanSummary};
@@ -295,33 +296,102 @@ impl Volume for LocalPosixVolume {
         })
     }
 
-    fn export_to_local<'a>(
+    fn supports_streaming(&self) -> bool {
+        true
+    }
+
+    fn open_read_stream<'a>(
         &'a self,
-        source: &'a Path,
-        local_dest: &'a Path,
-        _on_progress: &'a (dyn Fn(u64, u64) -> std::ops::ControlFlow<()> + Sync),
-    ) -> Pin<Box<dyn Future<Output = Result<u64, VolumeError>> + Send + 'a>> {
-        let src_abs = self.resolve(source);
-        let local_dest = local_dest.to_path_buf();
+        path: &'a Path,
+    ) -> Pin<Box<dyn Future<Output = Result<Box<dyn VolumeReadStream>, VolumeError>> + Send + 'a>> {
+        let abs_path = self.resolve(path);
         Box::pin(async move {
-            spawn_blocking(move || copy_recursive(&src_abs, &local_dest))
-                .await
-                .unwrap()
+            spawn_blocking(move || {
+                let metadata = std::fs::metadata(&abs_path)?;
+                if metadata.is_dir() {
+                    return Err(VolumeError::IoError {
+                        message: "Cannot stream a directory".into(),
+                        raw_os_error: None,
+                    });
+                }
+                let total_size = metadata.len();
+                let file = std::fs::File::open(&abs_path)?;
+                Ok(Box::new(LocalPosixReadStream {
+                    file: Some(file),
+                    total_size,
+                    bytes_read: 0,
+                }) as Box<dyn VolumeReadStream>)
+            })
+            .await
+            .unwrap()
         })
     }
 
-    fn import_from_local<'a>(
+    fn write_from_stream<'a>(
         &'a self,
-        local_source: &'a Path,
         dest: &'a Path,
-        _on_progress: &'a (dyn Fn(u64, u64) -> std::ops::ControlFlow<()> + Sync),
+        size: u64,
+        mut stream: Box<dyn VolumeReadStream>,
+        on_progress: &'a (dyn Fn(u64, u64) -> std::ops::ControlFlow<()> + Sync),
     ) -> Pin<Box<dyn Future<Output = Result<u64, VolumeError>> + Send + 'a>> {
-        let local_source = local_source.to_path_buf();
         let dest_abs = self.resolve(dest);
         Box::pin(async move {
-            spawn_blocking(move || copy_recursive(&local_source, &dest_abs))
+            // Ensure parent directory exists
+            if let Some(parent) = dest_abs.parent() {
+                let parent = parent.to_path_buf();
+                spawn_blocking(move || std::fs::create_dir_all(&parent))
+                    .await
+                    .unwrap()
+                    .map_err(VolumeError::from)?;
+            }
+
+            // Open destination file on the blocking pool.
+            let dest_for_open = dest_abs.clone();
+            let mut file = spawn_blocking(move || std::fs::File::create(&dest_for_open))
                 .await
                 .unwrap()
+                .map_err(VolumeError::from)?;
+
+            let mut bytes_written = 0u64;
+            while let Some(chunk_result) = stream.next_chunk().await {
+                let chunk = chunk_result?;
+                if chunk.is_empty() {
+                    continue;
+                }
+                let chunk_len = chunk.len() as u64;
+
+                // Write the chunk on the blocking pool.
+                let (file_ret, write_res) = spawn_blocking(move || {
+                    use std::io::Write;
+                    let res = file.write_all(&chunk);
+                    (file, res)
+                })
+                .await
+                .unwrap();
+                file = file_ret;
+                write_res.map_err(VolumeError::from)?;
+
+                bytes_written += chunk_len;
+
+                if on_progress(bytes_written, size) == std::ops::ControlFlow::Break(()) {
+                    // Drop the file handle and try to clean up the partial file.
+                    drop(file);
+                    let partial = dest_abs.clone();
+                    let _ = spawn_blocking(move || std::fs::remove_file(&partial)).await;
+                    return Err(VolumeError::Cancelled("Operation cancelled by user".to_string()));
+                }
+            }
+
+            // Flush and close on the blocking pool.
+            let flush_res = spawn_blocking(move || {
+                use std::io::Write;
+                file.flush()
+            })
+            .await
+            .unwrap();
+            flush_res.map_err(VolumeError::from)?;
+
+            Ok(bytes_written)
         })
     }
 
@@ -416,30 +486,65 @@ impl VolumeWatcher for LocalPosixWatcher {
     }
 }
 
-/// Recursively copies a file or directory from source to destination.
-/// Returns total bytes copied.
-fn copy_recursive(source: &Path, dest: &Path) -> Result<u64, VolumeError> {
-    let meta = std::fs::metadata(source)?;
-    let mut total_bytes = 0;
-
-    if meta.is_file() {
-        // Copy single file
-        std::fs::copy(source, dest)?;
-        total_bytes = meta.len();
-    } else if meta.is_dir() {
-        // Create destination directory
-        std::fs::create_dir_all(dest)?;
-
-        // Copy all contents
-        for entry in std::fs::read_dir(source)? {
-            let entry = entry?;
-            let src_path = entry.path();
-            let dest_path = dest.join(entry.file_name());
-            total_bytes += copy_recursive(&src_path, &dest_path)?;
-        }
+/// Streaming reader for `LocalPosixVolume` files.
+///
+/// Reads the file in 1 MiB chunks on the blocking thread pool via
+/// `tokio::task::spawn_blocking`. Each `next_chunk` call hands the file handle
+/// to the blocking pool, reads one chunk, and returns ownership along with the
+/// data.
+struct LocalPosixReadStream {
+    file: Option<std::fs::File>,
+    total_size: u64,
+    bytes_read: u64,
+}
+
+/// 1 MiB chunks — matches `chunked_copy.rs`'s constant.
+const LOCAL_STREAM_CHUNK_SIZE: usize = 1024 * 1024;
+
+impl VolumeReadStream for LocalPosixReadStream {
+    fn next_chunk(&mut self) -> Pin<Box<dyn Future<Output = Option<Result<Vec<u8>, VolumeError>>> + Send + '_>> {
+        Box::pin(async move {
+            let mut file = self.file.take()?;
+
+            let (file_ret, result) = spawn_blocking(move || {
+                use std::io::Read;
+                let mut buf = vec![0u8; LOCAL_STREAM_CHUNK_SIZE];
+                let n = match file.read(&mut buf) {
+                    Ok(n) => n,
+                    Err(e) => return (file, Err(VolumeError::from(e))),
+                };
+                buf.truncate(n);
+                (file, Ok(buf))
+            })
+            .await
+            .unwrap();
+
+            match result {
+                Ok(buf) if buf.is_empty() => {
+                    // EOF — drop the file handle.
+                    drop(file_ret);
+                    None
+                }
+                Ok(buf) => {
+                    self.bytes_read += buf.len() as u64;
+                    self.file = Some(file_ret);
+                    Some(Ok(buf))
+                }
+                Err(e) => {
+                    drop(file_ret);
+                    Some(Err(e))
+                }
+            }
+        })
     }
 
-    Ok(total_bytes)
+    fn total_size(&self) -> u64 {
+        self.total_size
+    }
+
+    fn bytes_read(&self) -> u64 {
+        self.bytes_read
+    }
 }
 
 /// Gets space information for a path.

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

@@ -102,11 +102,11 @@ fn test_supports_watching_returns_true() {
 }
 
 #[test]
-fn test_supports_streaming_returns_false() {
-    // LocalPosixVolume uses the default implementation which returns false.
-    // Streaming is primarily for MTP-to-MTP transfers.
+fn test_supports_streaming_returns_true() {
+    // Since Phase 4, LocalPosixVolume exposes open_read_stream and
+    // write_from_stream so cross-volume copies can pipe through it.
     let volume = LocalPosixVolume::new("Test", "/tmp");
-    assert!(!volume.supports_streaming());
+    assert!(volume.supports_streaming());
 }
 
 #[tokio::test]
@@ -350,83 +350,67 @@ async fn test_scan_for_copy_directory() {
 }
 
 #[tokio::test]
-async fn test_export_to_local_single_file() {
+async fn test_open_read_stream_single_file() {
     use std::fs;
 
-    let src_dir = std::env::temp_dir().join("cmdr_export_src_test");
-    let dst_dir = std::env::temp_dir().join("cmdr_export_dst_test");
+    let src_dir = std::env::temp_dir().join("cmdr_read_stream_src_test");
     let _ = fs::remove_dir_all(&src_dir);
-    let _ = fs::remove_dir_all(&dst_dir);
     fs::create_dir_all(&src_dir).unwrap();
-    fs::create_dir_all(&dst_dir).unwrap();
 
     fs::write(src_dir.join("source.txt"), "Test content").unwrap();
 
     let volume = LocalPosixVolume::new("Test", src_dir.to_str().unwrap());
-    let bytes = volume
-        .export_to_local(Path::new("source.txt"), &dst_dir.join("dest.txt"), &|_, _| {
-            std::ops::ControlFlow::Continue(())
-        })
-        .await
-        .unwrap();
+    let mut stream = volume.open_read_stream(Path::new("source.txt")).await.unwrap();
+    assert_eq!(stream.total_size(), 12);
 
-    assert_eq!(bytes, 12); // "Test content" is 12 bytes
-    assert_eq!(fs::read_to_string(dst_dir.join("dest.txt")).unwrap(), "Test content");
+    let mut content = Vec::new();
+    while let Some(chunk) = stream.next_chunk().await {
+        content.extend_from_slice(&chunk.unwrap());
+    }
+    assert_eq!(content, b"Test content");
 
     let _ = fs::remove_dir_all(&src_dir);
-    let _ = fs::remove_dir_all(&dst_dir);
 }
 
 #[tokio::test]
-async fn test_export_to_local_directory() {
+async fn test_open_read_stream_rejects_directory() {
     use std::fs;
 
-    let src_dir = std::env::temp_dir().join("cmdr_export_dir_src_test");
-    let dst_dir = std::env::temp_dir().join("cmdr_export_dir_dst_test");
+    let src_dir = std::env::temp_dir().join("cmdr_read_stream_dir_test");
     let _ = fs::remove_dir_all(&src_dir);
-    let _ = fs::remove_dir_all(&dst_dir);
     fs::create_dir_all(&src_dir).unwrap();
-    fs::create_dir_all(&dst_dir).unwrap();
 
-    // Create source directory with files
-    let source_subdir = src_dir.join("sourcedir");
-    fs::create_dir(&source_subdir).unwrap();
-    fs::write(source_subdir.join("file1.txt"), "AAA").unwrap();
-    fs::write(source_subdir.join("file2.txt"), "BBBBB").unwrap();
+    // Create a nested dir so we can attempt to stream it.
+    fs::create_dir(src_dir.join("sourcedir")).unwrap();
 
     let volume = LocalPosixVolume::new("Test", src_dir.to_str().unwrap());
-    let bytes = volume
-        .export_to_local(Path::new("sourcedir"), &dst_dir.join("destdir"), &|_, _| {
-            std::ops::ControlFlow::Continue(())
-        })
-        .await
-        .unwrap();
-
-    assert_eq!(bytes, 8); // 3 + 5 bytes
-    assert!(dst_dir.join("destdir").is_dir());
-    assert_eq!(fs::read_to_string(dst_dir.join("destdir/file1.txt")).unwrap(), "AAA");
-    assert_eq!(fs::read_to_string(dst_dir.join("destdir/file2.txt")).unwrap(), "BBBBB");
+    let result = volume.open_read_stream(Path::new("sourcedir")).await;
+    assert!(result.is_err(), "streaming a directory should fail");
 
     let _ = fs::remove_dir_all(&src_dir);
-    let _ = fs::remove_dir_all(&dst_dir);
 }
 
 #[tokio::test]
-async fn test_import_from_local_single_file() {
+async fn test_write_from_stream_creates_file() {
+    use super::InMemoryVolume;
     use std::fs;
 
-    let local_dir = std::env::temp_dir().join("cmdr_import_local_test");
-    let vol_dir = std::env::temp_dir().join("cmdr_import_vol_test");
-    let _ = fs::remove_dir_all(&local_dir);
+    let vol_dir = std::env::temp_dir().join("cmdr_write_from_stream_test");
     let _ = fs::remove_dir_all(&vol_dir);
-    fs::create_dir_all(&local_dir).unwrap();
     fs::create_dir_all(&vol_dir).unwrap();
 
-    fs::write(local_dir.join("local.txt"), "Imported content").unwrap();
+    // Source: an in-memory file that we stream into LocalPosix.
+    let source = InMemoryVolume::new("Source");
+    source
+        .create_file(Path::new("/local.txt"), b"Imported content")
+        .await
+        .unwrap();
 
+    let stream = source.open_read_stream(Path::new("/local.txt")).await.unwrap();
+    let size = stream.total_size();
     let volume = LocalPosixVolume::new("Test", vol_dir.to_str().unwrap());
     let bytes = volume
-        .import_from_local(&local_dir.join("local.txt"), Path::new("imported.txt"), &|_, _| {
+        .write_from_stream(Path::new("imported.txt"), size, stream, &|_, _| {
             std::ops::ControlFlow::Continue(())
         })
         .await
@@ -438,7 +422,6 @@ async fn test_import_from_local_single_file() {
         "Imported content"
     );
 
-    let _ = fs::remove_dir_all(&local_dir);
     let _ = fs::remove_dir_all(&vol_dir);
 }