Speedup: Pipeline SMB scan-phase stats (Phase 4 Fix 4)

- Override `Volume::scan_for_copy_batch` on `SmbVolume` to fan out all per-path stats over one SMB session instead of N sequential RTTs. Each path gets a cloned `smb2::Connection` under a brief client-mutex acquire (cheap `Arc::clone` — all clones multiplex over the same TCP session), then drives `tree.stat` inside a `FuturesUnordered`.
- Trait change: `scan_for_copy_batch` now returns `BatchScanResult { aggregate, per_path }`. The copy engine's `source_hints` map is populated from `per_path`, saving the old per-source re-stat. `MtpVolume` override updated; default impl preserves the serial per-path loop for local backends.
- Single-path batches fall through to `scan_recursive` (regression guard for drag-drop-one-file). Empty root paths skip the stat and route into the directory-recursion list directly.
- Wire trace measured at N=10 on the Tailscale link: all 10 stat compounds dispatched back-to-back before any response arrives, then 10 compound reads dispatched the same way.
- Bench result (100 × 10 KB, QNAP → local, ~60 ms RTT): 6.11 s → 947 ms, a 6.5× wall-clock win. Per-file: 61 ms → 9.5 ms. Effective concurrency 156 / 9.5 ≈ 16×.
- Docs: Updated `volume/CLAUDE.md` + `write_operations/CLAUDE.md` with the `scan_for_copy_batch` signature change and the SMB override's rationale. Appended a "Phase 4 Fix 4 results (2026-04-21)" section to `docs/notes/phase4-rtt-investigation.md` with the bench numbers, wire trace, and effective-concurrency analysis.
- Tooling: Marked `env_logger` as a `#[cfg(test)] use ... as _;` shim in `lib.rs` so the dev-dep doesn't trigger `unused_crate_dependencies` when the bench's optional `try_init` is inactive.

Author: vdavid

apps/desktop/src-tauri/src/file_system/volume/CLAUDE.md

@@ -363,6 +363,12 @@ spawned detached task. This is safe because the stream always lives in an async
 **Gotcha**: SMB write streaming fallback still holds the client mutex for the whole upload
 **Why**: `SmbClient::create_file_writer` / `Tree::create_file_writer` both return `FileWriter<'a>` which borrows `&'a mut Connection`. We can't release the client mutex while the writer is alive, so the streaming (large-file) write path serializes concurrent writes on one `SmbVolume`. This is acceptable because the compound fast-path in `write_from_stream` handles every file ≤ `max_write_size` (typically 1 MB on QNAP, the SMB2 spec ceiling is 8 MB) without touching the client mutex for the actual write. Large files are rare in the hot copy path; if this ever becomes a bottleneck, the fix is a future `smb2` release that exposes a `FileWriter` built from a cloned `Connection` + `Arc<Tree>` (both owned inside the writer) rather than borrowing.
 
