Speedup: Compound read/write fast-path for small SMB files

vdavid · vdavid · commit 9d6df0e9f612 · 2026-04-21T23:20:45.000+02:00
- Added `Volume::open_read_stream_with_hint(path, size_hint)` to the trait (default falls through to `open_read_stream`). Backends that can exploit a size hint (SMB) override it.
- `SmbVolume` now picks `Tree::read_file_compound` (1 RTT: CREATE+READ+CLOSE in one compound frame) when the caller-provided hint is ≤ negotiated `max_read_size`. Returned bytes are wrapped as a single-chunk `InlineReadStream` so the consumer API is unchanged. Falls back cleanly to the streaming path when the hint is missing, too big, or the compound read returns short.
- `SmbVolume::write_from_stream` drains the source stream when `size ≤ max_write_size` and calls `Tree::write_file_compound` (1 RTT: CREATE+WRITE+FLUSH+CLOSE). Only enabled when size is known up-front — streaming path handles unknown/large sizes.
- `copy_single_path` and `stream_pipe_file` thread a `source_size_hint: Option&lt;u64&gt;` through. The scan HashMap now carries a `SourceHint { is_directory, size }` struct; for top-level files the size is `CopyScanResult.total_bytes`. `copy_directory_streaming` uses `entry.size` from the directory listing.
- On a 60 ms RTT link, a 10 KB SMB→Local copy drops from 3 RTTs + 2 stat probes (Fix 1 removed 1) down to 1 RTT on the read side. The same treatment on the write side collapses 4 RTTs to 1 for small Local→SMB copies.

CLAUDE.md in `volume/` documents the decision; eval numbers live in `docs/notes/phase4-rtt-investigation.md`.
diff --git a/apps/desktop/src-tauri/src/file_system/volume/CLAUDE.md b/apps/desktop/src-tauri/src/file_system/volume/CLAUDE.md
@@ -360,6 +360,9 @@ spawned detached task. This is safe because the stream always lives in an async
 **Gotcha**: Watcher filenames are NFC (from server) but macOS mount paths are NFD
 **Why**: SMB servers return NFC-normalized filenames. macOS filesystem paths use NFD. The watcher NFD-normalizes filenames before constructing display paths used for cache lookups.
 
+**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.
+
 **Decision**: Phase 4 collapsed `export_to_local` / `import_from_local` onto `open_read_stream` / `write_from_stream`
 **Why**: The three pre-Phase-4 copy paths (local↔local, local↔volume, volume↔volume) duplicated the same "open a reader, pipe to a writer" logic in three different shapes. The APFS clonefile fast path is the only one with a real capability difference. Collapsing the other two to a single streaming path means new backends (S3, WebDAV, FTP) implement two methods instead of four, concurrency lives in one place (`volume_copy.rs` — Phase 4.2), and features like resume / checksum / progress benefit every direction at once. See `docs/notes/phase4-volume-copy-unification.md`.
 
diff --git a/apps/desktop/src-tauri/src/file_system/volume/mod.rs b/apps/desktop/src-tauri/src/file_system/volume/mod.rs
@@ -599,6 +599,29 @@ pub trait Volume: Send + Sync {
         Box::pin(async { Err(VolumeError::NotSupported) })
     }
 
+    /// Opens a streaming reader with an optional size hint from the caller.
+    ///
+    /// Network-backed volumes can use the hint to pick a faster compound
+    /// request path for small files (e.g., SMB's CREATE+READ+CLOSE compound)
+    /// instead of the 3-RTT streaming open. Backends that can't use the hint
+    /// fall through to `open_read_stream`.
+    ///
+    /// The hint is best-effort — callers pass `None` when they don't know
+    /// the size ahead of time, and the backend must work correctly either
+    /// way.
+    #[allow(
+        clippy::type_complexity,
+        reason = "async trait method returns a pinned boxed future by design"
+    )]
+    fn open_read_stream_with_hint<'a>(
+        &'a self,
+        path: &'a Path,
+        size_hint: Option<u64>,
+    ) -> Pin<Box<dyn Future<Output = Result<Box<dyn VolumeReadStream>, VolumeError>> + Send + 'a>> {
+        let _ = size_hint;
+        self.open_read_stream(path)
+    }
+
     /// Writes data from a stream to the given path.
     ///
     /// `on_progress(bytes_written, total_size)` is called after each chunk is
diff --git a/apps/desktop/src-tauri/src/file_system/volume/smb.rs b/apps/desktop/src-tauri/src/file_system/volume/smb.rs
@@ -582,6 +582,46 @@ impl VolumeReadStream for SmbReadStream {
     }
 }
 
