Skip to content

Commit ac71bd0

Browse files
vdavidvdavid
committed
SMB: Add streaming support for cross-volume copies
`SmbVolume` now implements `open_read_stream` and `write_from_stream`, enabling direct MTP↔SMB and SMB↔SMB file copies without temp files. Data flows through memory (smb2 pipelined read → 1 MB chunk iterator → smb2 pipelined write). - `SmbReadStream` yields fixed-size chunks from the in-memory buffer - `volume_strategy.rs` fallback changed from `NotSupported` to `copy_via_temp_local` for safety - Streaming is the universal path for non-local volume pairs; future volume types (FTP, S3) just implement these two methods
1 parent 70978c8 commit ac71bd0

7 files changed

Lines changed: 184 additions & 23 deletions

File tree

‎apps/desktop/src-tauri/src/commands/network.rs‎

Lines changed: 5 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -473,18 +473,12 @@ pub async fn upgrade_to_smb_volume_with_credentials(
473473
match result {
474474
Ok(()) => {
475475
// Save credentials on success if requested
476-
if remember_in_keychain
477-
&& let (Some(u), Some(p)) = (&username, &password) {
478-
let server_key = hostname.as_deref().unwrap_or(&info.server);
479-
if let Err(e) = keychain::save_credentials(
480-
server_key,
481-
Some(&info.share),
482-
u,
483-
p,
484-
) {
485-
log::warn!("Couldn't save credentials to Keychain: {}", e);
486-
}
476+
if remember_in_keychain && let (Some(u), Some(p)) = (&username, &password) {
477+
let server_key = hostname.as_deref().unwrap_or(&info.server);
478+
if let Err(e) = keychain::save_credentials(server_key, Some(&info.share), u, p) {
479+
log::warn!("Couldn't save credentials to Keychain: {}", e);
487480
}
481+
}
488482
Ok(UpgradeResult::Success)
489483
}
490484
Err(UpgradeError::Auth) => Ok(UpgradeResult::CredentialsNeeded {

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

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -220,7 +220,8 @@ pub fn upgrade_existing_smb_mounts() {
220220

221221
// Try Keychain creds
222222
let creds =
223-
crate::network::smb_upgrade::get_keychain_password(&info.server, hostname.as_deref(), &info.share).await;
223+
crate::network::smb_upgrade::get_keychain_password(&info.server, hostname.as_deref(), &info.share)
224+
.await;
224225

225226
let (username, password) = match &creds {
226227
Some((u, p)) => (Some(u.as_str()), Some(p.as_str())),

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

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -39,7 +39,7 @@ Optional methods default to `Err(VolumeError::NotSupported)` or `false`, so new
3939

4040
- `supports_watching()` — enables the `notify`-based *listing* file watcher in `operations.rs` (separate from the `VolumeWatcher` trait used for drive indexing). `MtpVolume` returns `false` (it has its own USB event loop).
4141
- `supports_export()` — enables copy/move UI. Local, MTP, and SMB return `true`.
42-
- `supports_streaming()` — enables chunked MTP-to-MTP transfers. Only `MtpVolume` returns `true`.
42+
- `supports_streaming()` — enables cross-volume transfers via `open_read_stream` / `write_from_stream`. `MtpVolume` and `SmbVolume` return `true`. The streaming path is the universal fallback for any non-local volume pair — future volume types (FTP, S3) just implement these two methods to get cross-volume copy for free.
4343
- `local_path()` — returns `Some` only for local volumes; allows `copyfile(2)` fast-path in copy operations. `SmbVolume` returns `None` so copies go through smb2 instead of the slow OS mount.
4444
- `supports_local_fs_access()` — whether `std::fs` operations (stat, read_dir) work on this volume's paths. Default `true`. `MtpVolume` and `SmbVolume` return `false`. Used to skip the legacy synthetic entry diff path (now superseded by `notify_mutation`).
4545
- `notify_mutation(volume_id, parent_path, mutation)` — called after a successful mutation (create, delete, rename) to update the listing cache immediately. Default impl uses `std::fs` (works for `LocalPosixVolume`). `SmbVolume` and `MtpVolume` override to use their own protocol's `get_metadata`. Fire-and-forget, no error propagation.

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

Lines changed: 164 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -6,7 +6,8 @@
66
//! performance and fail-fast behavior.
77
88
use super::{
9-
CopyScanResult, ScanConflict, SmbConnectionState, SourceItemInfo, SpaceInfo, Volume, VolumeError, path_to_id,
9+
CopyScanResult, ScanConflict, SmbConnectionState, SourceItemInfo, SpaceInfo, Volume, VolumeError, VolumeReadStream,
10+
path_to_id,
1011
};
1112
use crate::file_system::listing::FileEntry;
1213
use log::{debug, warn};
@@ -561,6 +562,41 @@ impl SmbVolume {
561562
}
562563
}
563564

565+
// ── Streaming support ────────────────────────��─────────────────────
566+
567+
/// Chunk size for `SmbReadStream` iteration (1 MB).
568+
const SMB_STREAM_CHUNK_SIZE: usize = 1024 * 1024;
569+
570+
/// Streaming reader for SMB files.
571+
///
572+
/// The file data is read into memory in one shot via smb2's pipelined read,
573+
/// then yielded in fixed-size chunks. This avoids temp files for cross-volume
574+
/// copies (MTP↔SMB) while keeping memory bounded to one file at a time.
575+
struct SmbReadStream {
576+
data: Vec<u8>,
577+
offset: usize,
578+
}
579+
580+
impl VolumeReadStream for SmbReadStream {
581+
fn next_chunk(&mut self) -> Option<Result<Vec<u8>, VolumeError>> {
582+
if self.offset >= self.data.len() {
583+
return None;
584+
}
585+
let end = (self.offset + SMB_STREAM_CHUNK_SIZE).min(self.data.len());
586+
let chunk = self.data[self.offset..end].to_vec();
587+
self.offset = end;
588+
Some(Ok(chunk))
589+
}
590+
591+
fn total_size(&self) -> u64 {
592+
self.data.len() as u64
593+
}
594+
595+
fn bytes_read(&self) -> u64 {
596+
self.offset as u64
597+
}
598+
}
599+
564600
impl Volume for SmbVolume {
565601
fn name(&self) -> &str {
566602
&self.name
@@ -1076,6 +1112,56 @@ impl Volume for SmbVolume {
10761112
Ok(conflicts)
10771113
}
10781114

1115+
fn supports_streaming(&self) -> bool {
1116+
true
1117+
}
1118+
1119+
fn open_read_stream(&self, path: &Path) -> Result<Box<dyn VolumeReadStream>, VolumeError> {
1120+
let smb_path = self.to_smb_path(path);
1121+
let handle = self.runtime_handle.clone();
1122+
let sp = smb_path.clone();
1123+
1124+
debug!(
1125+
"SmbVolume::open_read_stream: share={}, path={:?}",
1126+
self.share_name, smb_path
1127+
);
1128+
1129+
let data = self.with_smb("open_read_stream", |client, tree| {
1130+
handle.block_on(client.read_file_pipelined(tree, &sp))
1131+
})?;
1132+
1133+
Ok(Box::new(SmbReadStream { data, offset: 0 }))
1134+
}
1135+
1136+
fn write_from_stream(
1137+
&self,
1138+
dest: &Path,
1139+
_size: u64,
1140+
mut stream: Box<dyn VolumeReadStream>,
1141+
) -> Result<u64, VolumeError> {
1142+
let smb_path = self.to_smb_path(dest);
1143+
let handle = self.runtime_handle.clone();
1144+
1145+
debug!(
1146+
"SmbVolume::write_from_stream: share={}, path={:?}",
1147+
self.share_name, smb_path
1148+
);
1149+
1150+
// Collect all chunks into a buffer, then write in one pipelined call
1151+
let mut data = Vec::new();
1152+
while let Some(result) = stream.next_chunk() {
1153+
data.extend_from_slice(&result?);
1154+
}
1155+
1156+
let len = data.len() as u64;
1157+
let sp = smb_path;
1158+
self.with_smb("write_from_stream", |client, tree| {
1159+
handle.block_on(client.write_file_pipelined(tree, &sp, &data))
1160+
})?;
1161+
1162+
Ok(len)
1163+
}
1164+
10791165
fn smb_connection_state(&self) -> Option<SmbConnectionState> {
10801166
match self.connection_state() {
10811167
ConnectionState::Direct => Some(SmbConnectionState::Direct),
@@ -1799,4 +1885,81 @@ mod tests {
17991885
assert!(space.available_bytes > 0);
18001886
assert!(space.used_bytes <= space.total_bytes);
18011887
}
1888+
1889+
// ── SmbReadStream tests ────────────────────────────────────────
1890+
1891+
#[test]
1892+
fn smb_read_stream_empty_file() {
1893+
let mut stream = SmbReadStream {
1894+
data: vec![],
1895+
offset: 0,
1896+
};
1897+
assert_eq!(stream.total_size(), 0);
1898+
assert_eq!(stream.bytes_read(), 0);
1899+
assert!(stream.next_chunk().is_none());
1900+
}
1901+
1902+
#[test]
1903+
fn smb_read_stream_small_file_single_chunk() {
1904+
let data = vec![1u8; 100];
1905+
let mut stream = SmbReadStream { data, offset: 0 };
1906+
assert_eq!(stream.total_size(), 100);
1907+
1908+
let chunk = stream.next_chunk().unwrap().unwrap();
1909+
assert_eq!(chunk.len(), 100);
1910+
assert_eq!(stream.bytes_read(), 100);
1911+
assert!(stream.next_chunk().is_none());
1912+
}
1913+
1914+
#[test]
1915+
fn smb_read_stream_exact_chunk_boundary() {
1916+
let data = vec![0u8; SMB_STREAM_CHUNK_SIZE];
1917+
let mut stream = SmbReadStream { data, offset: 0 };
1918+
1919+
let chunk = stream.next_chunk().unwrap().unwrap();
1920+
assert_eq!(chunk.len(), SMB_STREAM_CHUNK_SIZE);
1921+
assert!(stream.next_chunk().is_none());
1922+
}
1923+
1924+
#[test]
1925+
fn smb_read_stream_multiple_chunks() {
1926+
let size = SMB_STREAM_CHUNK_SIZE * 2 + 500;
1927+
let data = vec![0xAB; size];
1928+
let mut stream = SmbReadStream { data, offset: 0 };
1929+
assert_eq!(stream.total_size(), size as u64);
1930+
1931+
let c1 = stream.next_chunk().unwrap().unwrap();
1932+
assert_eq!(c1.len(), SMB_STREAM_CHUNK_SIZE);
1933+
assert_eq!(stream.bytes_read(), SMB_STREAM_CHUNK_SIZE as u64);
1934+
1935+
let c2 = stream.next_chunk().unwrap().unwrap();
1936+
assert_eq!(c2.len(), SMB_STREAM_CHUNK_SIZE);
1937+
1938+
let c3 = stream.next_chunk().unwrap().unwrap();
1939+
assert_eq!(c3.len(), 500);
1940+
assert_eq!(stream.bytes_read(), size as u64);
1941+
1942+
assert!(stream.next_chunk().is_none());
1943+
}
1944+
1945+
#[test]
1946+
fn smb_read_stream_data_integrity() {
1947+
let data: Vec<u8> = (0..=255).cycle().take(SMB_STREAM_CHUNK_SIZE + 100).collect();
1948+
let expected = data.clone();
1949+
let mut stream = SmbReadStream { data, offset: 0 };
1950+
1951+
let mut reassembled = Vec::new();
1952+
while let Some(Ok(chunk)) = stream.next_chunk() {
1953+
reassembled.extend_from_slice(&chunk);
1954+
}
1955+
assert_eq!(reassembled, expected);
1956+
}
1957+
1958+
#[test]
1959+
fn smb_supports_streaming() {
1960+
// SmbVolume should report streaming support so cross-volume copies
1961+
// (MTP↔SMB) use the streaming path instead of NotSupported/temp files.
1962+
let (vol, _rt) = make_test_volume();
1963+
assert!(vol.supports_streaming());
1964+
}
18021965
}

‎apps/desktop/src-tauri/src/file_system/write_operations/volume_strategy.rs‎

Lines changed: 7 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -80,8 +80,13 @@ pub(super) fn copy_single_path(
8080
return dest_volume.write_from_stream(dest_path, size, stream);
8181
}
8282

83-
// Neither supports streaming and it's not a directory - not supported
84-
return Err(VolumeError::NotSupported);
83+
// Neither supports streaming — fall back to temp local (export then import)
84+
log::debug!(
85+
"copy_single_path: no streaming support, using temp local for {} -> {}",
86+
source_path.display(),
87+
dest_path.display()
88+
);
89+
return copy_via_temp_local(source_volume, source_path, dest_volume, dest_path);
8590
}
8691

8792
if source_is_local && !dest_is_local {

‎apps/desktop/src-tauri/src/mtp/connection/mod.rs‎

Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -447,7 +447,10 @@ impl MtpConnectionManager {
447447
Some(entry) => {
448448
// Skip if we already know about this storage (duplicate event)
449449
if entry.storages.iter().any(|s| s.id == storage_id) {
450-
debug!("handle_storage_added: storage {} already registered for {}", storage_id, device_id);
450+
debug!(
451+
"handle_storage_added: storage {} already registered for {}",
452+
storage_id, device_id
453+
);
451454
return;
452455
}
453456
entry.device.clone()

‎apps/desktop/src-tauri/src/network/smb_upgrade.rs‎

Lines changed: 1 addition & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -229,12 +229,7 @@ pub(crate) async fn get_keychain_password(
229229
}
230230
}
231231

232-
log::debug!(
233-
"No Keychain credentials for {:?} / {} / {}",
234-
hostname,
235-
server_ip,
236-
share
237-
);
232+
log::debug!("No Keychain credentials for {:?} / {} / {}", hostname, server_ip, share);
238233
None
239234
})
240235
.await

0 commit comments

Comments
 (0)