+**Decision**: `SmbVolume` overrides `scan_for_copy_batch` to pipeline per-path stats over a single SMB session
+**Why**: The copy pipeline's scan phase used to loop `scan_for_copy` per top-level source — N sequential RTTs on the wire before the copy phase could even start. For a 100-file copy over a ~60 ms Tailscale link that's ~5 s of serial stats. Fix 4 overrides `scan_for_copy_batch` to clone `smb2::Connection` per path under a brief client-mutex acquire (cheap `Arc::clone` — all clones multiplex over the same SMB session), release the lock, then drive `tree.stat(&mut conn, path)` on each clone inside a `FuturesUnordered`. Empty root paths skip the stat. Single-path batches fall through to `scan_recursive` so one-file drag-drops don't pay the batch machinery cost. Directories found during the stat phase recurse sequentially afterward — parallel directory recursion is a future "Fix 5". Measured 6.5× wall-clock win at 100 × 10 KB: 6.11 s → 947 ms. See `docs/notes/phase4-rtt-investigation.md` for the wire trace.
+
+**Decision**: `Volume::scan_for_copy_batch` returns `BatchScanResult { aggregate, per_path }` (changed in Phase 4 Fix 4)
+**Why**: The copy engine needs per-source type+size hints (`is_directory`, `total_bytes`) for its `source_hints` map, which seeds conflict detection and feeds the SMB compound fast-path's size hint. Pre-Fix-4 it paid N separate `scan_for_copy` calls to collect both aggregate stats and per-path info. Returning a `BatchScanResult` lets the batch scan surface both at once — one trait call, one round-trip to each backend. Scan-preview callers that only want the aggregate just read `.aggregate`. `LocalPosixVolume` and `InMemoryVolume` inherit the default (serial per-path loop, cheap); `MtpVolume` preserves its "group by parent dir" batch; `SmbVolume` overrides with the pipelined stat path.
+
 **Decision**: `SmbVolume` has a compound fast-path in `open_read_stream_with_hint` and `write_from_stream` for files ≤ `max_read_size` / `max_write_size`
 **Why**: The streaming open+read+close sequence costs 3 RTTs per file. For small files (typical 10 KB copies on a NAS) that dominates wall-clock at high-latency links (~60 ms RTT → ~180 ms/file just for protocol overhead, not data). `smb2` already exposes `Tree::read_file_compound` (CREATE+READ+CLOSE in a single compound frame = 1 RTT) and `Tree::write_file_compound` (CREATE+WRITE+FLUSH+CLOSE = 1 RTT). The copy pipeline feeds per-file size hints from the pre-copy scan; when the size is known and fits in one READ/WRITE, we take the compound path. Falls back cleanly to the streaming reader/writer when the hint is missing or the file is too big. Small compound reads return a `Vec<u8>` wrapped as a single-chunk `InlineReadStream` so the consumer API stays shaped the same. See `docs/notes/phase4-rtt-investigation.md` for the measurement.
 

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

@@ -517,9 +517,10 @@ async fn test_scan_for_copy_batch_multiple_files_same_dir() {
         PathBuf::from("/photos/c.jpg"),
     ];
     let result = volume.scan_for_copy_batch(&paths).await.unwrap();
-    assert_eq!(result.file_count, 3);
-    assert_eq!(result.dir_count, 0);
-    assert_eq!(result.total_bytes, 600);
+    assert_eq!(result.aggregate.file_count, 3);
+    assert_eq!(result.aggregate.dir_count, 0);
+    assert_eq!(result.aggregate.total_bytes, 600);
+    assert_eq!(result.per_path.len(), 3);
 }
 
 #[tokio::test]
@@ -538,18 +539,34 @@ async fn test_scan_for_copy_batch_mixed_files_and_dirs() {
 
     let paths = vec![PathBuf::from("/stuff/readme.txt"), PathBuf::from("/stuff/subdir")];
     let result = volume.scan_for_copy_batch(&paths).await.unwrap();
-    assert_eq!(result.file_count, 2); // readme.txt + deep.txt
-    assert_eq!(result.dir_count, 0); // subdir's children don't include extra dirs
-    assert_eq!(result.total_bytes, 55); // 5 + 50
+    assert_eq!(result.aggregate.file_count, 2); // readme.txt + deep.txt
+    assert_eq!(result.aggregate.dir_count, 0); // subdir's children don't include extra dirs
+    assert_eq!(result.aggregate.total_bytes, 55); // 5 + 50
+    assert_eq!(result.per_path.len(), 2);
+    // The file entry should report top_level_is_directory=false; the dir one true.
+    let readme = result
+        .per_path
+        .iter()
+        .find(|(p, _)| p == Path::new("/stuff/readme.txt"))
+        .unwrap();
+    assert!(!readme.1.top_level_is_directory);
+    assert_eq!(readme.1.total_bytes, 5);
+    let subdir = result
+        .per_path
+        .iter()
+        .find(|(p, _)| p == Path::new("/stuff/subdir"))
+        .unwrap();
+    assert!(subdir.1.top_level_is_directory);
 }
 
 #[tokio::test]
 async fn test_scan_for_copy_batch_empty_input() {
     let volume = InMemoryVolume::new("Test");
     let result = volume.scan_for_copy_batch(&[]).await.unwrap();
-    assert_eq!(result.file_count, 0);
-    assert_eq!(result.dir_count, 0);
-    assert_eq!(result.total_bytes, 0);
+    assert_eq!(result.aggregate.file_count, 0);
+    assert_eq!(result.aggregate.dir_count, 0);
+    assert_eq!(result.aggregate.total_bytes, 0);
+    assert!(result.per_path.is_empty());
 }
 
 #[tokio::test]