+/// Wraps a pre-read `Vec<u8>` as a `VolumeReadStream` that yields the whole
+/// buffer as a single chunk. Used by the compound fast-path in
+/// `open_read_stream_with_hint`, where the full file body came back inside one
+/// SMB compound response — there's no more I/O to drive, just hand the bytes
+/// to the consumer.
+struct InlineReadStream {
+    data: Option<Vec<u8>>,
+    total_size: u64,
+    bytes_read: u64,
+}
+
+impl InlineReadStream {
+    fn new(data: Vec<u8>) -> Self {
+        let total_size = data.len() as u64;
+        Self {
+            data: Some(data),
+            total_size,
+            bytes_read: 0,
+        }
+    }
+}
+
+impl VolumeReadStream for InlineReadStream {
+    fn next_chunk(&mut self) -> Pin<Box<dyn Future<Output = Option<Result<Vec<u8>, VolumeError>>> + Send + '_>> {
+        Box::pin(async move {
+            let data = self.data.take()?;
+            self.bytes_read = data.len() as u64;
+            Some(Ok(data))
+        })
+    }
+
+    fn total_size(&self) -> u64 {
+        self.total_size
+    }
+
+    fn bytes_read(&self) -> u64 {
+        self.bytes_read
+    }
+}
+
 /// If an smb2 error indicates the session is dead, transition state to
 /// `Disconnected`. Mirrors `handle_smb_result` for contexts without `&self`.
 fn update_state_on_smb_error(state: &AtomicU8, err: &smb2::Error) {
@@ -1074,6 +1114,54 @@ impl Volume for SmbVolume {
         })
     }
 
+    fn open_read_stream_with_hint<'a>(
+        &'a self,
+        path: &'a Path,
+        size_hint: Option<u64>,
+    ) -> Pin<Box<dyn Future<Output = Result<Box<dyn VolumeReadStream>, VolumeError>> + Send + 'a>> {
+        Box::pin(async move {
+            let smb_path = self.to_smb_path(path);
+
+            // Compound fast-path: if the caller-provided hint fits in one READ,
+            // send CREATE+READ+CLOSE as a single compound frame (1 RTT) instead
+            // of the 3-RTT streaming open. Falls through to the streaming path
+            // when the hint is missing, too large, or the compound read returns
+            // short (truncated file — rare but possible if size changed since
+            // the scan).
+            if let Some(size) = size_hint {
+                let mut guard = self.acquire_smb().await?;
+                let (client, tree) = guard.as_mut().unwrap();
+                let max_read = client.params().map(|p| p.max_read_size).unwrap_or(65536) as u64;
+                if size > 0 && size <= max_read {
+                    debug!(
+                        "SmbVolume::open_read_stream_with_hint: share={}, path={:?}, size={} — using compound fast-path",
+                        self.share_name, smb_path, size
+                    );
+                    let read_result = tree.read_file_compound(client.connection_mut(), &smb_path).await;
+                    let data = self.handle_smb_result("open_read_stream_with_hint(compound)", read_result)?;
+                    if data.len() as u64 == size {
+                        drop(guard);
+                        return Ok(Box::new(InlineReadStream::new(data)) as Box<dyn VolumeReadStream>);
+                    }
+                    debug!(
+                        "SmbVolume::open_read_stream_with_hint: compound read returned {} bytes, expected {} — falling back to streaming",
+                        data.len(),
+                        size
+                    );
+                    // Fall through — release the guard first.
+                }
+                drop(guard);
+            }
+
+            debug!(
+                "SmbVolume::open_read_stream_with_hint: share={}, path={:?} — using streaming path",
+                self.share_name, smb_path
+            );
+            let stream = self.open_smb_download_stream(&smb_path).await?;
+            Ok(Box::new(stream) as Box<dyn VolumeReadStream>)
+        })
+    }
+
     fn write_from_stream<'a>(
         &'a self,
         dest: &'a Path,
