put: remove partial uploads when the transfer fails

A failed or interrupted upload left the already-uploaded payloads — or
a truncated .changes — in the PPA's incoming area. On failure the
already-uploaded files are now removed best-effort in reverse upload
order with the failed file first, so a .changes never outlives the
payloads it references; the original upload error keeps precedence over
cleanup failures, and record-after-success semantics are unchanged (a
failed upload must not count as uploaded).

src/put/mod.rs

@@ -6,7 +6,10 @@
 //!
 //! Payload files are uploaded first and the `.changes` file last, like
 //! dput does, so a partially uploaded set cannot be picked up by the
-//! server-side queue processors.
+//! server-side queue processors. A run that fails mid-upload removes its
+//! already-uploaded files from the incoming queue (best effort), so a
+//! retried upload starts from a clean queue; a failed upload is never
+//! recorded in the upload log, so a re-run replays every file.
 
 pub mod changes;
 pub mod ssh;
@@ -147,23 +150,50 @@ pub async fn put(
         .to_string();
     uploads.push((changes_path.clone(), changes_name.clone()));
 
+    let incoming = target.incoming.trim_end_matches('/');
+
+    // Remote names of the files uploaded so far, in upload order (the
+    // .changes last). A run failing mid-upload removes these from the
+    // write-only incoming queue before returning: the uploaded payloads
+    // would otherwise linger in the queue area forever, and a .changes
+    // truncated by a failed close could even be picked up by the scanner.
+    let mut uploaded: Vec<String> = Vec::new();
+
     for (path, name) in &uploads {
-        let size = path
-            .metadata()
-            .map_err(|e| format!("cannot stat '{}': {}", path.display(), e))?
-            .len();
+        let size = match path.metadata() {
+            Ok(metadata) => metadata.len(),
+            Err(e) => {
+                // Nothing was attempted for this file: only what earlier
+                // iterations uploaded needs removing
+                let error: Box<dyn std::error::Error> =
+                    format!("cannot stat '{}': {}", path.display(), e).into();
+                cleanup_partial_upload(&sftp, incoming, &uploaded, None, &host);
+                return Err(error);
+            }
+        };
         // Same transfer view as pull: prefix line, bar on its own line
         let bar = multi.add(ProgressBar::new(size));
         bar.enable_steady_tick(std::time::Duration::from_millis(50));
         bar.set_style(ui::transfer_style());
         bar.set_prefix(format!("Uploading {name}..."));
 
-        let remote = format!("{}/{}", target.incoming.trim_end_matches('/'), name);
+        let remote = format!("{incoming}/{name}");
         let result = ssh::upload_file(&sftp, path, &remote, &host, &bar);
         bar.finish_and_clear();
-        result?;
+        if let Err(e) = result {
+            // The failed file itself joins the cleanup: its remote `create`
+            // may have succeeded before the failure, leaving a partial — or,
+            // on a failed close, a truncated — file behind
+            cleanup_partial_upload(&sftp, incoming, &uploaded, Some(name), &host);
+            return Err(e);
+        }
+        uploaded.push(name.clone());
     }
 
+    // Recorded only once the whole upload succeeded: the log backs the
+    // duplicate-upload guard, and a failed upload must not count as
+    // uploaded (a re-run replays every file — `sftp.create` truncates, so
+    // replaying is safe).
     record_upload(&upload_log_path()?, &record)?;
 
     // The completion lines replace the summary bar; the guard's own clear
@@ -180,6 +210,48 @@ pub async fn put(
     Ok(())
 }
 
+/// The remote names to attempt removing after a failed upload: everything
+/// already uploaded plus, when set, `failed` (the file whose upload just
+/// failed — its remote `create` may have succeeded before the failure,
+/// leaving a partial or truncated file behind) — in reverse upload order,
+/// so a `.changes` is removed before the payloads it references and the
+/// queue scanner never observes the payload set shrinking under a
+/// still-present `.changes`. Pure so the ordering decision is testable
+/// without a server; the network side is [`cleanup_partial_upload`].
+fn cleanup_list(uploaded: &[String], failed: Option<&str>) -> Vec<String> {
+    let mut names: Vec<String> = uploaded.to_vec();
+    if let Some(failed) = failed {
+        names.push(failed.to_string());
+    }
+    names.reverse();
+    names
+}
+
+/// Best-effort removal of what a failed upload left in the target's
+/// incoming queue: the already-uploaded payloads, and a `.changes`
+/// truncated by a failed close, would otherwise linger in the write-only
+/// area until the queue is manually cleaned. Launchpad re-validates every
+/// upload, so this is pure hygiene: a removal failure is logged and
+/// skipped, and the caller returns the original upload error — never a
+/// cleanup one.
+fn cleanup_partial_upload(
+    sftp: &ssh2::Sftp,
+    incoming: &str,
+    uploaded: &[String],
+    failed: Option<&str>,
+    host: &str,
+) {
+    for name in cleanup_list(uploaded, failed) {
+        let remote = format!("{incoming}/{name}");
+        match ssh::remove_file(sftp, &remote, host) {
+            Ok(()) => info!("Removed leftover {remote} from the failed upload"),
+            Err(e) => {
+                log::warn!("Could not remove the leftover {remote} of the failed upload: {e}")
+            }
+        }
+    }
+}
+
 /// Steady tick interval of every bar rendered by a `put` run
 const TICK: std::time::Duration = std::time::Duration::from_millis(50);
 
@@ -864,4 +936,56 @@ mod tests {
         let changes = changes_fixture("1.7-1");
         check_not_superseded(&target, &changes, &published, false).unwrap();
     }
+
+    /// The upload loop is network-bound and `ssh2::Sftp` cannot be built
+    /// without a live SSH session, so there is no stub seam for it without
+    /// heavy refactoring; the cleanup decision is factored into the pure
+    /// [`cleanup_list`] and tested directly instead.
+    ///
+    /// Failure during the final `.changes` upload (the dangerous case: a
+    /// partial or close-truncated `.changes` is scanner-visible): the
+    /// `.changes` is removed first, then the payloads it references, so the
+    /// queue never sees a `.changes` over a shrinking payload set.
+    #[test]
+    fn cleanup_list_removes_changes_first_then_payloads_in_reverse() {
+        let uploaded = vec![
+            "hello_1.0.orig.tar.xz".to_string(),
+            "hello_1.0-1.debian.tar.xz".to_string(),
+            "hello_1.0-1.dsc".to_string(),
+        ];
+        assert_eq!(
+            cleanup_list(&uploaded, Some("hello_1.0-1_source.changes")),
+            vec![
+                "hello_1.0-1_source.changes",
+                "hello_1.0-1.dsc",
+                "hello_1.0-1.debian.tar.xz",
+                "hello_1.0.orig.tar.xz",
+            ]
+        );
+    }
+
+    /// A failure during a payload upload: the `.changes` was never
+    /// attempted (it is uploaded last), so it must not appear in the
+    /// cleanup list — only the failed payload (a partial may exist) and the
+    /// earlier payloads.
+    #[test]
+    fn cleanup_list_before_the_changes_covers_the_failed_payload() {
+        let uploaded = vec!["hello_1.0.orig.tar.xz".to_string()];
+        assert_eq!(
+            cleanup_list(&uploaded, Some("hello_1.0-1.debian.tar.xz")),
+            vec!["hello_1.0-1.debian.tar.xz", "hello_1.0.orig.tar.xz"]
+        );
+    }
+
+    /// A stat failure happens before anything was attempted for that file,
+    /// so only what earlier iterations uploaded is removed — and with
+    /// nothing uploaded yet there is nothing to remove at all.
+    #[test]
+    fn cleanup_list_without_failed_file_keeps_uploaded_only() {
+        assert!(cleanup_list(&[], None).is_empty());
+        assert_eq!(
+            cleanup_list(&["hello_1.0.orig.tar.xz".to_string()], None),
+            vec!["hello_1.0.orig.tar.xz"]
+        );
+    }
 }

src/put/ssh.rs

@@ -632,6 +632,18 @@ pub fn upload_file(
     Ok(())
 }
 
+/// Remove the remote file `remote` on `host` (SFTP remove). Used by the
+/// best-effort cleanup of an interrupted upload: the queue is write-only,
+/// so the only remote operation ever needed besides `create` is this one.
+pub fn remove_file(
+    sftp: &ssh2::Sftp,
+    remote: &str,
+    host: &str,
+) -> Result<(), Box<dyn std::error::Error>> {
+    sftp.unlink(Path::new(remote))
+        .map_err(|e| format!("cannot remove remote file {remote} on {host}: {e}").into())
+}
+
 /// Open the SFTP subsystem on `session`
 pub fn sftp(session: &Session) -> Result<ssh2::Sftp, Box<dyn std::error::Error>> {
     session