@@ -563,9 +580,10 @@ async fn test_scan_for_copy_batch_single_item_matches_single_scan() {
         .scan_for_copy_batch(&[PathBuf::from("/docs/a.txt")])
         .await
         .unwrap();
-    assert_eq!(single.file_count, batch.file_count);
-    assert_eq!(single.dir_count, batch.dir_count);
-    assert_eq!(single.total_bytes, batch.total_bytes);
+    assert_eq!(single.file_count, batch.aggregate.file_count);
+    assert_eq!(single.dir_count, batch.aggregate.dir_count);
+    assert_eq!(single.total_bytes, batch.aggregate.total_bytes);
+    assert_eq!(batch.per_path.len(), 1);
 }
 
 #[tokio::test]
@@ -578,8 +596,9 @@ async fn test_scan_for_copy_batch_files_from_different_dirs() {
 
     let paths = vec![PathBuf::from("/a/file1.txt"), PathBuf::from("/b/file2.txt")];
     let result = volume.scan_for_copy_batch(&paths).await.unwrap();
-    assert_eq!(result.file_count, 2);
-    assert_eq!(result.total_bytes, 30);
+    assert_eq!(result.aggregate.file_count, 2);
+    assert_eq!(result.aggregate.total_bytes, 30);
+    assert_eq!(result.per_path.len(), 2);
 }
 
 // ============================================================================

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

@@ -72,6 +72,24 @@ pub struct CopyScanResult {
     pub top_level_is_directory: bool,
 }
 
+/// Result of a batch scan over multiple source paths.
+///
+/// Returned by `Volume::scan_for_copy_batch`. Bundles the aggregate stats that
+/// the pre-flight / scan-preview callers want with a per-path breakdown that
+/// the copy engine uses to seed its `source_hints` map (without re-issuing N
+/// stat probes). `per_path[i].0` is the caller's input path verbatim; `.1`
+/// carries `top_level_is_directory` and `total_bytes` (for top-level files,
+/// that's the file size — used by the SMB compound fast-path).
+#[derive(Debug, Clone)]
+pub struct BatchScanResult {
+    /// Aggregate stats across all input paths.
+    pub aggregate: CopyScanResult,
+    /// Per-input-path result, in the same order as the `paths` slice the
+    /// caller passed in. Paths that failed to scan won't appear — on a
+    /// per-path failure the method returns `Err` without partial data.
+    pub per_path: Vec<(PathBuf, CopyScanResult)>,
+}
+
 /// A conflict detected during pre-copy scanning: a source item that already exists at the destination.
 #[derive(Debug, Clone, Serialize, Deserialize)]
 #[serde(rename_all = "camelCase")]
@@ -497,33 +515,41 @@ pub trait Volume: Send + Sync {
         Box::pin(async { Err(VolumeError::NotSupported) })
     }
 
-    /// Scans multiple paths to get aggregate copy statistics.
+    /// Scans multiple paths to get aggregate + per-path copy statistics.
     ///
     /// The default iterates over `scan_for_copy` per path, which is correct for
     /// volumes where per-path I/O is cheap (local FS, in-memory). Volume types
     /// with expensive per-path I/O (MTP, SMB, FTP, S3) should override this to
-    /// batch — typically by grouping paths by parent directory, listing each
-    /// parent once, and resolving files from the listing.
+    /// batch — typically by pipelining per-path stats over a shared session
+    /// (SMB) or grouping paths by parent directory and listing each parent
+    /// once (MTP).
+    ///
+    /// The returned `BatchScanResult` carries both the rolled-up `aggregate`
+    /// (what the scan-preview / pre-flight checks want) and a `per_path` vec
+    /// (what the copy engine uses to seed its per-source hints, so it doesn't
+    /// have to re-probe each source's type and size with a separate stat).
     fn scan_for_copy_batch<'a>(
         &'a self,
         paths: &'a [PathBuf],