@@ -1085,8 +1173,8 @@ impl Volume for SmbVolume {
             let smb_path = self.to_smb_path(dest);
 
             debug!(
-                "SmbVolume::write_from_stream: share={}, path={:?}",
-                self.share_name, smb_path
+                "SmbVolume::write_from_stream: share={}, path={:?}, size={}",
+                self.share_name, smb_path, size
             );
 
             // Holds the SMB session mutex for the duration of the transfer.
@@ -1095,6 +1183,51 @@ impl Volume for SmbVolume {
             let mut guard = self.acquire_smb().await?;
             let (client, tree) = guard.as_mut().unwrap();
 
+            // Compound fast-path: when the caller promised a size that fits in
+            // one WRITE, drain the source stream into a buffer and send
+            // CREATE+WRITE+FLUSH+CLOSE as a single compound frame (1 RTT
+            // instead of 4). Small files are the hot case — for anything
+            // larger we use the streaming writer below.
+            let max_write = client.params().map(|p| p.max_write_size).unwrap_or(65536) as u64;
+            if size > 0 && size <= max_write {
+                let mut buffer = Vec::with_capacity(size as usize);
+                while let Some(chunk_result) = stream.next_chunk().await {
+                    let chunk = chunk_result?;
+                    buffer.extend_from_slice(&chunk);
+                }
+                if buffer.len() as u64 == size {
+                    debug!(
+                        "SmbVolume::write_from_stream: using compound fast-path ({} bytes)",
+                        buffer.len()
+                    );
+                    let write_result = tree
+                        .write_file_compound(client.connection_mut(), &smb_path, &buffer)
+                        .await;
+                    let bytes_written = self.handle_smb_result("write_from_stream(compound)", write_result)?;
+                    // Emit a single progress tick so counters match the
+                    // streaming path's post-loop state.
+                    let _ = on_progress(bytes_written, size);
+                    return Ok(bytes_written);
+                }
+                // Size mismatch — fall back to the streaming path so a
+                // subsequent open writes whatever came through.
+                debug!(
+                    "SmbVolume::write_from_stream: compound fast-path source yielded {} bytes, expected {} — falling back",
+                    buffer.len(),
+                    size
+                );
+                // Re-feed the already-drained buffer via the streaming writer.
+                let writer_result = client.create_file_writer(tree, &smb_path).await;
+                let mut writer = self.handle_smb_result("write_from_stream(open)", writer_result)?;
+                if !buffer.is_empty() {
+                    let write_result = writer.write_chunk(&buffer).await;
+                    self.handle_smb_result("write_from_stream(write_chunk)", write_result)?;
+                }
+                let finish_result = writer.finish().await;
+                self.handle_smb_result("write_from_stream(finish)", finish_result)?;
+                return Ok(buffer.len() as u64);
+            }
+
             let writer_result = client.create_file_writer(tree, &smb_path).await;
             let mut writer = self.handle_smb_result("write_from_stream(open)", writer_result)?;
 
diff --git a/apps/desktop/src-tauri/src/file_system/write_operations/volume_copy.rs b/apps/desktop/src-tauri/src/file_system/write_operations/volume_copy.rs
@@ -28,6 +28,16 @@ use futures_util::StreamExt;
 use futures_util::stream::FuturesUnordered;
 
 use super::scan::take_cached_scan_result;
+
+/// Per-source hints collected during the scan phase, so the copy loop can
+/// skip re-probing the source type/size per file. `size` is only meaningful
+/// when `is_directory == false`; it's the top-level file's size and feeds
+/// the SMB compound fast-path.
+#[derive(Clone, Copy, Default)]
+struct SourceHint {
+    is_directory: bool,
+    size: u64,
+}
 use super::state::{
     OperationIntent, WRITE_OPERATION_STATE, WriteOperationState, is_cancelled, load_intent, register_operation_status,
     unregister_operation_status, update_operation_status,
@@ -298,10 +308,11 @@ async fn copy_volumes_with_progress(
     let mut total_files;
     let mut total_bytes;
 
-    // Per-source `is_directory` hint collected during the scan, so the copy
-    // loop doesn't need to re-stat each source path via `is_directory`.
-    // Saves one round-trip per file on network-backed volumes (SMB, MTP).
-    let mut source_is_directory: HashMap<PathBuf, bool> = HashMap::with_capacity(source_paths.len());
+    // Per-source hint collected during the scan: whether the top-level path
+    // is a directory and, for top-level files, the file size. The copy loop
+    // reuses these to skip an `is_directory` probe per file and, for SMB, to
+    // pick the 1-RTT compound fast-path when the file fits in one READ.
+    let mut source_hints: HashMap<PathBuf, SourceHint> = HashMap::with_capacity(source_paths.len());
 
     if let Some(cached) = config.preview_id.as_deref().and_then(take_cached_scan_result) {
         total_files = cached.file_count;
@@ -312,12 +323,19 @@ async fn copy_volumes_with_progress(
             total_files,
             total_bytes
         );
-        // TODO: extend the preview cache to carry per-source type so this branch
-        // doesn't need to re-stat. For now, the preview path already saved one
-        // scan per source — this extra stat is bounded by source count.
+        // TODO: extend the preview cache to carry per-source type + size so this
+        // branch doesn't need to re-stat. For now, the preview path already saved
+        // one full scan per source — this extra stat is bounded by source count
+        // and the compound fast-path falls back cleanly when size is unknown.
         for source_path in source_paths {
             let is_dir = source_volume.is_directory(source_path).await.unwrap_or(false);
-            source_is_directory.insert(source_path.clone(), is_dir);
+            source_hints.insert(
+                source_path.clone(),
+                SourceHint {
+                    is_directory: is_dir,
+                    size: 0,
+                },
+            );
         }
     } else {
         log::debug!(
@@ -354,7 +372,20 @@ async fn copy_volumes_with_progress(
             total_files += scan.file_count;
             total_dirs += scan.dir_count;
             total_bytes += scan.total_bytes;
-            source_is_directory.insert(source_path.clone(), scan.top_level_is_directory);
+            // For top-level files, `scan.total_bytes` == the file size.
+            // For directories, we leave `size = 0` (unused downstream).
+            let size = if scan.top_level_is_directory {
+                0
+            } else {
+                scan.total_bytes
+            };
+            source_hints.insert(
+                source_path.clone(),
+                SourceHint {
+                    is_directory: scan.top_level_is_directory,
+                    size,
+                },
+            );
         }
 
         log::debug!(
@@ -485,7 +516,7 @@ async fn copy_volumes_with_progress(
                     // branch that didn't populate it), default to false —
                     // behaves as if the source is a file (worst case: one extra
                     // conflict prompt vs. silent merge for dir-over-dir).
-                    let source_is_dir = source_is_directory.get(source_path).copied().unwrap_or(false);
+                    let source_is_dir = source_hints.get(source_path).map(|h| h.is_directory).unwrap_or(false);
                     let dest_is_dir = dest_meta.is_directory;
                     if source_is_dir && dest_is_dir {
                         log::debug!(
@@ -546,7 +577,9 @@ async fn copy_volumes_with_progress(
                 let source_owned = source_path.clone();
                 let dest_owned = dest_item_path.clone();
                 let file_name_owned = file_name.clone();
-                let source_is_dir_hint = source_is_directory.get(source_path).copied().unwrap_or(false);
+                let hint = source_hints.get(source_path).copied().unwrap_or_default();
+                let source_is_dir_hint = hint.is_directory;
+                let source_size_hint = if hint.is_directory { None } else { Some(hint.size) };
 
                 in_flight.push(Box::pin(async move {
                     // Per-task `last_file_bytes` tracks bytes reported for the
@@ -593,6 +626,7 @@ async fn copy_volumes_with_progress(
                         &src_vol,
                         &source_owned,
                         source_is_dir_hint,
+                        source_size_hint,
                         &dst_vol,
                         &dest_owned,
                         &state_clone,
@@ -672,7 +706,7 @@ async fn copy_volumes_with_progress(
             if let Ok(dest_meta) = dest_volume.get_metadata(&dest_item_path).await {
                 // Reuse the per-source hint from the scan; see note in the
                 // concurrent path for the fallback.
-                let source_is_dir = source_is_directory.get(source_path).copied().unwrap_or(false);
+                let source_is_dir = source_hints.get(source_path).map(|h| h.is_directory).unwrap_or(false);
                 let dest_is_dir = dest_meta.is_directory;
                 if source_is_dir && dest_is_dir {
                     log::debug!(
@@ -766,11 +800,14 @@ async fn copy_volumes_with_progress(
 
             last_dest_path = Some(dest_item_path.clone());
 
-            let source_is_dir_hint = source_is_directory.get(source_path).copied().unwrap_or(false);
+            let hint = source_hints.get(source_path).copied().unwrap_or_default();
+            let source_is_dir_hint = hint.is_directory;
+            let source_size_hint = if hint.is_directory { None } else { Some(hint.size) };
             match copy_single_path(
                 &source_volume,
                 source_path,
                 source_is_dir_hint,
+                source_size_hint,
                 &dest_volume,
                 &dest_item_path,
                 state,
diff --git a/apps/desktop/src-tauri/src/file_system/write_operations/volume_move.rs b/apps/desktop/src-tauri/src/file_system/write_operations/volume_move.rs
@@ -186,6 +186,10 @@ pub async fn move_between_volumes(
                     &source_volume,
                     source_path,
                     source_is_dir,
+                    // Move has no scan phase to cache a size hint. The SMB
+                    // compound fast-path falls through to streaming cleanly
+                    // when the hint is missing.
+                    None,
                     &dest_volume,
                     &dest_item,
                     &state,
diff --git a/apps/desktop/src-tauri/src/file_system/write_operations/volume_strategy.rs b/apps/desktop/src-tauri/src/file_system/write_operations/volume_strategy.rs
