From 9154adfb0034e5132cff7f6fea7acb934224ce70 Mon Sep 17 00:00:00 2001 From: techartdev Date: Sat, 26 Sep 2026 20:34:48 +0300 Subject: [PATCH] Confirm remote replacements and recursive folder deletion --- crates/service-contracts/src/copy.rs | 21 +- crates/service-contracts/src/lib.rs | 17 ++ crates/ssh-core/src/transfers.rs | 112 ++++++++- src-tauri/src/lib.rs | 1 + src-tauri/src/transfer_folder_tests.rs | 74 ++++++ src-tauri/src/transfer_tests.rs | 102 ++++++++- src-tauri/src/transfer_tree.rs | 51 ++++- src-tauri/src/transfers.rs | 264 +++++++++++++++++++--- src-tauri/src/workspace_services.rs | 30 +++ src/apps/Files.tsx | 103 +++++---- src/components/DeleteFilesDialog.tsx | 51 +++-- src/components/FileActionDialog.css | 15 ++ src/components/TransferConflictDialog.tsx | 74 ++++++ src/components/TransferPanel.tsx | 1 + src/file-delete.test.ts | 111 +++++++++ src/file-delete.ts | 101 ++++++++- src/sdk.ts | 15 ++ src/services.ts | 5 +- src/session-services.ts | 39 +++- src/transfer-queue.test.ts | 119 +++++++++- src/transfer-queue.ts | 104 ++++++++- 21 files changed, 1285 insertions(+), 125 deletions(-) create mode 100644 src/components/TransferConflictDialog.tsx diff --git a/crates/service-contracts/src/copy.rs b/crates/service-contracts/src/copy.rs index ae1136e..897b3ac 100644 --- a/crates/service-contracts/src/copy.rs +++ b/crates/service-contracts/src/copy.rs @@ -21,6 +21,22 @@ pub async fn copy_regular_file( name: &str, canceled: impl Fn() -> bool + Send, progress: &mut (dyn FnMut(CopyProgress) + Send), +) -> Result { + copy_regular_file_with_replace( + service, path, revision, parent, name, None, canceled, progress, + ) + .await +} + +pub async fn copy_regular_file_with_replace( + service: Arc, + path: &str, + revision: &str, + parent: &str, + name: &str, + replace_revision: Option<&str>, + canceled: impl Fn() -> bool + Send, + progress: &mut (dyn FnMut(CopyProgress) + Send), ) -> Result { let checkpoint = || -> Result<()> { if canceled() { @@ -33,7 +49,10 @@ pub async fn copy_regular_file( let total = reader.file().size; let writer = async { checkpoint()?; - service.upload(parent, name, total).await + match replace_revision { + Some(expected) => service.upload_replace(parent, name, total, expected).await, + None => service.upload(parent, name, total).await, + } } .await; let mut writer = match writer { diff --git a/crates/service-contracts/src/lib.rs b/crates/service-contracts/src/lib.rs index c5041e1..94f05eb 100644 --- a/crates/service-contracts/src/lib.rs +++ b/crates/service-contracts/src/lib.rs @@ -58,6 +58,23 @@ pub trait TransferWriter: Send { } #[async_trait] pub trait FileTransferService: Send + Sync { + /// Optional destination inspection for explicit conflict review. + async fn destination_entry(&self, _parent: &str, _name: &str) -> Result> { + Ok(None) + } + /// A provider may offer revision-checked, atomic replacement of regular files. + fn supports_atomic_replace(&self) -> bool { + false + } + async fn upload_replace( + self: Arc, + _parent: &str, + _name: &str, + _size: u64, + _expected_revision: &str, + ) -> Result> { + anyhow::bail!("Safe file replacement is unavailable on this host") + } /// Inspect one selected entry without materializing its parent directory. async fn transfer_entry(self: Arc, path: &str, revision: &str) -> Result { let mut reader = self.download(path, revision).await?; diff --git a/crates/ssh-core/src/transfers.rs b/crates/ssh-core/src/transfers.rs index a8bf60b..b72f18c 100644 --- a/crates/ssh-core/src/transfers.rs +++ b/crates/ssh-core/src/transfers.rs @@ -1,6 +1,6 @@ // SPDX-License-Identifier: MPL-2.0 use crate::{ - entry_revision, provider::sftp_location, text::validate_path, FileLocation, + entry_revision, provider::sftp_location, text::validate_path, FileEntry, FileLocation, FileTransferService, SftpTextFiles, TransferFile, TransferReader, TransferWriter, TRANSFER_CHUNK, }; @@ -100,6 +100,7 @@ struct Writer { handle: Option, temporary: Option, destination: String, + expected_revision: Option, size: u64, offset: u64, } @@ -157,6 +158,29 @@ impl TransferWriter for Writer { if self.offset != self.size { bail!("The local file changed size during upload."); } + let _lock = self.service.save_lock.lock().await; + if let Some(expected) = &self.expected_revision { + let existing = self + .service + .checked_entry(&self.destination, expected) + .await?; + if !regular_file(&existing) { + bail!("Only an existing regular file can be replaced."); + } + self.service + .raw + .fsetstat( + self.handle.as_ref().context("Upload handle is closed")?, + FileAttributes { + uid: existing.uid, + gid: existing.gid, + permissions: existing.permissions, + ..FileAttributes::empty() + }, + ) + .await + .context("Cannot preserve the original file owner and permissions")?; + } if self.service.fsync { self.service .extension( @@ -168,9 +192,24 @@ impl TransferWriter for Writer { if let Some(handle) = self.handle.take() { self.service.raw.close(handle).await?; } - let _lock = self.service.save_lock.lock().await; - self.service.require_absent(&self.destination).await?; - self.service.raw.rename(self.temporary.as_ref().context("Upload already finished")?, &self.destination).await.context("Upload publication was not confirmed. Check the remote destination before retrying; it may already exist")?; + if let Some(expected) = &self.expected_revision { + self.service + .checked_entry(&self.destination, expected) + .await?; + self.service + .extension( + "posix-rename@openssh.com", + &[ + self.temporary.as_ref().context("Upload already finished")?, + &self.destination, + ], + ) + .await + .context("Replacement was not confirmed. Check the remote file before retrying")?; + } else { + self.service.require_absent(&self.destination).await?; + self.service.raw.rename(self.temporary.as_ref().context("Upload already finished")?, &self.destination).await.context("Upload publication was not confirmed. Check the remote destination before retrying; it may already exist")?; + } self.temporary = None; Ok(sftp_location(self.destination.clone())) } @@ -198,6 +237,49 @@ impl TransferWriter for Writer { #[async_trait] impl FileTransferService for SftpTextFiles { + async fn destination_entry(&self, parent: &str, name: &str) -> Result> { + let path = self.child_path(parent, name).await?; + let attrs = match self.raw.lstat(&path).await { + Ok(result) => result.attrs, + Err(SftpError::Status(status)) if status.status_code == StatusCode::NoSuchFile => { + return Ok(None) + } + Err(error) => return Err(error.into()), + }; + Ok(Some(FileEntry { + name: name.into(), + path, + kind: if attrs.is_dir() { + "directory" + } else if attrs.is_symlink() { + "symlink" + } else if regular_file(&attrs) { + "file" + } else { + "special" + } + .into(), + size: attrs.size.unwrap_or(0), + modified: attrs.mtime, + revision: entry_revision(&attrs), + })) + } + fn supports_atomic_replace(&self) -> bool { + self.can_replace_atomically() && !self.windows + } + async fn upload_replace( + self: Arc, + parent: &str, + name: &str, + size: u64, + expected_revision: &str, + ) -> Result> { + if !self.supports_atomic_replace() { + bail!("Safe file replacement is unavailable on this host"); + } + self.prepare_upload(parent, name, size, Some(expected_revision)) + .await + } async fn transfer_entry( self: Arc, path: &str, @@ -304,9 +386,28 @@ impl FileTransferService for SftpTextFiles { parent: &str, name: &str, size: u64, + ) -> Result> { + self.prepare_upload(parent, name, size, None).await + } +} + +impl SftpTextFiles { + async fn prepare_upload( + self: Arc, + parent: &str, + name: &str, + size: u64, + expected_revision: Option<&str>, ) -> Result> { let destination = self.child_path(parent, name).await?; - self.require_absent(&destination).await?; + if let Some(expected) = expected_revision { + let existing = self.checked_entry(&destination, expected).await?; + if !regular_file(&existing) { + bail!("Only an existing regular file can be replaced."); + } + } else { + self.require_absent(&destination).await?; + } let canonical_parent = sftp_location(destination.clone()) .parent .context("Upload destination has no parent")?; @@ -324,6 +425,7 @@ impl FileTransferService for SftpTextFiles { handle: Some(handle), temporary: Some(temporary), destination, + expected_revision: expected_revision.map(str::to_owned), size, offset: 0, })) diff --git a/src-tauri/src/lib.rs b/src-tauri/src/lib.rs index 3446ac0..672c3d7 100644 --- a/src-tauri/src/lib.rs +++ b/src-tauri/src/lib.rs @@ -1022,6 +1022,7 @@ pub fn run() { transfers::prepare_file_copy_selection, transfers::system_clipboard_sequence, transfers::prepare_file_copy, + transfers::transfer_conflicts, transfers::run_transfer, transfers::cancel_transfer, open_terminal, diff --git a/src-tauri/src/transfer_folder_tests.rs b/src-tauri/src/transfer_folder_tests.rs index 79dcb79..4147b81 100644 --- a/src-tauri/src/transfer_folder_tests.rs +++ b/src-tauri/src/transfer_folder_tests.rs @@ -19,6 +19,7 @@ struct Folders { data: Arc, created: Mutex>, bad_child: Option, + merge_destination: bool, } struct Page(Vec); #[async_trait] @@ -45,6 +46,31 @@ async fn scanned( } #[async_trait] impl FileTransferService for Folders { + async fn destination_entry(&self, parent: &str, name: &str) -> Result> { + if !self.merge_destination { + return Ok(None); + } + Ok(match (parent, name) { + ("destination@opaque", "Root") => Some(entry("existing-root@opaque", "Root", true, 0)), + ("existing-root@opaque", "binary.bin") => { + Some(entry("existing-file@opaque", "binary.bin", false, 3)) + } + _ => None, + }) + } + fn supports_atomic_replace(&self) -> bool { + self.merge_destination + } + async fn upload_replace( + self: Arc, + parent: &str, + name: &str, + size: u64, + expected_revision: &str, + ) -> Result> { + assert_eq!(expected_revision, "v1"); + self.data.clone().upload(parent, name, size).await + } fn supports_folders(&self) -> bool { true } @@ -103,8 +129,56 @@ fn provider(bad_child: Option) -> Arc { data: super::tests::Memory::new(false), created: Mutex::new(HashSet::new()), bad_child, + merge_destination: false, }) } +#[tokio::test] +async fn reviewed_folder_conflict_merges_and_replaces_only_matching_file() { + let memory = super::tests::Memory::new(false); + let service: Arc = Arc::new(Folders { + data: memory.clone(), + created: Mutex::new(HashSet::new()), + bad_child: None, + merge_destination: true, + }); + let (_, cancel) = watch::channel(false); + let plan = || Job::Tree { + tree: tree::Tree { + root: catalog::Node { + id: 0, + parent: 0, + display: "Root".into(), + entry: entry("root@opaque", "Root", true, 0), + local: None, + }, + }, + target: tree::Target::Remote("destination@opaque".into()), + }; + assert!(execute_with_policy( + plan(), + service.clone(), + &cancel, + &mut |_| {}, + &TransferPolicy::default() + ) + .await + .unwrap_err() + .to_string() + .contains("already exists")); + assert_eq!(memory.commits.load(std::sync::atomic::Ordering::SeqCst), 0); + let policy = TransferPolicy { + replace: vec![ReviewedConflict { + path: "existing-root@opaque".into(), + revision: "v1".into(), + }], + skip: Vec::new(), + }; + execute_with_policy(plan(), service, &cancel, &mut |_| {}, &policy) + .await + .unwrap(); + assert_eq!(memory.commits.load(std::sync::atomic::Ordering::SeqCst), 1); + assert_eq!(*memory.writes.lock().unwrap(), memory.data); +} fn download_plan(plan: tree::Tree, folder: &Path) -> Job { let catalog = Arc::new(catalog::Catalog::new(true).unwrap()); catalog diff --git a/src-tauri/src/transfer_tests.rs b/src-tauri/src/transfer_tests.rs index 5247154..a0404c5 100644 --- a/src-tauri/src/transfer_tests.rs +++ b/src-tauri/src/transfer_tests.rs @@ -193,9 +193,9 @@ use std::sync::{ }; pub(super) struct Memory { pub(super) data: Vec, - writes: Mutex>, + pub(super) writes: Mutex>, aborts: AtomicUsize, - commits: AtomicUsize, + pub(super) commits: AtomicUsize, fail_verify: bool, readers: AtomicUsize, max_readers: AtomicUsize, @@ -661,6 +661,104 @@ fn upload_job(path: &Path) -> Job { name: "binary.bin".into(), } } +struct ExistingDestination(Arc); +#[async_trait::async_trait] +impl FileTransferService for ExistingDestination { + async fn destination_entry( + &self, + _: &str, + _: &str, + ) -> Result> { + Ok(Some(shellcanvas_services::FileEntry { + name: "binary.bin".into(), + path: "existing@1".into(), + kind: "file".into(), + size: 3, + modified: None, + revision: "old-revision".into(), + })) + } + fn supports_atomic_replace(&self) -> bool { + true + } + async fn upload_replace( + self: Arc, + parent: &str, + name: &str, + size: u64, + expected_revision: &str, + ) -> Result> { + assert_eq!(expected_revision, "old-revision"); + self.0.clone().upload(parent, name, size).await + } + async fn upload( + self: Arc, + _: &str, + _: &str, + _: u64, + ) -> Result> { + bail!("Unreviewed upload must not start") + } + async fn download( + self: Arc, + path: &str, + revision: &str, + ) -> Result> { + self.0.clone().download(path, revision).await + } +} +#[tokio::test] +async fn existing_upload_needs_matching_review_before_replacement() { + let dir = tempfile::tempdir().unwrap(); + let input = dir.path().join("input.bin"); + let memory = Memory::new(false); + std::fs::write(&input, &memory.data).unwrap(); + let service: Arc = Arc::new(ExistingDestination(memory.clone())); + let (_, cancel) = watch::channel(false); + assert!(execute_with_policy( + upload_job(&input), + service.clone(), + &cancel, + &mut |_| {}, + &TransferPolicy::default(), + ) + .await + .unwrap_err() + .to_string() + .contains("already exists")); + assert_eq!(memory.commits.load(Ordering::SeqCst), 0); + let stale = TransferPolicy { + replace: vec![ReviewedConflict { + path: "existing@1".into(), + revision: "stale".into(), + }], + skip: vec![], + }; + assert!(execute_with_policy( + upload_job(&input), + service.clone(), + &cancel, + &mut |_| {}, + &stale, + ) + .await + .unwrap_err() + .to_string() + .contains("changed")); + assert_eq!(memory.commits.load(Ordering::SeqCst), 0); + let accepted = TransferPolicy { + replace: vec![ReviewedConflict { + path: "existing@1".into(), + revision: "old-revision".into(), + }], + skip: vec![], + }; + execute_with_policy(upload_job(&input), service, &cancel, &mut |_| {}, &accepted) + .await + .unwrap(); + assert_eq!(memory.commits.load(Ordering::SeqCst), 1); + assert_eq!(*memory.writes.lock().unwrap(), memory.data); +} #[tokio::test] async fn streams_binary_files_and_refuses_local_overwrite() { let dir = tempfile::tempdir().unwrap(); diff --git a/src-tauri/src/transfer_tree.rs b/src-tauri/src/transfer_tree.rs index 749ece4..40aab80 100644 --- a/src-tauri/src/transfer_tree.rs +++ b/src-tauri/src/transfer_tree.rs @@ -261,6 +261,7 @@ pub(super) async fn execute_tree( service: Arc, cancel: &watch::Receiver, progress: &mut (dyn FnMut(Progress) + Send), + policy: &TransferPolicy, ) -> Result { let catalog = scan( tree, @@ -270,7 +271,7 @@ pub(super) async fn execute_tree( progress, ) .await?; - execute_catalog(catalog, target, service, cancel, progress, None).await + execute_catalog(catalog, target, service, cancel, progress, None, policy).await } pub(super) async fn execute_selection( @@ -279,6 +280,7 @@ pub(super) async fn execute_selection( service: Arc, cancel: &watch::Receiver, progress: &mut (dyn FnMut(Progress) + Send), + policy: &TransferPolicy, ) -> Result { if catalog.has_directories()? && !service.supports_folders() { bail!("Folder transfers are unavailable on this device"); @@ -292,6 +294,7 @@ pub(super) async fn execute_selection( cancel, progress, destination, + policy, ) .await } @@ -348,6 +351,7 @@ pub(super) async fn execute_download_selection( cancel, progress, destination, + &TransferPolicy::default(), ) .await } @@ -359,6 +363,7 @@ async fn execute_catalog( cancel: &watch::Receiver, progress: &mut (dyn FnMut(Progress) + Send), selection_destination: Option, + policy: &TransferPolicy, ) -> Result { if let Target::Remote(parent) = &target { if catalog.has_directories()? && !service.supports_folders() { @@ -370,11 +375,17 @@ async fn execute_catalog( } let mut root = None; let mut completed = 0; + let mut approved = std::collections::HashSet::new(); + let mut skipped = std::collections::HashSet::new(); let total = catalog.size(); let work: Result = async { for id in 1..=catalog.len() { checkpoint(cancel)?; let node = catalog.get(id)?; + if skipped.contains(&node.parent) { + skipped.insert(id); + continue; + } let parent = if node.parent == 0 { match &target { Target::Remote(parent) => parent.clone(), @@ -391,10 +402,38 @@ async fn execute_catalog( } Target::Remote(_) => None, }; + let existing = if destination.is_none() { + service.destination_entry(&parent, &node.entry.name).await? + } else { + None + }; + if let Some(existing) = &existing { + if node.parent == 0 && policy.skip(existing)? { + skipped.insert(id); + continue; + } + if node.parent == 0 { + if policy.replacement(existing)?.is_none() { + bail!("An item already exists at this destination. Nothing was replaced."); + } + approved.insert(id); + } else if !approved.contains(&node.parent) { + bail!("A new destination conflict appeared during transfer. Nothing was replaced."); + } + if existing.kind != node.entry.kind || + (existing.kind == "file" && !service.supports_atomic_replace()) { + bail!("Safe replacement is unavailable for this destination item"); + } + } + if node.parent != 0 && approved.contains(&node.parent) { + approved.insert(id); + } let path = if node.entry.kind == "directory" { if let Some(destination) = destination { std::fs::create_dir(&destination)?; destination.canonicalize()?.to_string_lossy().into_owned() + } else if let Some(existing) = &existing { + existing.path.clone() } else { service .transfer_mkdir(&parent, &node.entry.name) @@ -426,14 +465,20 @@ async fn execute_catalog( name: node.entry.name.clone(), } }; - let path = Box::pin(super::execute(job, service.clone(), cancel, &mut |event| { + let node_policy = TransferPolicy { + replace: existing.iter().map(|entry| ReviewedConflict { + path: entry.path.clone(), revision: entry.revision.clone(), + }).collect(), + skip: Vec::new(), + }; + let path = Box::pin(super::execute_with_policy(job, service.clone(), cancel, &mut |event| { progress(Progress { items: None, bytes: completed + event.bytes, total, phase: "running", }) - })) + }, &node_policy)) .await?; completed += node.entry.size; path diff --git a/src-tauri/src/transfers.rs b/src-tauri/src/transfers.rs index 09b10fc..f3525e2 100644 --- a/src-tauri/src/transfers.rs +++ b/src-tauri/src/transfers.rs @@ -235,6 +235,53 @@ pub struct Ticket { } #[derive(Clone, Serialize)] #[serde(rename_all = "camelCase")] +pub struct TransferConflict { + pub destination: shellcanvas_services::FileEntry, + pub source_kind: String, +} +#[derive(Serialize)] +#[serde(rename_all = "camelCase")] +pub struct TransferConflictReview { + pub conflicts: Vec, + pub can_replace: bool, +} +#[derive(Clone, serde::Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct ReviewedConflict { + path: String, + revision: String, +} +#[derive(Clone, Default, serde::Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct TransferPolicy { + replace: Vec, + skip: Vec, +} +impl TransferPolicy { + fn replacement<'a>( + &'a self, + entry: &shellcanvas_services::FileEntry, + ) -> Result> { + let reviewed = self.replace.iter().find(|item| item.path == entry.path); + if let Some(reviewed) = reviewed { + if reviewed.revision != entry.revision { + bail!("The destination changed after replacement was confirmed. Refresh and try again."); + } + } + Ok(reviewed.map(|item| item.revision.as_str())) + } + fn skip(&self, entry: &shellcanvas_services::FileEntry) -> Result { + let reviewed = self.skip.iter().find(|item| item.path == entry.path); + if let Some(reviewed) = reviewed { + if reviewed.revision != entry.revision { + bail!("The destination changed after it was reviewed. Refresh and try again."); + } + } + Ok(reviewed.is_some()) + } +} +#[derive(Clone, Serialize)] +#[serde(rename_all = "camelCase")] pub struct Progress { pub bytes: u64, pub total: u64, @@ -285,6 +332,32 @@ enum Job { revision: String, }, } +impl Job { + fn destination_roots(&self) -> Result)>> { + match self { + Job::Selection { catalog, parent } => { + let mut roots = Vec::new(); + let mut after = 0; + while let Some(root) = catalog.root_after(after)? { + after = root.id; + roots.push((root.entry.name, root.entry.kind)); + } + Ok(Some((parent.clone(), roots))) + } + Job::Tree { + tree, + target: tree::Target::Remote(parent), + } => Ok(Some(( + parent.clone(), + vec![(tree.root.entry.name.clone(), tree.root.entry.kind.clone())], + ))), + Job::Copy { parent, name, .. } | Job::Upload { parent, name, .. } => { + Ok(Some((parent.clone(), vec![(name.clone(), "file".into())]))) + } + _ => Ok(None), + } + } +} enum Work { Transfer { job: Box, @@ -320,6 +393,24 @@ impl Default for TransferRegistry { } } impl TransferRegistry { + fn conflict_candidates( + &self, + owner: u64, + id: u64, + ) -> Result, String, Vec<(String, String)>)>, String> { + let pending = self + .jobs + .get(&id) + .filter(|pending| pending.owner == owner) + .ok_or("Transfer is closed or belongs to another host")?; + match pending.work.as_ref() { + Some(Work::Transfer { job, service }) => Ok(job + .destination_roots() + .map_err(error)? + .map(|(parent, roots)| (service.clone(), parent, roots))), + _ => Ok(None), + } + } pub fn has_pending(&self) -> bool { !self.jobs.is_empty() || !self.preparations.is_empty() } @@ -1162,11 +1253,37 @@ fn cleanup_error(original: anyhow::Error, cleanup: Result<()>) -> anyhow::Error Err(error) => anyhow::anyhow!("{original:#}; cleanup failed: {error:#}"), } } -async fn execute( +enum DestinationAction { + New, + Replace(String), + Skip(String), +} +async fn destination_action( + service: &Arc, + parent: &str, + name: &str, + policy: &TransferPolicy, +) -> Result { + let Some(existing) = service.destination_entry(parent, name).await? else { + return Ok(DestinationAction::New); + }; + if policy.skip(&existing)? { + return Ok(DestinationAction::Skip(existing.path)); + } + if let Some(revision) = policy.replacement(&existing)? { + if existing.kind != "file" || !service.supports_atomic_replace() { + bail!("Safe replacement is unavailable for this destination item"); + } + return Ok(DestinationAction::Replace(revision.into())); + } + bail!("An item already exists at this destination. Nothing was replaced.") +} +async fn execute_with_policy( job: Job, service: Arc, cancel: &watch::Receiver, progress: &mut (dyn FnMut(Progress) + Send), + policy: &TransferPolicy, ) -> Result { checkpoint(cancel)?; match job { @@ -1174,38 +1291,48 @@ async fn execute( tree::execute_download_selection(catalog, folder, service, cancel, progress).await } Job::Selection { catalog, parent } => { - tree::execute_selection(catalog, parent, service, cancel, progress).await + tree::execute_selection(catalog, parent, service, cancel, progress, policy).await } Job::Tree { tree, target } => { - tree::execute_tree(tree, target, service, cancel, progress).await + tree::execute_tree(tree, target, service, cancel, progress, policy).await } Job::Copy { path, revision, parent, name, - } => shellcanvas_services::copy_regular_file( - service, - &path, - &revision, - &parent, - &name, - || *cancel.borrow(), - &mut |event| { - progress(Progress { - items: None, - bytes: event.bytes, - total: event.total, - phase: if event.finishing { - "finishing" - } else { - "running" - }, - }); - }, - ) - .await - .map(|location| location.path), + } => { + let action = destination_action(&service, &parent, &name, policy).await?; + if let DestinationAction::Skip(path) = action { + return Ok(path); + } + shellcanvas_services::copy_regular_file_with_replace( + service, + &path, + &revision, + &parent, + &name, + match &action { + DestinationAction::Replace(revision) => Some(revision.as_str()), + _ => None, + }, + || *cancel.borrow(), + &mut |event| { + progress(Progress { + items: None, + bytes: event.bytes, + total: event.total, + phase: if event.finishing { + "finishing" + } else { + "running" + }, + }); + }, + ) + .await + .map(|location| location.path) + } Job::Upload { file, parent, @@ -1213,12 +1340,24 @@ async fn execute( size, modified, } => { + let action = destination_action(&service, &parent, &name, policy).await?; + if let DestinationAction::Skip(path) = action { + return Ok(path); + } let mut file = tokio::fs::File::from_std(file); let current = file.metadata().await?; if current.len() != size || current.modified().ok() != modified { bail!("The local file changed after selection. Choose it again."); } - let mut remote = service.upload(&parent, &name, size).await?; + let mut remote = match action { + DestinationAction::Replace(revision) => { + service + .clone() + .upload_replace(&parent, &name, size, &revision) + .await? + } + _ => service.clone().upload(&parent, &name, size).await?, + }; let work = async { let mut bytes = vec![0; TRANSFER_CHUNK]; let mut offset = 0; @@ -1333,12 +1472,58 @@ async fn execute( } } } +#[cfg(test)] +async fn execute( + job: Job, + service: Arc, + cancel: &watch::Receiver, + progress: &mut (dyn FnMut(Progress) + Send), +) -> Result { + execute_with_policy(job, service, cancel, progress, &TransferPolicy::default()).await +} +#[tauri::command] +pub async fn transfer_conflicts( + session_id: u64, + transfer_id: u64, + state: State<'_, DesktopState>, +) -> Result { + let candidates = state + .transfers + .lock() + .await + .conflict_candidates(session_id, transfer_id)?; + let Some((service, parent, roots)) = candidates else { + return Ok(TransferConflictReview { + conflicts: Vec::new(), + can_replace: false, + }); + }; + let mut conflicts = Vec::new(); + for (name, source_kind) in roots { + if let Some(destination) = service + .destination_entry(&parent, &name) + .await + .map_err(error)? + { + conflicts.push(TransferConflict { + destination, + source_kind, + }); + } + } + Ok(TransferConflictReview { + conflicts, + can_replace: service.supports_atomic_replace(), + }) +} + #[tauri::command] pub async fn run_transfer( session_id: u64, transfer_id: u64, on_event: Channel, tracked: Option>, + policy: Option, state: State<'_, DesktopState>, ) -> Result { let _update_operation = crate::update_gate::operation()?; @@ -1360,6 +1545,7 @@ pub async fn run_transfer( phase: "preparing", }; let mut sent = Instant::now(); + let policy = policy.unwrap_or_default(); let result = async { checkpoint(&cancel)?; let _permit = tokio::select! { @@ -1372,16 +1558,22 @@ pub async fn run_transfer( let relocation = claim.run(&cancel, &tracked.unwrap_or_default()).await?; Ok((relocation.path.clone(), Some(relocation))) } - Work::Transfer { job, service } => execute(*job, service, &cancel, &mut |event| { - if event.phase != last.phase - || sent.elapsed().as_millis() >= 100 - || event.bytes == event.total - { - let _ = on_event.send(event.clone()); - sent = Instant::now(); - } - last = event; - }) + Work::Transfer { job, service } => execute_with_policy( + *job, + service, + &cancel, + &mut |event| { + if event.phase != last.phase + || sent.elapsed().as_millis() >= 100 + || event.bytes == event.total + { + let _ = on_event.send(event.clone()); + sent = Instant::now(); + } + last = event; + }, + &policy, + ) .await .map(|path| (path, None)), } diff --git a/src-tauri/src/workspace_services.rs b/src-tauri/src/workspace_services.rs index 2a58cba..69311a1 100644 --- a/src-tauri/src/workspace_services.rs +++ b/src-tauri/src/workspace_services.rs @@ -828,6 +828,36 @@ impl TransferWriter for Upload { } #[async_trait] impl FileTransferService for Bound { + async fn destination_entry(&self, parent: &str, name: &str) -> Result> { + self.binding + .run(false, self.service.destination_entry(parent, name)) + .await + } + fn supports_atomic_replace(&self) -> bool { + self.service.supports_atomic_replace() + } + async fn upload_replace( + self: Arc, + parent: &str, + name: &str, + size: u64, + expected_revision: &str, + ) -> Result> { + self.binding.check()?; + let mut inner = self + .service + .clone() + .upload_replace(parent, name, size, expected_revision) + .await?; + if let Err(error) = self.binding.after(true) { + let _ = tokio::time::timeout(Duration::from_secs(3), inner.abort()).await; + return Err(error); + } + Ok(Box::new(Upload { + binding: self.binding.clone(), + inner, + })) + } async fn transfer_entry(self: Arc, path: &str, revision: &str) -> Result { self.binding .run(false, self.service.clone().transfer_entry(path, revision)) diff --git a/src/apps/Files.tsx b/src/apps/Files.tsx index 1d57db1..7f8d541 100644 --- a/src/apps/Files.tsx +++ b/src/apps/Files.tsx @@ -37,6 +37,7 @@ import type { ClipboardPreparation, Directory, FileEntry, + TransferConflictReview, } from "../sdk"; import { ContextMenu, type MenuAction } from "../components/ContextMenu"; import { clipboard } from "../clipboard"; @@ -56,8 +57,13 @@ import { AttachDriveDialog } from "../components/AttachDriveDialog"; import { MoveFileDialog } from "../components/MoveFileDialog"; import { watchFileChanges, watchFileLocations } from "../file-events"; import { relocateNavigation, trackedNavigation } from "../file-navigation"; -import { TransferQueue, pendingTransfer } from "../transfer-queue"; +import { + TransferQueue, + pendingTransfer, + type ConflictDecision, +} from "../transfer-queue"; import { TransferPanel } from "../components/TransferPanel"; +import { TransferConflictDialog } from "../components/TransferConflictDialog"; import { selectFiles } from "../file-selection"; import { DeleteFilesDialog } from "../components/DeleteFilesDialog"; import { fileSourceKey } from "../workspace-bindings"; @@ -127,6 +133,7 @@ export function Files({ const selectionAnchor = useRef(null); const [deleteTarget, setDeleteTarget] = useState<{ entries: Readonly[]; + parent: string; services: AppContext["services"]; sessionId: number; } | null>(null); @@ -136,8 +143,37 @@ export function Files({ const relocatingRef = useRef(false); const loading = reading || relocating; const [busy, setBusy] = useState(false); + const conflictSequence = useRef(0); + const [conflictPrompt, setConflictPrompt] = useState<{ + key: number; + conflict: TransferConflictReview["conflicts"][number]; + remaining: number; + decide(decision: ConflictDecision): void; + } | null>(null); const queue = useMemo( - () => new TransferQueue(services, reportError), + () => + new TransferQueue( + services, + reportError, + (conflict, remaining, signal) => + new Promise((resolve) => { + if (signal.aborted) return resolve("cancel"); + const key = ++conflictSequence.current; + let settled = false; + const decide = (decision: ConflictDecision) => { + if (settled) return; + settled = true; + signal.removeEventListener("abort", abort); + setConflictPrompt((current) => + current?.key === key ? null : current, + ); + resolve(decision); + }; + const abort = () => decide("cancel"); + signal.addEventListener("abort", abort, { once: true }); + setConflictPrompt({ key, conflict, remaining, decide }); + }), + ), [services], ); const transfers = useSyncExternalStore(queue.subscribe, queue.snapshot); @@ -295,7 +331,7 @@ export function Files({ } } const [operation, setOperation] = useState<{ - kind: "mkdir" | "rename" | "delete"; + kind: "mkdir" | "rename"; parent: string; entry?: FileEntry; } | null>(null); @@ -976,6 +1012,7 @@ export function Files({ } setDeleteTarget({ entries: items.map((entry) => Object.freeze({ ...entry })), + parent: directory.path, services, sessionId: session.id, }); @@ -1225,12 +1262,10 @@ export function Files({ }, { id: "delete", - label: - entry.kind === "directory" ? "Delete empty folder…" : "Delete…", + label: entry.kind === "directory" ? "Delete folder…" : "Delete…", shortcut: "Delete", disabled: !canManage || !entry.revision, - run: () => - setOperation({ kind: "delete", parent: directory.path, entry }), + run: () => reviewDelete([entry]), }, ] : [ @@ -1456,11 +1491,9 @@ export function Files({ const entry = singleEntry; if (entry?.revision) { event.preventDefault(); - setOperation({ - kind: event.key === "F2" ? "rename" : "delete", - parent: directory.path, - entry, - }); + if (event.key === "Delete") reviewDelete([entry]); + else + setOperation({ kind: "rename", parent: directory.path, entry }); } } else if (event.key === "F5") { event.preventDefault(); @@ -2059,49 +2092,22 @@ export function Files({ )} {operation && ( setOperation(null)} setBusy={setBusy} disabled={!connected} execute={async (name) => { if (operation.kind === "mkdir") await services.makeDirectory(operation.parent, name); - else if (operation.kind === "rename") + else await services.renameEntry( operation.entry!.path, name, operation.entry!.revision!, ); - else - await services.removeEntry( - operation.entry!.path, - operation.entry!.revision!, - ); setError(""); }} /> @@ -2109,6 +2115,7 @@ export function Files({ {deleteTarget && ( )} + {conflictPrompt && ( + + )} ); } diff --git a/src/components/DeleteFilesDialog.tsx b/src/components/DeleteFilesDialog.tsx index 5341d0c..ee674fb 100644 --- a/src/components/DeleteFilesDialog.tsx +++ b/src/components/DeleteFilesDialog.tsx @@ -8,12 +8,14 @@ import { deleteFiles, type DeleteResult } from "../file-delete"; import "./FileActionDialog.css"; export function DeleteFilesDialog({ entries, + parent, services, available, close, setBusy, }: { entries: readonly Readonly[]; + parent: string; services: SessionServices; available: boolean; close(): void; @@ -30,6 +32,8 @@ export function DeleteFilesDialog({ const [finished, setFinished] = useState(false); const [results, setResults] = useState([]); const [error, setError] = useState(""); + const [recursive, setRecursive] = useState(false); + const hasFolders = entries.some((entry) => entry.kind === "directory"); if (started.current && !available) stop.current = true; useEffect(() => { live.current = true; @@ -55,6 +59,7 @@ export function DeleteFilesDialog({ (result) => { if (live.current) setResults((old) => [...old, result]); }, + { recursive, parent }, ); } catch (error) { if (live.current) setError(String(error)); @@ -70,6 +75,7 @@ export function DeleteFilesDialog({ (result) => result.status === "deleted", ).length; const failed = results.find((result) => result.status === "failed"); + const partial = results.find((result) => result.status === "partial"); return createPortal(

- Files, links and empty folders are deleted permanently. Link targets are - kept. There is no trash or undo. Processing stops at the first failure. + Files and links are deleted permanently. Link targets are kept. There is + no trash or undo. Processing stops at the first failure.

+ {hasFolders && !finished && ( + + )}
    {entries.map((entry, index) => { const result = results[index]; @@ -103,24 +122,28 @@ export function DeleteFilesDialog({ {result?.status === "deleted" ? "Deleted" - : result?.status === "failed" - ? "Not confirmed" - : working && index === results.length - ? "Deleting…" - : finished - ? "Not attempted" - : entry.kind === "directory" - ? "Empty folder only" - : entry.kind} + : result?.status === "partial" + ? "Partially deleted" + : result?.status === "failed" + ? "Not confirmed" + : working && index === results.length + ? "Deleting…" + : finished + ? "Not attempted" + : entry.kind === "directory" + ? recursive + ? "Folder and contents" + : "Empty folder only" + : entry.kind} ); })}
- {(error || failed) && ( + {(error || failed || partial?.error) && (
{error || - `${failed!.entry.name}: ${failed!.error} Verify this item's remote state before retrying.`} + `${(failed || partial)!.entry.name}: ${(failed || partial)!.error} Verify this item's remote state before retrying.`}
)} {!available && ( @@ -131,7 +154,7 @@ export function DeleteFilesDialog({ )}

{finished - ? `${deleted} deleted${failed ? " · 1 not confirmed" : ""} · ${entries.length - results.length} not attempted` + ? `${deleted} deleted${failed ? " · 1 not confirmed" : ""}${partial ? " · 1 partially deleted" : ""} · ${entries.length - results.length} not attempted` : working ? `${deleted} of ${entries.length} deleted${stopping ? " · Stopping after the current item…" : ""}` : "Review the selected names and locations before deleting."} diff --git a/src/components/FileActionDialog.css b/src/components/FileActionDialog.css index 9e66d84..7f25dd3 100644 --- a/src/components/FileActionDialog.css +++ b/src/components/FileActionDialog.css @@ -99,6 +99,21 @@ overflow: auto; overflow-wrap: anywhere; } +.file-action-dialog .delete-folder-option { + display: flex; + align-items: flex-start; + gap: 0.615385rem; + font-size: 0.923077rem; + line-height: 1.5; +} +.file-action-dialog .delete-folder-option input { + display: inline-block; + width: auto; + flex: none; + margin: 0.230769rem 0 0; + padding: 0; + accent-color: var(--sc-danger, #dfb0ac); +} .move-file-dialog { width: min(40.769231rem, calc(100vw - 2.461538rem)); padding: 1.846154rem; diff --git a/src/components/TransferConflictDialog.tsx b/src/components/TransferConflictDialog.tsx new file mode 100644 index 0000000..cf6962c --- /dev/null +++ b/src/components/TransferConflictDialog.tsx @@ -0,0 +1,74 @@ +// SPDX-License-Identifier: MPL-2.0 +import { useEffect, useRef } from "react"; +import { createPortal } from "react-dom"; +import { Files } from "lucide-react"; +import { showModal } from "../dialog-compat"; +import type { TransferConflictReview } from "../sdk"; +import type { ConflictDecision } from "../transfer-queue"; +import "./FileActionDialog.css"; + +export function TransferConflictDialog({ + conflict, + remaining, + decide, +}: { + conflict: TransferConflictReview["conflicts"][number]; + remaining: number; + decide(decision: ConflictDecision): void; +}) { + const dialog = useRef(null); + useEffect(() => { + const previous = document.activeElement as HTMLElement | null; + showModal(dialog.current); + return () => { + if (previous?.isConnected) previous.focus({ preventScroll: true }); + }; + }, []); + const folder = conflict.sourceKind === "directory"; + return createPortal( +

{ + event.preventDefault(); + decide("cancel"); + }} + > +
+ +
+

REMOTE FILES

+

{folder ? "Folder already exists" : "File already exists"}

+

+ {conflict.destination.name} already exists on the host. + {folder + ? " Merge the folders and replace matching files? Files found only in the destination will stay." + : " Replace the existing file with the copied file?"}{" "} + There is no undo. +

+

{conflict.destination.path}

+ {remaining > 1 && ( +

+ {remaining - 1} more name {remaining === 2 ? "conflict" : "conflicts"}{" "} + in this transfer. +

+ )} +
+ + + {remaining > 1 && ( + + )} +
+ , + document.body, + ); +} diff --git a/src/components/TransferPanel.tsx b/src/components/TransferPanel.tsx index c9ba768..a3d586f 100644 --- a/src/components/TransferPanel.tsx +++ b/src/components/TransferPanel.tsx @@ -88,6 +88,7 @@ export function TransferPanel({ : "Transferring…" : { queued: "Queued", + reviewing: "Waiting for replacement choice", canceling: "Canceling…", "cancel-failed": "Cancel failed", completed: "Completed", diff --git a/src/file-delete.test.ts b/src/file-delete.test.ts index 7d30239..d65591a 100644 --- a/src/file-delete.test.ts +++ b/src/file-delete.test.ts @@ -76,3 +76,114 @@ it("validates the entire batch before deleting anything", async () => { ); expect(removeEntry).not.toHaveBeenCalled(); }); +it("deletes nested contents before folders without following links", async () => { + const folder: FileEntry = { + ...entries[0], + kind: "directory", + path: "opaque@folder", + }; + const child: FileEntry = { + ...entries[1], + kind: "directory", + path: "opaque@child", + }; + const link: FileEntry = { + ...entries[2], + kind: "symlink", + path: "opaque@link", + }; + const file = { ...entries[0], path: "opaque@file" }; + const list = vi.fn(async (path: string) => ({ + path, + entries: + path === "opaque@parent" + ? [folder] + : path === folder.path + ? [child, link] + : [file], + })); + const removeEntry = vi.fn(async () => {}); + const reports: DeleteResult[] = []; + await deleteFiles( + { list, removeEntry }, + [folder], + () => true, + (result) => reports.push(result), + { recursive: true, parent: "opaque@parent" }, + ); + expect(list.mock.calls.map(([path]) => path)).toEqual([ + "opaque@parent", + "opaque@folder", + "opaque@child", + ]); + expect(removeEntry.mock.calls).toEqual([ + [file.path, file.revision], + [child.path, child.revision], + [link.path, link.revision], + [folder.path, folder.revision], + ]); + expect(reports.map((result) => result.status)).toEqual(["deleted"]); +}); +it("does not descend into a selected folder that changed since confirmation", async () => { + const folder: FileEntry = { ...entries[0], kind: "directory" }; + const list = vi.fn(async (path: string) => ({ + path, + entries: [{ ...folder, revision: "changed" }], + })); + const removeEntry = vi.fn(); + await expect( + deleteFiles( + { list, removeEntry }, + [folder], + () => true, + () => {}, + { + recursive: true, + parent: "opaque@parent", + }, + ), + ).rejects.toThrow("changed"); + expect(removeEntry).not.toHaveBeenCalled(); +}); +it("reports partial folder deletion when stopped before the next item", async () => { + const folder: FileEntry = { ...entries[0], kind: "directory" }; + let allowed = true; + const list = vi.fn(async (path: string) => ({ + path, + entries: path === "opaque@parent" ? [folder] : [entries[1], entries[2]], + })); + const removeEntry = vi.fn(async () => { + allowed = false; + }); + const reports: DeleteResult[] = []; + await deleteFiles( + { list, removeEntry }, + [folder], + () => allowed, + (result) => reports.push(result), + { recursive: true, parent: "opaque@parent" }, + ); + expect(removeEntry).toHaveBeenCalledTimes(1); + expect(reports.map((result) => result.status)).toEqual(["partial"]); +}); +it("reports a partial folder when a child fails after another was removed", async () => { + const folder: FileEntry = { ...entries[0], kind: "directory" }; + const list = vi.fn(async (path: string) => ({ + path, + entries: path === "opaque@parent" ? [folder] : [entries[1], entries[2]], + })); + const removeEntry = vi + .fn() + .mockResolvedValueOnce(undefined) + .mockRejectedValueOnce(new Error("Denied")); + const reports: DeleteResult[] = []; + await deleteFiles( + { list, removeEntry }, + [folder], + () => true, + (result) => reports.push(result), + { recursive: true, parent: "opaque@parent" }, + ); + expect(reports[0].status).toBe("partial"); + expect(reports[0].error).toContain("Denied"); +}); diff --git a/src/file-delete.ts b/src/file-delete.ts index b93eb65..5e3ab5d 100644 --- a/src/file-delete.ts +++ b/src/file-delete.ts @@ -1,16 +1,65 @@ // SPDX-License-Identifier: MPL-2.0 -import type { FileEntry, SessionServices } from "./sdk"; +import type { Directory, FileEntry, SessionServices } from "./sdk"; export type DeleteResult = { entry: Readonly; - status: "deleted" | "failed"; + status: "deleted" | "partial" | "failed"; error?: string; }; +type DeleteServices = Pick & { + list(path: string): Promise>; +}; + +/** Walk provider-owned paths only. A link is removed as a link, never traversed. */ +async function deleteFolder( + services: DeleteServices, + root: Readonly, + canContinue: () => boolean, + onDelete: () => void, +): Promise { + const visited = new Set(); + const pending: { entry: Readonly; contentsListed: boolean }[] = [ + { entry: root, contentsListed: false }, + ]; + while (pending.length) { + if (!canContinue()) return false; + const current = pending.pop()!; + if (current.entry.kind === "directory" && !current.contentsListed) { + if (visited.has(current.entry.path)) + throw new Error( + "Folder listing contains a repeated path. Refresh before retrying.", + ); + visited.add(current.entry.path); + const directory = await services.list(current.entry.path); + if (directory.path !== current.entry.path) + throw new Error( + "Folder location changed during deletion. Refresh before retrying.", + ); + if ( + directory.entries.some((entry) => !entry.revision) || + new Set(directory.entries.map((entry) => entry.path)).size !== + directory.entries.length + ) + throw new Error( + "Folder contents changed or lack revisions. Refresh before retrying.", + ); + pending.push({ entry: current.entry, contentsListed: true }); + for (const entry of directory.entries.reverse()) + pending.push({ entry, contentsListed: false }); + } else { + await services.removeEntry(current.entry.path, current.entry.revision!); + onDelete(); + } + } + return true; +} + /** Sequential, revision-checked deletion; never retry an uncertain outcome. */ export async function deleteFiles( - services: Pick, + services: Pick & Partial, entries: readonly Readonly[], canContinue: () => boolean, report: (result: DeleteResult) => void, + options: { recursive?: boolean; parent?: string } = {}, ) { if ( !entries.length || @@ -21,13 +70,55 @@ export async function deleteFiles( throw new Error( "Choose up to 100 distinct items with current revisions. Refresh the folder first.", ); + if (options.recursive && !services.list) + throw new Error( + "This file provider cannot list folder contents for deletion.", + ); + if ( + options.recursive && + entries.some((entry) => entry.kind === "directory") + ) { + if (!options.parent) + throw new Error("Refresh the parent folder before deleting."); + const parent = await services.list!(options.parent); + if (parent.path !== options.parent) + throw new Error("The parent folder changed. Refresh before deleting."); + for (const entry of entries) { + const current = parent.entries.find((item) => item.path === entry.path); + if ( + !current || + current.revision !== entry.revision || + current.kind !== entry.kind + ) + throw new Error( + `${entry.name} changed. Refresh the folder before deleting.`, + ); + } + } for (const entry of entries) { if (!canContinue()) break; + let deletedWithinFolder = 0; try { - await services.removeEntry(entry.path, entry.revision!); + if (entry.kind === "directory" && options.recursive) { + if ( + !(await deleteFolder( + services as DeleteServices, + entry, + canContinue, + () => deletedWithinFolder++, + )) + ) { + report({ entry, status: "partial" }); + break; + } + } else await services.removeEntry(entry.path, entry.revision!); report({ entry, status: "deleted" }); } catch (error) { - report({ entry, status: "failed", error: String(error) }); + report({ + entry, + status: deletedWithinFolder ? "partial" : "failed", + error: String(error), + }); break; } } diff --git a/src/sdk.ts b/src/sdk.ts index c828502..c2f7c91 100644 --- a/src/sdk.ts +++ b/src/sdk.ts @@ -44,6 +44,14 @@ export interface TransferTicket { size: number; direction: "upload" | "download" | "copy" | "move"; } +export interface TransferConflictReview { + conflicts: { destination: FileEntry; sourceKind: string }[]; + canReplace: boolean; +} +export interface TransferPolicy { + replace: { path: string; revision: string }[]; + skip: { path: string; revision: string }[]; +} export function transferCapability( direction: TransferTicket["direction"], ): Capability { @@ -330,7 +338,12 @@ export interface HostServices { id: number, onProgress: (event: TransferProgress) => void, tracked?: string[], + policy?: TransferPolicy, ): Promise; + transferConflicts?( + sessionId: number, + id: number, + ): Promise; cancelTransfer(sessionId: number, id: number): Promise; createText( sessionId: number, @@ -456,7 +469,9 @@ export interface SessionServices { runTransfer( ticket: TransferTicket, onProgress: (event: TransferProgress) => void, + policy?: TransferPolicy, ): Promise; + transferConflicts?(ticket: TransferTicket): Promise; cancelTransfer(id: number): Promise; createText(parent: string, name: string, text: string): Promise; makeDirectory(parent: string, name: string): Promise; diff --git a/src/services.ts b/src/services.ts index 9771900..c5bc43a 100644 --- a/src/services.ts +++ b/src/services.ts @@ -161,7 +161,9 @@ function createNativeServices(pins?: SourcePins): HostServices { invoke("choose_upload_files", { sessionId, parent, folder }), chooseDownload: (sessionId, path, revision) => invoke("choose_download_file", { sessionId, path, revision }), - runTransfer: (sessionId, transferId, onProgress, tracked) => { + transferConflicts: (sessionId, transferId) => + invoke("transfer_conflicts", { sessionId, transferId }), + runTransfer: (sessionId, transferId, onProgress, tracked, policy) => { const onEvent = new Channel(); onEvent.onmessage = onProgress; return invoke("run_transfer", { @@ -169,6 +171,7 @@ function createNativeServices(pins?: SourcePins): HostServices { transferId, onEvent, tracked, + policy, }); }, cancelTransfer: (sessionId, transferId) => diff --git a/src/session-services.ts b/src/session-services.ts index 0dbc899..966a8a5 100644 --- a/src/session-services.ts +++ b/src/session-services.ts @@ -4,6 +4,7 @@ import type { HostServices, Session, SessionServices, + TransferProgress, TransferTicket, } from "./sdk"; import { @@ -403,7 +404,21 @@ export function bindSession( null ); }, - runTransfer: async (ticket, onProgress) => { + transferConflicts: backend.transferConflicts + ? async (ticket) => { + const owned = tickets.get(ticket.id); + if (!owned) + throw new Error("Transfer does not belong to this workspace"); + const expected = generation; + const result = await backend.transferConflicts!( + check(transferCapability(owned.direction)), + ticket.id, + ); + check(transferCapability(owned.direction), expected); + return result; + } + : undefined, + runTransfer: async (ticket, onProgress, policy) => { const owned = tickets.get(ticket.id); if (!owned) throw new Error("Transfer does not belong to this workspace"); if (cleanupTickets.has(ticket.id)) @@ -422,15 +437,19 @@ export function bindSession( fileClipboard(services).trackedPaths(), ); dispatched = true; - const result = await backend.runTransfer( - id, - ticket.id, - (event) => { - if (valid(transferCapability(owned.direction), expected)) - onProgress(event); - }, - follow?.tracked, - ); + const progress = (event: TransferProgress) => { + if (valid(transferCapability(owned.direction), expected)) + onProgress(event); + }; + const result = policy + ? await backend.runTransfer( + id, + ticket.id, + progress, + follow?.tracked, + policy, + ) + : await backend.runTransfer(id, ticket.id, progress, follow?.tracked); // Interrupted folders can contain successfully completed children. if (owned.direction !== "download") mutationCompleted( diff --git a/src/transfer-queue.test.ts b/src/transfer-queue.test.ts index 28081cb..befadca 100644 --- a/src/transfer-queue.test.ts +++ b/src/transfer-queue.test.ts @@ -3,7 +3,13 @@ import { expect, it, vi } from "vitest"; import { pendingTransfer, TransferQueue } from "./transfer-queue"; import { bindSession } from "./session-services"; import { previewServices, previewSession } from "./preview"; -import type { TransferOutcome, TransferTicket } from "./sdk"; +import type { + SessionServices, + TransferConflictReview, + TransferOutcome, + TransferPolicy, + TransferTicket, +} from "./sdk"; const ticket = (id: number): TransferTicket => ({ id, name: `file${id}`, @@ -166,3 +172,114 @@ it("does not replace a confirmed completion with a late cancellation error", asy await cancellation; expect(queue.snapshot()[0].status).toBe("completed"); }); +it("asks once per conflict and passes reviewed replacements to the backend", async () => { + const conflicts = ["one", "two"].map((name) => ({ + sourceKind: "file", + destination: { + name, + path: `remote@${name}`, + revision: `rev-${name}`, + kind: "file" as const, + size: 1, + modified: null, + }, + })); + const runTransfer = vi.fn( + async ( + _ticket: TransferTicket, + _progress: (event: never) => void, + _policy?: TransferPolicy, + ) => ({ + status: "completed" as const, + bytes: 64, + total: 64, + }), + ); + const services = { + transferConflicts: async () => ({ conflicts, canReplace: true }), + runTransfer, + cancelTransfer: vi.fn(async () => {}), + } as unknown as SessionServices; + const choices = vi.fn(async () => "replace-all" as const); + const queue = new TransferQueue(services, undefined, choices); + queue.enqueue([ticket(1)]); + await vi.waitFor(() => expect(queue.snapshot()[0].status).toBe("completed")); + expect(choices).toHaveBeenCalledTimes(1); + expect(runTransfer.mock.calls[0][2]).toEqual({ + replace: [ + { path: "remote@one", revision: "rev-one" }, + { path: "remote@two", revision: "rev-two" }, + ], + skip: [], + }); +}); +it("cancels a reviewed conflict without starting a transfer", async () => { + const runTransfer = vi.fn(); + const cancelTransfer = vi.fn(async () => {}); + const services = { + transferConflicts: async () => ({ + canReplace: true, + conflicts: [ + { + sourceKind: "file", + destination: { + name: "one", + path: "remote@one", + revision: "rev-one", + kind: "file", + size: 1, + modified: null, + }, + }, + ], + }), + runTransfer, + cancelTransfer, + } as unknown as SessionServices; + const queue = new TransferQueue(services, undefined, async () => "cancel"); + queue.enqueue([ticket(1)]); + await vi.waitFor(() => expect(queue.snapshot()[0].status).toBe("canceled")); + expect(cancelTransfer).toHaveBeenCalledWith(1); + expect(runTransfer).not.toHaveBeenCalled(); +}); +it("does not open a conflict prompt after cancellation during inspection", async () => { + let finishReview!: (value: TransferConflictReview) => void; + const transferConflicts = vi.fn( + () => + new Promise((resolve) => { + finishReview = resolve; + }), + ); + const runTransfer = vi.fn(); + const cancelTransfer = vi.fn(async () => {}); + const resolveConflict = vi.fn(async () => "replace" as const); + const services = { + transferConflicts, + runTransfer, + cancelTransfer, + } as unknown as SessionServices; + const queue = new TransferQueue(services, undefined, resolveConflict); + queue.enqueue([ticket(1)]); + await vi.waitFor(() => expect(transferConflicts).toHaveBeenCalledOnce()); + await queue.cancel(1); + finishReview({ + canReplace: true, + conflicts: [ + { + sourceKind: "file", + destination: { + name: "one", + path: "remote@one", + revision: "rev-one", + kind: "file", + size: 1, + modified: null, + }, + }, + ], + }); + await vi.waitFor(() => expect(queue.snapshot()[0].status).toBe("canceled")); + expect(resolveConflict).not.toHaveBeenCalled(); + expect(runTransfer).not.toHaveBeenCalled(); + expect(cancelTransfer).toHaveBeenCalledWith(1); +}); diff --git a/src/transfer-queue.ts b/src/transfer-queue.ts index 21bbee2..1ed103c 100644 --- a/src/transfer-queue.ts +++ b/src/transfer-queue.ts @@ -4,12 +4,15 @@ import type { TransferOutcome, TransferProgress, TransferTicket, + TransferPolicy, + TransferConflictReview, } from "./sdk"; import { TransferCleanupError } from "./transfer-errors"; export interface TransferRow extends TransferTicket { status: | "queued" | "running" + | "reviewing" | "canceling" | "cancel-failed" | TransferOutcome["status"]; @@ -21,16 +24,26 @@ export interface TransferRow extends TransferTicket { path?: string | null; } export const pendingTransfer = (row: TransferRow) => - ["queued", "running", "canceling", "cancel-failed"].includes(row.status); + ["queued", "running", "reviewing", "canceling", "cancel-failed"].includes( + row.status, + ); +export type ConflictDecision = "replace" | "replace-all" | "cancel"; +export type ConflictResolver = ( + conflict: TransferConflictReview["conflicts"][number], + remaining: number, + signal: AbortSignal, +) => Promise; /** One stream per Files window; the native scheduler also caps cross-window work. */ export class TransferQueue { private rows: TransferRow[] = []; private listeners = new Set<() => void>(); private running: number | null = null; private disposed = false; + private reviewAbort: AbortController | null = null; constructor( private services: SessionServices, private reportError: (message: string) => void = console.warn, + private resolveConflict?: ConflictResolver, ) {} activate() { this.disposed = false; @@ -80,18 +93,97 @@ export class TransferQueue { if (!row) return; this.running = row.id; this.update(row.id, { status: "running" }); + let dispatched = false; + const reviewAbort = new AbortController(); + this.reviewAbort = reviewAbort; try { - const result = await this.services.runTransfer(row, (event) => - this.update(row.id, event), + let policy: TransferPolicy | undefined; + if ( + (row.direction === "upload" || row.direction === "copy") && + this.services.transferConflicts + ) { + const review = await this.services.transferConflicts(row); + if (reviewAbort.signal.aborted) { + await this.services.cancelTransfer(row.id); + this.update(row.id, { status: "canceled", message: null }); + return; + } + if (review.conflicts.length) { + if ( + !review.canReplace || + !this.resolveConflict || + review.conflicts.some( + ({ destination, sourceKind }) => + destination.kind !== sourceKind || + !["file", "directory"].includes(sourceKind), + ) + ) + throw new Error( + "Safe replacement is unavailable for one or more destination items. Nothing was replaced.", + ); + this.update(row.id, { status: "reviewing" }); + const replace: TransferPolicy["replace"] = []; + for (let i = 0; i < review.conflicts.length; i++) { + if (reviewAbort.signal.aborted) break; + const decision = await this.resolveConflict( + review.conflicts[i], + review.conflicts.length - i, + reviewAbort.signal, + ); + if (decision === "cancel") break; + const selected = + decision === "replace-all" + ? review.conflicts.slice(i) + : [review.conflicts[i]]; + replace.push( + ...selected.map(({ destination }) => ({ + path: destination.path, + revision: destination.revision!, + })), + ); + if (decision === "replace-all") break; + } + if ( + reviewAbort.signal.aborted || + replace.length !== review.conflicts.length + ) { + await this.services.cancelTransfer(row.id); + this.update(row.id, { status: "canceled", message: null }); + return; + } + policy = { replace, skip: [] }; + this.update(row.id, { status: "running" }); + } + } + this.reviewAbort = null; + dispatched = true; + const result = await this.services.runTransfer( + row, + (event) => this.update(row.id, event), + policy, ); this.update(row.id, result); } catch (error) { + if (!dispatched) { + try { + await this.services.cancelTransfer(row.id); + if (reviewAbort.signal.aborted) { + this.update(row.id, { status: "canceled", message: null }); + return; + } + } catch (cleanup) { + error = new TransferCleanupError( + `${error}; transfer cleanup failed: ${cleanup}`, + ); + } + } this.update(row.id, { status: error instanceof TransferCleanupError ? "cancel-failed" : "failed", message: String(error), }); } finally { + this.reviewAbort = null; this.running = null; void this.pump(); } @@ -100,6 +192,11 @@ export class TransferQueue { const row = this.rows.find((row) => row.id === id); if (!row || !pendingTransfer(row) || row.status === "canceling") return; const queued = this.running !== id; + if (!queued && this.reviewAbort) { + this.reviewAbort?.abort(); + this.update(id, { status: "canceling", message: null }); + return; + } // Keep ownership and close guards until native cancellation is acknowledged. // A failed queued cancellation must never start that unwanted upload later. this.update(id, { status: "canceling", message: null }); @@ -121,6 +218,7 @@ export class TransferQueue { } dispose() { this.disposed = true; + this.reviewAbort?.abort(); for (const row of this.rows.filter(pendingTransfer)) void this.services .cancelTransfer(row.id)