-    ) -> Pin<Box<dyn Future<Output = Result<CopyScanResult, VolumeError>> + Send + 'a>> {
+    ) -> Pin<Box<dyn Future<Output = Result<BatchScanResult, VolumeError>> + Send + 'a>> {
         Box::pin(async move {
-            let mut result = CopyScanResult {
+            let mut aggregate = CopyScanResult {
                 file_count: 0,
                 dir_count: 0,
                 total_bytes: 0,
                 // Aggregate over multiple paths — meaningless for a batch.
-                // Callers that need per-path type should use `scan_for_copy`.
+                // Callers that need per-path type should read `per_path`.
                 top_level_is_directory: false,
             };
+            let mut per_path = Vec::with_capacity(paths.len());
             for path in paths {
                 let scan = self.scan_for_copy(path).await?;
-                result.file_count += scan.file_count;
-                result.dir_count += scan.dir_count;
-                result.total_bytes += scan.total_bytes;
+                aggregate.file_count += scan.file_count;
+                aggregate.dir_count += scan.dir_count;
+                aggregate.total_bytes += scan.total_bytes;
+                per_path.push((path.clone(), scan));
             }
-            Ok(result)
+            Ok(BatchScanResult { aggregate, per_path })
         })
     }
 

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

@@ -4,7 +4,8 @@
 //! the standard file listing pipeline (same icons, sorting, view modes as local files).
 
 use super::{
-    CopyScanResult, MutationEvent, ScanConflict, SourceItemInfo, SpaceInfo, Volume, VolumeError, VolumeReadStream,
+    BatchScanResult, CopyScanResult, MutationEvent, ScanConflict, SourceItemInfo, SpaceInfo, Volume, VolumeError,
+    VolumeReadStream,
 };
 use crate::file_system::listing::FileEntry;
 use crate::mtp::connection::{MtpConnectionError, connection_manager};
@@ -461,14 +462,17 @@ impl Volume for MtpVolume {
     fn scan_for_copy_batch<'a>(
         &'a self,
         paths: &'a [PathBuf],
-    ) -> Pin<Box<dyn Future<Output = Result<CopyScanResult, VolumeError>> + Send + 'a>> {
+    ) -> Pin<Box<dyn Future<Output = Result<BatchScanResult, VolumeError>> + Send + 'a>> {
         Box::pin(async move {
             if paths.is_empty() {
-                return Ok(CopyScanResult {
-                    file_count: 0,
-                    dir_count: 0,
-                    total_bytes: 0,
-                    top_level_is_directory: false,
+                return Ok(BatchScanResult {
+                    aggregate: CopyScanResult {
+                        file_count: 0,
+                        dir_count: 0,
+                        total_bytes: 0,
+                        top_level_is_directory: false,
+                    },
+                    per_path: Vec::new(),
                 });
             }
 
@@ -487,7 +491,12 @@ impl Volume for MtpVolume {
                 by_parent.len()
             );
 
-            let mut result = CopyScanResult {
+            // Stage per-path results in a map so the final per_path vec
+            // preserves the caller's input order.
+            let mut per_path_results: std::collections::HashMap<PathBuf, CopyScanResult> =
+                std::collections::HashMap::with_capacity(paths.len());
+
+            let mut aggregate = CopyScanResult {
                 file_count: 0,
                 dir_count: 0,
                 total_bytes: 0,
@@ -507,18 +516,34 @@ impl Volume for MtpVolume {
                     if let Some(entry) = entries.iter().find(|e| e.name == name) {
                         if entry.is_directory {
                             let scan = self.scan_for_copy(child_path).await?;
-                            result.file_count += scan.file_count;
-                            result.dir_count += scan.dir_count;
-                            result.total_bytes += scan.total_bytes;
+                            aggregate.file_count += scan.file_count;
+                            aggregate.dir_count += scan.dir_count;
+                            aggregate.total_bytes += scan.total_bytes;
+                            per_path_results.insert((*child_path).clone(), scan);
                         } else {
-                            result.file_count += 1;
-                            result.total_bytes += entry.size.unwrap_or(0);
+                            let size = entry.size.unwrap_or(0);
+                            aggregate.file_count += 1;
+                            aggregate.total_bytes += size;
+                            per_path_results.insert(
+                                (*child_path).clone(),
+                                CopyScanResult {
+                                    file_count: 1,
+                                    dir_count: 0,
+                                    total_bytes: size,
+                                    top_level_is_directory: false,
+                                },
+                            );
                         }
                     }
                 }
             }
 
-            Ok(result)
+            let per_path = paths
+                .iter()
+                .filter_map(|p| per_path_results.remove(p).map(|r| (p.clone(), r)))
+                .collect();
+
+            Ok(BatchScanResult { aggregate, per_path })
         })
     }