Confirm file replacements and non-empty folder deletion

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<FileLocation> {
+    copy_regular_file_with_replace(
+        service, path, revision, parent, name, None, canceled, progress,
+    )
+    .await
+}
+
+pub async fn copy_regular_file_with_replace(
+    service: Arc<dyn FileTransferService>,
+    path: &str,
+    revision: &str,
+    parent: &str,
+    name: &str,
+    replace_revision: Option<&str>,
+    canceled: impl Fn() -> bool + Send,
+    progress: &mut (dyn FnMut(CopyProgress) + Send),
 ) -> Result<FileLocation> {
     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 {

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<Option<FileEntry>> {
+        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<Self>,
+        _parent: &str,
+        _name: &str,
+        _size: u64,
+        _expected_revision: &str,
+    ) -> Result<Box<dyn TransferWriter>> {
+        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<Self>, path: &str, revision: &str) -> Result<FileEntry> {
         let mut reader = self.download(path, revision).await?;

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<String>,
     temporary: Option<String>,
     destination: String,
+    expected_revision: Option<String>,
     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<Option<FileEntry>> {
+        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<Self>,
+        parent: &str,
+        name: &str,
+        size: u64,
+        expected_revision: &str,
+    ) -> Result<Box<dyn TransferWriter>> {
+        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<Self>,
         path: &str,
@@ -304,9 +386,28 @@ impl FileTransferService for SftpTextFiles {
         parent: &str,
         name: &str,
         size: u64,
+    ) -> Result<Box<dyn TransferWriter>> {
+        self.prepare_upload(parent, name, size, None).await
+    }
+}
+
+impl SftpTextFiles {
+    async fn prepare_upload(
+        self: Arc<Self>,
+        parent: &str,
+        name: &str,
+        size: u64,
+        expected_revision: Option<&str>,
     ) -> Result<Box<dyn TransferWriter>> {
         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,
         }))

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,

src-tauri/src/transfer_folder_tests.rs

@@ -19,6 +19,7 @@ struct Folders {
     data: Arc<super::tests::Memory>,
     created: Mutex<HashSet<String>>,
     bad_child: Option<FileEntry>,
+    merge_destination: bool,
 }
 struct Page(Vec<FileEntry>);
 #[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<Option<FileEntry>> {
+        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<Self>,
+        parent: &str,
+        name: &str,
+        size: u64,
+        expected_revision: &str,
+    ) -> Result<Box<dyn shellcanvas_core::TransferWriter>> {
+        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<FileEntry>) -> Arc<dyn FileTransferService> {
         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<dyn FileTransferService> = 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