put: degrade to the anonymous FTP queue when the SSH transport fails

pkh put only spoke SFTP to the PPA queue, so a failure of the SSH
transport itself (TCP, banner exchange) failed the upload even though
dput happily pushes the same files: its plain ppa: profile goes over
the anonymous FTP queue of ppa.launchpad.net, the same destination
over another port.

Classify the SSH connection failures: Transport (the connection never
came up: resolution, TCP, banner or key exchange) degrades to that FTP
queue — the upload order (payload first, .changes last), the
reverse-order DELE cleanup of a failed upload and the per-chunk
progress reporting all mirror the SFTP path, sharing cleanup_list.
Refused failures (host key not accepted, no matching authentication)
stay errors: silently switching transport would bypass the refusal.

The FTP client is suppaftp's blocking stream, with the time bounds it
does not carry by itself: the control channel's reads and writes, the
data channel's writes and connect (through a custom passive stream
builder), and the NAT workaround for PASV replies announcing an
unroutable address. The queue endpoints (host, port) join
data/launchpad.yml next to the SFTP ones, and the FTP transport is
covered by unit tests against an in-process fake queue plus a live
control-channel handshake with the real server (ignored, network).

.gitignore

@@ -1,4 +1,3 @@
-*.lock
 target
 
 # Local snapcraft builds

Cargo.toml

@@ -35,6 +35,8 @@ gpgme = "0.11"
 serde_yaml = "0.9"
 lazy_static = "1.4.0"
 unicode-width = "0.2"
+parking_lot = "0.12"
+suppaftp = "12"
 
 [dev-dependencies]
 test-log = "0.2.19"

README.md

@@ -2,6 +2,22 @@
 
 `pkh` is a packaging helper for Debian/Ubuntu packages.
 
+## Installation
+
+No distribution channel is published yet; build from source:
+
+```
+sudo apt install pkg-config libssl-dev libgpg-error-dev libgpgme-dev
+git clone https://git.vhaudiquet.fr/vhaudiquet/pkh.git
+cd pkh
+cargo install --path .
+```
+
+At runtime pkh shells out to the Debian packaging toolchain (git,
+dpkg-dev, quilt, mmdebstrap, lintian, pristine-tar, ...): install the
+ones your workflows use, or build the classic snap from
+`snap/snapcraft.yaml` (`snapcraft pack`), which carries them.
+
 ## Usage and features
 
 ### Basic concepts
@@ -25,12 +41,19 @@ Options:
 Commands and workflows include:
 ```
 Commands:
+  new    Scaffold a new Debian source package (buildable right away)
   pull   Pull a source package from the archive or git
   chlog  Auto-generate changelog entry, editing it, committing it afterwards
   build  Build the source package (into a .dsc)
-  deb    Build the source package into binary package (.deb)
   put    Upload the built source package to a PPA
+  deb    Build the source package into binary package (.deb)
+  lint   Lint the package (lintian wrapper + pkh-native checks)
+  prune  Prune residual pkh build artifacts and caches
   help   Print this message or the help of the given subcommand(s)
+
+Options:
+  -h, --help     Print help
+  -V, --version  Print version
 ```
 
 ### Examples
@@ -66,14 +89,15 @@ That is a lot of different tools and operations. With pkh, the same workflow:
 pkh pull hello # needs -d ubuntu if you are not running Ubuntu
 # Apply the patch to the package
 ...
-pkh commit -m "Applied patch xxx"
+git add debian/patches/xxx.patch
+git commit -m "Applied patch xxx"
 pkh chlog
 # Test that the package builds
 pkh build
 pkh deb
 # Upload the package to a ppa
 pkh put --ppa user/hello_xxx
-# Push previously commited changes
+# Push the commits to your fork
 git push xxx user-fork
 ```
 
@@ -85,20 +109,23 @@ Missing features:
   - [x] Obtain package sources from git
   - [x] Obtain package sources from the archive (fallback)
   - [x] Obtain package source from PPA (--ppa)
-  - [ ] Obtain a specific version of the package
+  - [x] Obtain a specific version of the package
   - [x] Fetch the correct git branch for series on Ubuntu
   - [ ] Try to fetch the correct git branch for series on Debian, or fallback to the archive
 - [ ] `pkh chlog`
   - [x] Auto-generate changelog entry
   - [x] Extra flags: backport, non-maintainer upload, no change rebuild, ...
+  - [x] Select the target series, matching changelog suite names with their series (unstable ≡ sid)
   - [ ] Commit changelog entry
+- [x] `pkh new`
+  - [x] Scaffold a new Debian source package (interactive, multiple languages)
 - [ ] `pkh build`
   - [x] Build the source package
 - [ ] `pkh deb`
   - [x] Build the binary package
   - [x] Build for a specific architecture
   - [ ] Three build modes:
-    - [ ] Build locally (discouraged)
+    - [x] Build locally (discouraged)
     - [x] Build using unshare chroot, with binary emulation (default)
     - [x] Cross-compilation
   - [ ] Async build
@@ -106,11 +133,14 @@ Missing features:
   - [ ] Show build status
 - [ ] `pkh put`
   - [x] Upload the source package to a PPA (native SFTP, no `dput` dependency)
+  - [x] Degrade to the anonymous FTP queue when the SSH connection never comes up (dput's upload method)
   - [ ] Upload the source package to the archive
 - [ ] `pkh commit`
   - [ ] Commit the changes to git
-- [ ] `pkh lint`
+- [x] `pkh lint`
-  - [ ] Lint the package
+  - [x] Lint the package
+- [x] `pkh prune`
+  - [x] Prune residual pkh build artifacts and caches
 - [ ] `pkh test`
   - [ ] Run autopkgtest
   - [ ] Provide options: local (discouraged), chroot, VM?, ppa

data/distro_info.yml

@@ -28,6 +28,12 @@
 ##                   series (`<series>-updates`, ...). Deliberately not the
 ##                   `pockets` key: that one is the *search order* of pull,
 ##                   where backports must not fold in.
+##   suite_aliases:  the changelog suite names that alias a series
+##                   codename: Debian packages conventionally target
+##                   'unstable' where the series data carries 'sid'.
+##                   Mapped suite name -> series codename; the two
+##                   names identify the same series, and the selector
+##                   offers the aliased entry as '<suite> (<series>)'.
 ##   build_profiles: the vendor's default DEB_BUILD_PROFILES (Ubuntu
 ##                   activates derivative.ubuntu noudeb, Debian none),
 ##                   mirroring what Dpkg::BuildProfiles resolves when the
@@ -47,6 +53,10 @@ dist:
       - updates
       - security
       - proposed-updates
+    # Debian changelogs conventionally target 'unstable'; the series data
+    # knows the same series as 'sid'.
+    suite_aliases:
+      unstable: sid
     sections:
       # Valid Section values for debian/control: the Debian policy section
       # list unioned with the sections observed in the live Ubuntu archive.

data/launchpad.yml

@@ -13,6 +13,9 @@
 ##   ssh_*:        the PPA upload queue, as expanded by dput-ng's
 ##                 ppa:user/ppa profile (ppa.launchpad.net:22, incoming
 ##                 ~<user>/<ppa>)
+##   ftp_*:        the same upload queue over anonymous FTP, dput-ng's
+##                 plain ppa: profile: the transport pkh degrades to
+##                 when the SSH connection itself never comes up
 ##   content_host_template: ppa.launchpadcontent.net serves PPA apt
 ##                 repositories since the 2022 move off ppa.launchpad.net
 ##   git_web_template: Launchpad's CGit mirrors of Ubuntu source packages
@@ -21,6 +24,10 @@
 api_base: https://api.launchpad.net/1.0
 ssh_host: ppa.launchpad.net
 ssh_port: 22
+## The anonymous FTP upload queue dput-ng's plain ppa: profile uses:
+## pkh degrades to it when the SSH connection itself never comes up.
+ftp_host: ppa.launchpad.net
+ftp_port: 21
 incoming_template: "~{owner}/{ppa}"
 content_host_template: https://ppa.launchpadcontent.net/{owner}/{ppa}/ubuntu
 git_web_template: https://git.launchpad.net/ubuntu/+source/{package}

src/changelog.rs

@@ -294,9 +294,16 @@ pub enum SeriesCandidates {
     /// cannot be made (cancelled, no interactive user) `fallback` — the
     /// changelog's current series — is used instead.
     Choose {
-        /// Series names to offer.
+        /// Selector labels: each series name, or `<suite> (<series>)`
+        /// for a series aliased by a changelog suite name (Debian's
+        /// 'unstable (sid)').
         options: Vec<String>,
-        /// Preselected series.
+        /// Changelog distribution each label of `options` maps to,
+        /// parallel to it: an aliased label selects its suite name —
+        /// what a changelog distribution field expects — while every
+        /// other label selects itself.
+        values: Vec<String>,
+        /// Preselected label.
         default: String,
         /// Series to fall back to when nothing can be selected.
         fallback: String,
@@ -312,8 +319,10 @@ pub enum SeriesCandidates {
 /// An UNRELEASED entry offers itself as a pinned first option (selecting it
 /// keeps the changelog unreleased) on top of the current vendor's series
 /// list, defaulting to the development series; any other series resolves
-/// through the series list of its own distribution. `None` when the
+/// through the series list of its own distribution. A changelog suite name
-/// changelog cannot be parsed (no default to derive at all).
+/// (Debian's 'unstable') identifies the same series as its alias's codename
+/// ('sid') and resolves to that dist's list. `None` when the changelog
+/// cannot be parsed (no default to derive at all).
 pub async fn series_candidates(changelog_path: &Path) -> Option<SeriesCandidates> {
     let (_package, _version, current) = parse_changelog_header(changelog_path).ok()?;
 
@@ -321,10 +330,13 @@ pub async fn series_candidates(changelog_path: &Path) -> Option<SeriesCandidates
         // Vendors keep original casing ("Ubuntu"), while the series data
         // keys are lowercase
         let dist = crate::build::env::current_vendor().to_lowercase();
-        let mut options = vec![crate::distro_info::UNRELEASED.to_string()];
         match crate::distro_info::get_ordered_series_name(&dist).await {
             Ok(series_list) => {
-                options.extend(series_list);
+                let (labels, series_values, _) = selector_options(&dist, &series_list, "");
+                let mut options = vec![crate::distro_info::UNRELEASED.to_string()];
+                let mut values = vec![crate::distro_info::UNRELEASED.to_string()];
+                options.extend(labels);
+                values.extend(series_values);
                 // Default to the development series (the first real entry),
                 // not to the pinned UNRELEASED entry itself
                 let default = if options.len() > 1 {
@@ -334,6 +346,7 @@ pub async fn series_candidates(changelog_path: &Path) -> Option<SeriesCandidates
                 };
                 Some(SeriesCandidates::Choose {
                     options,
+                    values,
                     default,
                     fallback: current,
                 })
@@ -341,20 +354,79 @@ pub async fn series_candidates(changelog_path: &Path) -> Option<SeriesCandidates
             Err(_) => Some(SeriesCandidates::Keep(current)),
         }
     } else {
-        match crate::distro_info::get_dist_from_series(&current).await {
+        // The changelog may target a suite name instead of a series
-            Ok(dist) => match crate::distro_info::get_ordered_series_name(&dist).await {
+        // codename: Debian conventionally writes 'unstable' where the
+        // series data carries 'sid'. The two identify the same series:
+        // the alias resolves to its codename's dist for the lookup.
+        let resolved = match crate::distro_info::get_dist_from_series(&current).await {
+            Ok(dist) => Some((dist, current.clone())),
+            Err(_) => crate::distro_info::resolve_suite_alias(&current),
+        };
+        match resolved {
+            Some((dist, canonical)) => {
+                match crate::distro_info::get_ordered_series_name(&dist).await {
                     // Even an empty list goes through the selector: its
                     // fallback prints and takes the default, like it always has
-                Ok(options) => Some(SeriesCandidates::Choose {
+                    Ok(series_list) => {
+                        let (options, values, default) =
+                            selector_options(&dist, &series_list, &canonical);
+                        Some(SeriesCandidates::Choose {
                             options,
-                    default: current.clone(),
+                            values,
+                            // A stale alias whose codename left the series
+                            // list offers the raw name instead
+                            default: default.unwrap_or_else(|| canonical.clone()),
                             fallback: current,
-                }),
+                        })
-                Err(_) => Some(SeriesCandidates::Keep(current)),
+                    }
-            },
                     Err(_) => Some(SeriesCandidates::Keep(current)),
                 }
             }
+            None => Some(SeriesCandidates::Keep(current)),
+        }
+    }
+}
+
+/// The selector entries for a dist's series list: (labels, changelog
+/// targets, label of `current`'s entry). A series aliased by a
+/// changelog suite name (Debian's 'unstable' for 'sid') is offered as
+/// `<suite> (<series>)` but targets the suite name — what a changelog
+/// distribution field expects — while every other series is offered,
+/// and targeted, under its own name. `current` may be a name that is
+/// not in the list at all (e.g. UNRELEASED), in which case no default
+/// is returned.
+fn selector_options(
+    dist: &str,
+    series: &[String],
+    current: &str,
+) -> (Vec<String>, Vec<String>, Option<String>) {
+    let mut labels = Vec::with_capacity(series.len());
+    let mut values = Vec::with_capacity(series.len());
+    let mut default = None;
+    for s in series {
+        let (label, value) = match crate::distro_info::series_suite_alias(dist, s) {
+            Some(suite) => (format!("{suite} ({s})"), suite),
+            None => (s.clone(), s.clone()),
+        };
+        if s == current {
+            default = Some(label.clone());
+        }
+        labels.push(label);
+        values.push(value);
+    }
+    (labels, values, default)
+}
+
+/// The changelog distribution a selected series-selector label maps to
+/// ([`SeriesCandidates::Choose`]): an aliased entry ('unstable (sid)')
+/// targets its suite name, any other label targets itself, and a
+/// free-typed series the selector does not offer is its own target.
+pub fn selected_series(options: &[String], values: &[String], selected: String) -> String {
+    options
+        .iter()
+        .position(|o| *o == selected)
+        .and_then(|idx| values.get(idx).cloned())
+        .unwrap_or(selected)
 }
 
 /// Parse a changelog file footer to extract maintainer information
@@ -639,12 +711,18 @@ mod tests {
         match series_candidates(&path).await {
             Some(SeriesCandidates::Choose {
                 options,
+                values,
                 default,
                 fallback,
             }) => {
                 assert_eq!(options[0], "UNRELEASED");
+                assert_eq!(values[0], "UNRELEASED");
                 assert!(options.len() > 1, "the vendor series list is offered");
                 assert_eq!(default, options[1]);
+                assert_eq!(
+                    selected_series(&options, &values, default.clone()),
+                    values[1]
+                );
                 assert_eq!(fallback, "UNRELEASED");
             }
             other => panic!("expected Choose, got {other:?}"),
@@ -677,17 +755,105 @@ mod tests {
         match series_candidates(&path).await {
             Some(SeriesCandidates::Choose {
                 options,
+                values,
                 default,
                 fallback,
             }) => {
-                assert_eq!(options, vendor_series);
+                // The current series is preselected through its label, and
-                assert_eq!(default, *current);
+                // selecting it targets the name the changelog already carries
+                let idx = values
+                    .iter()
+                    .position(|v| v == current)
+                    .expect("the current series is offered");
+                assert_eq!(default, options[idx]);
+                assert_eq!(
+                    selected_series(&options, &values, options[idx].clone()),
+                    *current
+                );
                 assert_eq!(fallback, *current);
             }
             other => panic!("expected Choose, got {other:?}"),
         }
     }
 
+    /// A changelog targeting Debian's 'unstable' suite — the conventional
+    /// Debian development distribution, which the series data knows as the
+    /// codename 'sid' — resolves to the Debian series list: the selector
+    /// offers the aliased entry as 'unstable (sid)', preselected, and
+    /// selecting it targets 'unstable' itself.
+    #[tokio::test]
+    async fn series_candidates_suite_alias_matches_unstable_and_sid() {
+        let dir = tempfile::tempdir().unwrap();
+        let path = dir.path().join("changelog");
+        std::fs::write(
+            &path,
+            "hello (1.0-1) unstable; urgency=medium\n\n  * Something.\n\n \
+             -- A B <a@b.c>  Mon, 01 Jan 2024 00:00:00 +0000\n",
+        )
+        .unwrap();
+
+        match series_candidates(&path).await {
+            Some(SeriesCandidates::Choose {
+                options,
+                values,
+                default,
+                fallback,
+            }) => {
+                let idx = options
+                    .iter()
+                    .position(|o| o == "unstable (sid)")
+                    .expect("sid is offered as its suite alias");
+                assert_eq!(values[idx], "unstable");
+                assert_eq!(default, "unstable (sid)");
+                assert_eq!(fallback, "unstable");
+                // Selecting the aliased entry targets the suite name
+                assert_eq!(
+                    selected_series(&options, &values, options[idx].clone()),
+                    "unstable"
+                );
+                // A free-typed series the selector does not offer is its own
+                // target
+                assert_eq!(
+                    selected_series(&options, &values, "trixie".to_string()),
+                    "trixie"
+                );
+            }
+            other => panic!("expected Choose, got {other:?}"),
+        }
+    }
+
+    /// A changelog naming the codename ('sid') resolves to the same
+    /// selector entry as the suite alias ('unstable'): the two identify
+    /// the same series.
+    #[tokio::test]
+    async fn series_candidates_sid_defaults_to_the_suite_alias_label() {
+        let dir = tempfile::tempdir().unwrap();
+        let path = dir.path().join("changelog");
+        std::fs::write(
+            &path,
+            "hello (1.0-1) sid; urgency=medium\n\n  * Something.\n\n \
+             -- A B <a@b.c>  Mon, 01 Jan 2024 00:00:00 +0000\n",
+        )
+        .unwrap();
+
+        match series_candidates(&path).await {
+            Some(SeriesCandidates::Choose {
+                options,
+                values,
+                default,
+                fallback,
+            }) => {
+                assert_eq!(default, "unstable (sid)");
+                assert_eq!(
+                    selected_series(&options, &values, default.clone()),
+                    "unstable"
+                );
+                assert_eq!(fallback, "sid");
+            }
+            other => panic!("expected Choose, got {other:?}"),
+        }
+    }
+
     /// Without a parsable changelog there is no candidate at all.
     #[tokio::test]
     async fn series_candidates_none_without_changelog() {

src/distro_info.rs

@@ -87,6 +87,11 @@ struct DistData {
     cross_pockets: Vec<String>,
     #[serde(default)]
     build_profiles: Vec<String>,
+    /// Changelog suite names aliasing a distro-info series codename
+    /// ('unstable' for Debian's 'sid'): the two names identify the same
+    /// series ([`series_suite_alias`], [`resolve_suite_alias`])
+    #[serde(default)]
+    suite_aliases: HashMap<String, String>,
     series: SeriesInfo,
 }
 
@@ -413,6 +418,30 @@ pub async fn get_dist_from_series(series: &str) -> Result<String, Box<dyn Error>
     Err(format!("Unknown series: {}", series).into())
 }
 
+/// The changelog suite name that aliases the series codename of `dist`
+/// (Debian's 'unstable' for 'sid'): the two names identify the same
+/// series. `None` when the series carries no suite alias.
+pub fn series_suite_alias(dist: &str, series: &str) -> Option<String> {
+    dist_data(dist)
+        .ok()?
+        .suite_aliases
+        .iter()
+        .find(|(_suite, codename)| codename.as_str() == series)
+        .map(|(suite, _)| suite.clone())
+}
+
+/// Identify a changelog suite name with the distro-info series codename
+/// it aliases (Debian's 'unstable' is 'sid'), and the dist that codename
+/// belongs to. `None` when `suite` is not a known alias of any dist.
+pub fn resolve_suite_alias(suite: &str) -> Option<(String, String)> {
+    for (dist, data) in DATA.dist.iter() {
+        if let Some(codename) = data.suite_aliases.get(suite) {
+            return Some((dist.clone(), codename.clone()));
+        }
+    }
+    None
+}
+
 /// Get the package pockets available for a given distribution, in search order
 ///
 /// The main archive ('') comes first so that a search without an explicit
@@ -1035,6 +1064,41 @@ mod tests {
         assert!(series.contains(&"jammy".to_string()));
     }
 
+    /// Suite aliases identify a changelog suite name with the series
+    /// codename of the same series: Debian's 'unstable' is 'sid'
+    #[test]
+    fn test_suite_aliases() {
+        assert_eq!(
+            resolve_suite_alias("unstable"),
+            Some(("debian".to_string(), "sid".to_string()))
+        );
+        // A series codename or unknown suite is not an alias
+        assert_eq!(resolve_suite_alias("sid"), None);
+        assert_eq!(resolve_suite_alias("noble"), None);
+        assert_eq!(
+            series_suite_alias("debian", "sid"),
+            Some("unstable".to_string())
+        );
+        assert_eq!(series_suite_alias("debian", "trixie"), None);
+        assert_eq!(series_suite_alias("ubuntu", "noble"), None);
+    }
+
+    /// Every suite alias must map to a real series of its dist, or the
+    /// selector would offer a phantom entry
+    #[tokio::test]
+    async fn test_suite_aliases_target_real_series() {
+        for (dist, data) in DATA.dist.iter() {
+            for (suite, codename) in &data.suite_aliases {
+                let series = get_ordered_series_name(dist).await.unwrap_or_default();
+                assert!(
+                    series.contains(codename),
+                    "suite alias '{suite}' of {dist} maps to '{codename}', \
+                     which is not a known series"
+                );
+            }
+        }
+    }
+
     #[tokio::test]
     async fn test_get_dist_from_series() {
         assert_eq!(get_dist_from_series("sid").await.unwrap(), "debian");

src/launchpad.rs

@@ -41,6 +41,11 @@ struct LaunchpadData {
     ssh_host: String,
     /// Port of the PPA SFTP upload server
     ssh_port: u16,
+    /// Host of the PPA upload queue over anonymous FTP (the transport
+    /// `pkh put` degrades to when the SSH connection never comes up)
+    ftp_host: String,
+    /// Port of the anonymous FTP upload queue
+    ftp_port: u16,
     /// Upload queue incoming directory template (`{owner}`/`{ppa}`)
     incoming_template: String,
     /// PPA package-content (apt repository) URL template
@@ -53,6 +58,13 @@ embed_data! {
     static ref LAUNCHPAD_DATA: LaunchpadData = "../data/launchpad.yml"
 }
 
+/// The PPA upload queue over anonymous FTP (host, port): the transport
+/// dput-ng's plain `ppa:` profile pushes over, and the one `pkh put`
+/// degrades to when the SSH connection itself never comes up.
+pub(crate) fn ppa_ftp_queue() -> (String, u16) {
+    (LAUNCHPAD_DATA.ftp_host.clone(), LAUNCHPAD_DATA.ftp_port)
+}
+
 /// Base URL of the Launchpad REST API
 fn api_base() -> &'static str {
     &LAUNCHPAD_DATA.api_base
@@ -422,6 +434,13 @@ mod tests {
         assert_eq!(target.login, None);
     }
 
+    /// The anonymous FTP fallback queue resolves from the same data the
+    /// dput-ng `ppa:` profile uses.
+    #[test]
+    fn ppa_ftp_queue_resolves() {
+        assert_eq!(ppa_ftp_queue(), ("ppa.launchpad.net".to_string(), 21));
+    }
+
     #[test]
     fn ppa_target_rejects_missing_separator() {
         assert!(ppa_target("just-a-name").is_err());

src/main.rs

@@ -31,7 +31,6 @@ fn main() {
     LogWrapper::new(multi.clone(), logger).try_init().unwrap();
     let matches = command!()
         .subcommand_required(true)
-        .disable_version_flag(true)
         .subcommand(
             Command::new("new")
                 .about("Scaffold a new Debian source package (buildable right away)")
@@ -446,10 +445,13 @@ fn main() {
                 match rt.block_on(pkh::changelog::series_candidates(&changelog_path)) {
                     Some(pkh::changelog::SeriesCandidates::Choose {
                         options,
+                        values,
                         default,
                         fallback,
                     }) => match pkh::ui::select_series(&options, &default) {
-                        Ok(selected) => Some(selected),
+                        Ok(selected) => {
+                            Some(pkh::changelog::selected_series(&options, &values, selected))
+                        }
                         Err(e) => {
                             error!(
                                 "Series selection failed: {}. Using current series '{}' instead.",

src/pull.rs

@@ -452,15 +452,18 @@ async fn fetch_orig_tarball(
         Path::new(&info.stanza.package).to_path_buf()
     };
 
-    // Find the orig tarball in the file list
+    // Upstream tarballs in the file list: the main orig tarball plus, for
-    // Usually ends with .orig.tar.gz or .orig.tar.xz
+    // multi-orig ("3.0 (quilt)" extra-component) sources, one component
-    let orig_file = info
+    // tarball per bundled module (`*.orig-<component>.tar.<ext>`). dpkg-source
+    // unpacks all of them side by side, so a git pull must fetch them all.
+    let orig_files: Vec<_> = info
         .stanza
         .files
         .iter()
-        .find(|f| f.name.contains(".orig.tar."))
+        .filter(|f| crate::build::changes::is_orig_tarball(&f.name))
-        .ok_or_else(|| {
+        .collect();
-            format!(
+    if orig_files.is_empty() {
+        return Err(format!(
             "Could not find orig tarball in file list for package '{}'. \
              Available files: {:?}",
             info.stanza.package,
@@ -470,14 +473,17 @@ async fn fetch_orig_tarball(
                 .map(|f| &f.name)
                 .collect::<Vec<_>>()
         )
-        })?;
+        .into());
-    let filename = &orig_file.name;
+    }
 
     // 1. Try executing pristine-tar
 
     // Setup pristine-tar branch if needed (by tracking remote branch)
     let _ = setup_pristine_tar_branch(&package_dir, info.dist.as_str());
 
+    for orig_file in orig_files {
+        let filename = &orig_file.name;
+
         if let Err(e) = checkout_pristine_tar(&package_dir, filename.as_str()) {
             debug!(
                 "pristine-tar failed: {}. Falling back to archive download.",
@@ -497,6 +503,7 @@ async fn fetch_orig_tarball(
             )
             .await?;
         }
+    }
 
     Ok(())
 }
@@ -844,23 +851,29 @@ mod tests {
             }
         }
 
-        // Check for orig tarball in package dir (only for non-native packages)
+        // Check for the orig tarballs in the package dir (only for non-native
-        let mut found_tarball = false;
+        // packages): every orig listed in the stanza must be present, including
+        // the component tarballs of multi-orig packages (dpkg-source needs them
+        // all to unpack the merged upstream tree)
         let mut found_dsc = false;
-        for entry in std::fs::read_dir(package_dir).unwrap() {
+        for entry in std::fs::read_dir(&package_dir).unwrap() {
             let entry = entry.unwrap();
             let name = entry.file_name().to_string_lossy().to_string();
-            if name.contains(".orig.tar.") {
-                found_tarball = true;
-            }
             if name.ends_with(".dsc") {
                 found_dsc = true;
             }
         }
 
-        // Only check for orig tarball if the package is not native
         if !info.is_native() {
-            assert!(found_tarball, "Orig tarball not found in package dir");
+            for file in &info.stanza.files {
+                if crate::build::changes::is_orig_tarball(&file.name) {
+                    assert!(
+                        package_dir.join(&file.name).exists(),
+                        "Orig tarball '{}' not found in package dir",
+                        file.name
+                    );
+                }
+            }
         }
         assert!(found_dsc, "DSC file not found in package dir");
     }
@@ -914,6 +927,17 @@ mod tests {
         test_pull_package_end_to_end("paraview", Some("noble"), None, None).await;
     }
 
+    /// Multi-orig ("3.0 (quilt)" extra component) regression test: node-jest
+    /// ships its bundled modules as separate `*.orig-<component>.tar.xz`
+    /// tarballs next to the main orig. The git pull path must fetch every
+    /// component, or the later `dpkg-source -b` quilt verification fails
+    /// with "can't find file to patch" on the first patch touching a
+    /// component directory.
+    #[tokio::test]
+    async fn test_pull_node_jest_debian_end_to_end() {
+        test_pull_package_end_to_end("node-jest", Some("trixie"), None, None).await;
+    }
+
     /// Build a minimal uncompressed ustar archive from (name, data) entries.
     ///
     /// Raw header blocks are crafted instead of using `tar::Builder` because

src/put/ftp.rs

@@ -0,0 +1,456 @@
+//! Anonymous FTP transport for the Launchpad PPA upload queue: the graceful
+//! degradation of the SFTP transport when the SSH connection itself never
+//! comes up (name resolution, TCP, banner or key exchange). dput-ng's plain
+//! `ppa:user/ppa` profile pushes over this same queue — ppa.launchpad.net
+//! over FTP, anonymous login, incoming `~user/ppa` — so it is the
+//! interoperability-tested path.
+//!
+//! The client is [`suppaftp`]'s (plain-FTP, no TLS) blocking stream on the
+//! same time-bounded sockets as the ssh2 transport: the TCP connect and
+//! the control/data channel reads and writes all carry timeouts, so a
+//! black-holed or stalled server fails the upload instead of hanging it
+//! (suppaftp's defaults do not bound them). The upload order (payload
+//! first, `.changes` last — the caller passes the files in that order) and
+//! the best-effort cleanup of a failed upload (`DELE` of what was already
+//! pushed, in reverse upload order) mirror the SFTP path exactly.
+
+use std::fs::File;
+use std::io::{Read, Write};
+use std::net::{SocketAddr, TcpStream};
+use std::path::{Path, PathBuf};
+use std::time::Duration;
+
+use suppaftp::types::FileType;
+use suppaftp::{FtpError, FtpStream};
+
+use super::ssh;
+
+/// Bounds one TCP connection attempt, to the control channel or a passive
+/// data port: `connect(2)` would otherwise block for minutes (or forever,
+/// behind a silent firewall). Generous enough for slow links to Launchpad,
+/// short enough that a dead target fails in seconds — the same value and
+/// rationale as the SSH path's bound.
+const TCP_CONNECT_TIMEOUT: Duration = Duration::from_secs(15);
+
+/// Read/write timeout on the control channel: every reply is a few bytes,
+/// so a stalled server has had its say by the time this expires.
+const CONTROL_TIMEOUT: Duration = Duration::from_secs(60);
+
+/// Write timeout of one data-transfer chunk: the clock restarts at every
+/// write, so the wall-clock duration of a large upload is not bounded by
+/// design — but a black-holed data connection fails one chunk after this
+/// budget.
+const DATA_TIMEOUT: Duration = Duration::from_secs(120);
+
+/// Data transfers stream in chunks of this size.
+const CHUNK_SIZE: usize = 32 * 1024;
+
+/// Upload `files` to the `incoming` queue of `host:port` over anonymous FTP,
+/// in the order given (payload first, `.changes` last — the queue processor
+/// must never observe a `.changes` without its payload).
+/// `on_progress(name, uploaded_bytes, total_bytes)` reports progress.
+///
+/// A failure mid-upload best-effort removes what was already pushed (`DELE`,
+/// reverse upload order — a lingering payload in the write-only queue area
+/// is only hygiene) before returning the original error, like the SFTP
+/// path's [`super::cleanup_partial_upload`].
+pub fn upload_queue(
+    host: &str,
+    port: u16,
+    incoming: &str,
+    files: &[(PathBuf, String)],
+    on_progress: &dyn Fn(&str, u64, u64),
+) -> Result<(), Box<dyn std::error::Error>> {
+    let mut ftp = connect(host, port)?;
+    let who = std::env::var("USER").unwrap_or_else(|_| "anonymous".to_string());
+    ftp.login("anonymous".to_string(), format!("{who}@pkh.invalid"))
+        .map_err(|e| format!("the FTP queue rejected the anonymous login: {e}"))?;
+    ftp.cwd(incoming)
+        .map_err(|e| format!("cannot enter the upload queue '{incoming}': {e}"))?;
+    ftp.transfer_type(FileType::Binary)
+        .map_err(|e| format!("the FTP queue refused binary transfers: {e}"))?;
+
+    // Remote names pushed so far, in upload order, for the cleanup
+    let mut uploaded: Vec<String> = Vec::new();
+    for (path, name) in files {
+        if let Err(e) = store(&mut ftp, path, name, on_progress) {
+            // The failed file itself joins the cleanup: its `STOR` was
+            // accepted before the transfer failure, so a partial may be
+            // sitting in the queue
+            for leftover in super::cleanup_list(&uploaded, Some(name)) {
+                match ftp.rm(leftover.as_str()) {
+                    Ok(()) => {
+                        log::info!("Removed leftover {leftover} from the failed upload")
+                    }
+                    Err(e) => log::warn!(
+                        "Could not remove the leftover {leftover} of the \
+                         failed upload: {e}"
+                    ),
+                }
+            }
+            return Err(e);
+        }
+        uploaded.push(name.clone());
+    }
+
+    // Best effort: the queue keeps what it accepted, so a failure here must
+    // not fail a completed upload
+    if let Err(e) = ftp.quit() {
+        log::debug!("closing the FTP session: {e}");
+    }
+    Ok(())
+}
+
+/// Connect to the queue, read its banner and set every time bound and
+/// workaround the plain suppaftp stream does not carry by itself.
+fn connect(host: &str, port: u16) -> Result<FtpStream, Box<dyn std::error::Error>> {
+    let tcp = ssh::tcp_connect(host, port)?;
+    let mut ftp = FtpStream::connect_with_stream(tcp)?.passive_stream_builder(data_connect);
+    {
+        let control = ftp.get_ref();
+        control.set_read_timeout(Some(CONTROL_TIMEOUT))?;
+        control.set_write_timeout(Some(CONTROL_TIMEOUT))?;
+    }
+    // A `PASV` reply announcing an unroutable address (a server behind NAT
+    // that does not know its public IP) means the control connection's peer
+    ftp.set_passive_nat_workaround(true);
+    Ok(ftp)
+}
+
+/// The passive data-channel connect: bounded like every other network call,
+/// where suppaftp's default builder is a plain, unbounded
+/// `TcpStream::connect`.
+fn data_connect(addr: SocketAddr) -> Result<TcpStream, FtpError> {
+    TcpStream::connect_timeout(&addr, TCP_CONNECT_TIMEOUT).map_err(FtpError::ConnectionError)
+}
+
+/// Upload `path` as `name` over a passive data connection, reporting
+/// progress through `on_progress` per chunk. `finish` reads the transfer
+/// completion reply — the only way to learn the server accepted the file.
+fn store(
+    ftp: &mut FtpStream,
+    path: &Path,
+    name: &str,
+    on_progress: &dyn Fn(&str, u64, u64),
+) -> Result<(), Box<dyn std::error::Error>> {
+    let mut upload = ftp
+        .put_with_stream(name)
+        .map_err(|e| format!("the FTP queue rejected '{name}': {e}"))?;
+    if let suppaftp::DataStream::Tcp(socket) = upload.get_mut() {
+        socket.set_write_timeout(Some(DATA_TIMEOUT))?;
+    }
+
+    let mut file =
+        File::open(path).map_err(|e| format!("cannot read '{}': {e}", path.display()))?;
+    let total = file.metadata().map(|m| m.len()).unwrap_or(0);
+    let mut buffer = vec![0u8; CHUNK_SIZE];
+    let mut uploaded: u64 = 0;
+    loop {
+        let read = file.read(&mut buffer)?;
+        if read == 0 {
+            break;
+        }
+        upload.write_all(&buffer[..read])?;
+        uploaded += read as u64;
+        on_progress(name, uploaded, total);
+    }
+    upload
+        .finish()
+        .map_err(|e| format!("upload of '{name}' failed: {e}"))?;
+    Ok(())
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::collections::HashMap;
+    use std::net::TcpListener;
+    use std::sync::Arc;
+    use std::sync::atomic::{AtomicUsize, Ordering};
+
+    use parking_lot::Mutex;
+
+    /// In-process fake of the Launchpad FTP upload queue: one control
+    /// session speaking the protocol subset the client uses (banner,
+    /// USER/PASS, CWD, TYPE, PASV, STOR, DELE, QUIT). Records every
+    /// command, stores received bytes under the `STOR` name, and can be
+    /// told to reject one `STOR` (by index) to exercise the cleanup. Its
+    /// `PASV` replies announce `0.0.0.0`, the NAT form, so the
+    /// happy-path test proves the control-peer fallback too.
+    struct FakeQueue {
+        addr: SocketAddr,
+        commands: Arc<Mutex<Vec<String>>>,
+        files: Arc<Mutex<HashMap<String, Vec<u8>>>>,
+    }
+
+    impl FakeQueue {
+        /// Serve one session on a background thread, greeting it with
+        /// `banner` and rejecting the `STOR` number `reject_stor`, when
+        /// set.
+        fn start(banner: &str, reject_stor: Option<usize>) -> FakeQueue {
+            let listener = TcpListener::bind("127.0.0.1:0").unwrap();
+            let addr = listener.local_addr().unwrap();
+            let commands = Arc::new(Mutex::new(Vec::new()));
+            let files = Arc::new(Mutex::new(HashMap::new()));
+            let (c, f) = (Arc::clone(&commands), Arc::clone(&files));
+            let banner = banner.to_string();
+            std::thread::spawn(move || serve(listener, &banner, c, f, reject_stor));
+            FakeQueue {
+                addr,
+                commands,
+                files,
+            }
+        }
+    }
+
+    fn serve(
+        listener: TcpListener,
+        banner: &str,
+        commands: Arc<Mutex<Vec<String>>>,
+        files: Arc<Mutex<HashMap<String, Vec<u8>>>>,
+        reject_stor: Option<usize>,
+    ) {
+        use std::io::BufRead;
+
+        let (stream, _) = listener.accept().unwrap();
+        let mut reader = std::io::BufReader::new(stream.try_clone().unwrap());
+        let mut writer = stream;
+        writer
+            .write_all(format!("{banner}\r\n").as_bytes())
+            .unwrap();
+
+        let mut pending_data: Option<TcpListener> = None;
+        let stor_index = AtomicUsize::new(0);
+        loop {
+            let mut line = String::new();
+            if reader.read_line(&mut line).unwrap_or(0) == 0 {
+                break;
+            }
+            let cmd = line.trim_end().to_string();
+            commands.lock().push(cmd.clone());
+            let (verb, arg) = cmd.split_once(' ').unwrap_or((cmd.as_str(), ""));
+            match verb {
+                "USER" | "PASS" => {
+                    writer
+                        .write_all(b"230 Anonymous login ok, access restrictions apply.\r\n")
+                        .unwrap();
+                }
+                "CWD" => {
+                    writer.write_all(b"250 Command successful\r\n").unwrap();
+                }
+                "TYPE" => {
+                    writer.write_all(b"200 Type set to I\r\n").unwrap();
+                }
+                "PASV" => {
+                    let data = TcpListener::bind("127.0.0.1:0").unwrap();
+                    let port = data.local_addr().unwrap().port();
+                    writer
+                        .write_all(
+                            format!(
+                                "227 Entering Passive Mode (0,0,0,0,{},{})\r\n",
+                                port / 256,
+                                port % 256
+                            )
+                            .as_bytes(),
+                        )
+                        .unwrap();
+                    pending_data = Some(data);
+                }
+                "STOR" => {
+                    let index = stor_index.fetch_add(1, Ordering::SeqCst);
+                    if Some(index) == reject_stor {
+                        writer.write_all(b"550 Permission denied\r\n").unwrap();
+                        continue;
+                    }
+                    writer.write_all(b"150 Ok to send data\r\n").unwrap();
+                    let (mut data, _) = pending_data.take().unwrap().accept().unwrap();
+                    let mut bytes = Vec::new();
+                    data.read_to_end(&mut bytes).unwrap();
+                    files.lock().insert(arg.to_string(), bytes);
+                    writer.write_all(b"226 Transfer complete\r\n").unwrap();
+                }
+                "DELE" => {
+                    if files.lock().remove(arg).is_some() {
+                        writer.write_all(b"250 File deleted\r\n").unwrap();
+                    } else {
+                        writer.write_all(b"550 No such file\r\n").unwrap();
+                    }
+                }
+                "QUIT" => {
+                    writer.write_all(b"221 Bye\r\n").unwrap();
+                    break;
+                }
+                other => {
+                    writer
+                        .write_all(format!("502 Command '{other}' not implemented\r\n").as_bytes())
+                        .unwrap();
+                }
+            }
+        }
+    }
+
+    /// A temp file with `content`, to upload.
+    fn upload_file(dir: &Path, name: &str, content: &[u8]) -> (PathBuf, String) {
+        let path = dir.join(name);
+        std::fs::write(&path, content).unwrap();
+        (path, name.to_string())
+    }
+
+    /// The full push: anonymous login, the queue directory entered,
+    /// payload files uploaded in order before the `.changes`, their bytes
+    /// intact, and progress reported up to each file's size.
+    #[test]
+    fn upload_queue_pushes_payload_before_changes() {
+        let queue = FakeQueue::start("220 Launchpad upload server", None);
+        let dir = tempfile::tempdir().unwrap();
+        let files = vec![
+            upload_file(dir.path(), "pkg_1.0.orig.tar.xz", b"orig bytes"),
+            upload_file(dir.path(), "pkg_1.0-1.dsc", b"dsc bytes"),
+            upload_file(dir.path(), "pkg_1.0-1_source.changes", b"changes bytes"),
+        ];
+        let seen = Arc::new(Mutex::new(Vec::new()));
+        let progress_log = Arc::clone(&seen);
+
+        upload_queue(
+            "127.0.0.1",
+            queue.addr.port(),
+            "~vhaudiquet/lp2167827",
+            &files,
+            &|name, uploaded, total| {
+                progress_log
+                    .lock()
+                    .push((name.to_string(), uploaded, total))
+            },
+        )
+        .unwrap();
+
+        let commands = queue.commands.lock().clone();
+        assert_eq!(
+            commands[0], "USER anonymous",
+            "the anonymous login comes first"
+        );
+        assert_eq!(
+            commands
+                .iter()
+                .filter(|c| c == &"CWD ~vhaudiquet/lp2167827")
+                .count(),
+            1
+        );
+        let cwd = commands.iter().position(|c| c.starts_with("CWD")).unwrap();
+        let first_stor = commands.iter().position(|c| c.starts_with("STOR")).unwrap();
+        assert!(cwd < first_stor, "the queue directory is entered first");
+        assert!(
+            commands.iter().any(|c| c == "TYPE I"),
+            "binary transfers are requested"
+        );
+        // Upload order: the payload first, the .changes last
+        let stors: Vec<&String> = commands.iter().filter(|c| c.starts_with("STOR")).collect();
+        assert_eq!(
+            stors.iter().map(|s| s.as_str()).collect::<Vec<_>>(),
+            vec![
+                "STOR pkg_1.0.orig.tar.xz",
+                "STOR pkg_1.0-1.dsc",
+                "STOR pkg_1.0-1_source.changes",
+            ]
+        );
+        assert_eq!(*commands.last().unwrap(), "QUIT");
+
+        // The received bytes are intact
+        let stored = queue.files.lock();
+        assert_eq!(stored.get("pkg_1.0.orig.tar.xz").unwrap(), b"orig bytes");
+        assert_eq!(stored.get("pkg_1.0-1.dsc").unwrap(), b"dsc bytes");
+        assert_eq!(
+            stored.get("pkg_1.0-1_source.changes").unwrap(),
+            b"changes bytes"
+        );
+
+        // Progress reached each file's size
+        let progress = seen.lock();
+        for (path, name) in &files {
+            let size = path.metadata().unwrap().len();
+            let reached = progress
+                .iter()
+                .any(|(n, uploaded, total)| n == name && *uploaded == size && *total == size);
+            assert!(reached, "no progress report completed for '{name}'");
+        }
+    }
+
+    /// A `STOR` the queue rejects fails the upload, and the client removes
+    /// what it already pushed — the failed file first, then the earlier
+    /// payloads, in reverse upload order — before returning the error.
+    #[test]
+    fn upload_queue_cleans_up_after_a_rejected_stor() {
+        let queue = FakeQueue::start("220 Launchpad upload server", Some(1));
+        let dir = tempfile::tempdir().unwrap();
+        let files = vec![
+            upload_file(dir.path(), "one.dsc", b"one"),
+            upload_file(dir.path(), "two.tar.xz", b"two"),
+            upload_file(dir.path(), "three.changes", b"three"),
+        ];
+
+        let err = upload_queue(
+            "127.0.0.1",
+            queue.addr.port(),
+            "~user/ppa",
+            &files,
+            &|_, _, _| {},
+        )
+        .unwrap_err();
+        assert!(
+            err.to_string().contains("rejected 'two.tar.xz'"),
+            "the error names the rejected file, got: {err}"
+        );
+
+        let commands = queue.commands.lock();
+        let deles: Vec<&String> = commands.iter().filter(|c| c.starts_with("DELE")).collect();
+        assert_eq!(
+            deles.iter().map(|d| d.as_str()).collect::<Vec<_>>(),
+            vec!["DELE two.tar.xz", "DELE one.dsc"],
+            "the failed file is removed first, then the earlier payloads"
+        );
+        let files = queue.files.lock();
+        assert!(
+            !files.contains_key("one.dsc"),
+            "the pushed payload is removed"
+        );
+        assert!(!files.contains_key("three.changes"), "never reached");
+    }
+
+    /// A multi-line banner (`220-first` closed by `220 last`) parses like
+    /// the single-line form.
+    #[test]
+    fn upload_queue_reads_multiline_replies() {
+        let queue = FakeQueue::start("220-Launchpad\r\n220 upload server", None);
+        let dir = tempfile::tempdir().unwrap();
+        let files = vec![upload_file(dir.path(), "pkg.dsc", b"bytes")];
+
+        upload_queue(
+            "127.0.0.1",
+            queue.addr.port(),
+            "~user/ppa",
+            &files,
+            &|_, _, _| {},
+        )
+        .unwrap();
+
+        let stored = queue.files.lock();
+        assert_eq!(stored.get("pkg.dsc").unwrap(), b"bytes");
+    }
+
+    /// Live control-channel handshake with the real Launchpad FTP queue:
+    /// banner, anonymous login and a `CWD` — nothing is uploaded, the
+    /// queue is left untouched. For deliberate ad-hoc runs
+    /// (`cargo test -- --ignored`), not the pre-commit pass: it hits the
+    /// network.
+    #[test]
+    #[ignore = "hits the network: the real Launchpad FTP queue"]
+    fn live_launchpad_control_channel() {
+        let (host, port) = crate::launchpad::ppa_ftp_queue();
+        assert_eq!((host.as_str(), port), ("ppa.launchpad.net", 21));
+
+        let mut ftp = connect(&host, port).unwrap();
+        ftp.login("anonymous", "pkh@invalid").unwrap();
+        ftp.cwd("/").unwrap();
+        ftp.quit().unwrap();
+    }
+}

src/put/mod.rs

@@ -1,17 +1,12 @@
-//! Native upload of built source packages (`pkh put`): the dput
+//! PPA upload: the dput replacement. Resolves the upload target, discovers
-//! replacement. Resolves the upload target, discovers and validates the
+//! and validates the `.changes` file and its artifacts, then pushes them
-//! `.changes` file and its artifacts, then pushes them over SFTP with
+//! over SFTP with host-key verification and an upload record preventing
-//! host-key verification and an upload record preventing accidental
+//! accidental duplicate uploads. When the SSH connection itself never
-//! duplicate uploads.
+//! comes up, the upload degrades to the anonymous FTP queue — the
-//!
+//! transport dput's plain `ppa:` profiles use — through [`ftp`].
-//! 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. 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 ftp;
 pub mod ssh;
 pub mod target;
 
@@ -139,11 +134,6 @@ async fn put_impl(opts: &PutOptions<'_>) -> Result<(), Box<dyn std::error::Error
         .into());
     }
 
-    opts.view
-        .message(&format!("Connecting to {login}@{host}:{port}..."));
-    let session = ssh::connect(&host, port, &login, &ssh_config, opts.prompter)?;
-    let sftp = ssh::sftp(&session)?;
-
     // Payload first, the .changes file last (like dput), so the server-side
     // queue processor can never pick up an incomplete upload
     let dir = changes_path.parent().unwrap_or_else(|| Path::new("."));
@@ -157,48 +147,50 @@ async fn put_impl(opts: &PutOptions<'_>) -> Result<(), Box<dyn std::error::Error
         .and_then(|n| n.to_str())
         .ok_or_else(|| format!("invalid .changes path: {}", changes_path.display()))?
         .to_string();
-    uploads.push((changes_path.clone(), changes_name.clone()));
+    uploads.push((changes_path.clone(), changes_name));
 
     let incoming = target.incoming.trim_end_matches('/');
 
-    // Remote names of the files uploaded so far, in upload order (the
+    // The upload queue is SFTP first, degrading to the anonymous FTP
-    // .changes last). A run failing mid-upload removes these from the
+    // queue when the SSH connection itself never comes up (name
-    // write-only incoming queue before returning: the uploaded payloads
+    // resolution, TCP, banner or key exchange): dput pushes PPAs over
-    // would otherwise linger in the queue area forever, and a .changes
+    // that FTP queue by default, so it is the interoperability-tested
-    // truncated by a failed close could even be picked up by the scanner.
+    // fallback. A server that answers but refuses the upload (host key
-    let mut uploaded: Vec<String> = Vec::new();
+    // not accepted, no matching key) stays an error: silently switching
-
+    // transport would bypass the refusal.
-    for (path, name) in &uploads {
+    opts.view
-        let size = match path.metadata() {
+        .message(&format!("Connecting to {login}@{host}:{port}..."));
-            Ok(metadata) => metadata.len(),
+    let transfer = match ssh::connect(&host, port, &login, &ssh_config, opts.prompter) {
-            Err(e) => {
+        Ok(session) => sftp_transfer(&session, &uploads, incoming, &host, opts.view),
-                // Nothing was attempted for this file: only what earlier
+        Err(ssh::ConnectFailure::Transport(e)) => {
-                // iterations uploaded needs removing
+            log::warn!("SSH transport to {host}:{port} failed: {e}");
-                let error: Box<dyn std::error::Error> =
+            let (ftp_host, ftp_port) = launchpad::ppa_ftp_queue();
-                    format!("cannot stat '{}': {}", path.display(), e).into();
+            opts.view.message(&format!(
-                cleanup_partial_upload(&sftp, incoming, &uploaded, None, &host);
+                "Falling back to the anonymous FTP queue on {ftp_host}:{ftp_port} \
-                return Err(error);
+                 (dput's upload method)..."
+            ));
+            ftp::upload_queue(
+                &ftp_host,
+                ftp_port,
+                incoming,
+                &uploads,
+                &|name, uploaded, total| {
+                    opts.view.progress(
+                        &format!("Uploading {name}"),
+                        uploaded as usize,
+                        total as usize,
+                    );
+                },
+            )
         }
+        Err(ssh::ConnectFailure::Refused(e)) => return Err(e),
     };
-        let remote = format!("{incoming}/{name}");
+    transfer?;
-        let label = format!("Uploading {name}");
-        let view = opts.view;
-        let on_progress = |uploaded: u64| view.progress(&label, uploaded as usize, size as usize);
-        let result = ssh::upload_file(&sftp, path, &remote, &host, &on_progress);
-        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
+    // uploaded (a re-run replays every file — `sftp.create` truncates and
-    // replaying is safe).
+    // the FTP `STOR` overwrites, so replaying is safe).
     record_upload(&upload_log_path()?, &record)?;
 
     info!(
@@ -212,6 +204,54 @@ async fn put_impl(opts: &PutOptions<'_>) -> Result<(), Box<dyn std::error::Error
     Ok(())
 }
 
+/// Push `uploads` over SFTP: payload files first, the `.changes` last
+/// (like dput), so the server-side queue processor can never pick up an
+/// incomplete upload. Best-effort removal of a partial upload, mirroring
+/// the FTP transport's `DELE` cleanup ([`ftp::upload_queue`]).
+fn sftp_transfer(
+    session: &ssh2::Session,
+    uploads: &[(PathBuf, String)],
+    incoming: &str,
+    host: &str,
+    view: &dyn BuildView,
+) -> Result<(), Box<dyn std::error::Error>> {
+    let sftp = ssh::sftp(session)?;
+
+    // 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 = 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);
+            }
+        };
+        let remote = format!("{incoming}/{name}");
+        let label = format!("Uploading {name}");
+        let on_progress = |uploaded: u64| view.progress(&label, uploaded as usize, size as usize);
+        let result = ssh::upload_file(&sftp, path, &remote, host, &on_progress);
+        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());
+    }
+    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,
@@ -219,8 +259,9 @@ async fn put_impl(opts: &PutOptions<'_>) -> Result<(), Box<dyn std::error::Error
 /// 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`].
+/// without a server; the network sides are [`cleanup_partial_upload`] and
-fn cleanup_list(uploaded: &[String], failed: Option<&str>) -> Vec<String> {
+/// the FTP transport's `DELE` loop ([`ftp`]).
+pub(crate) 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());

src/put/ssh.rs

@@ -253,7 +253,9 @@ fn duration_ms(timeout: Duration) -> u32 {
 /// address in order (like `TcpStream::connect` does) with
 /// [`TCP_CONNECT_TIMEOUT`] per attempt instead of blocking indefinitely.
 /// Fails with a message naming the target and every per-address error.
-fn tcp_connect(host: &str, port: u16) -> Result<TcpStream, String> {
+/// Also the TCP layer of the anonymous FTP fallback transport
+/// ([`super::ftp`]), whose connection semantics are identical.
+pub(crate) fn tcp_connect(host: &str, port: u16) -> Result<TcpStream, String> {
     let addrs: Vec<SocketAddr> = (host, port)
         .to_socket_addrs()
         .map_err(|e| format!("cannot resolve {host}:{port}: {e}"))?
@@ -292,6 +294,39 @@ fn connect_failed_message(
         attempts.len()
     )
 }
+/// Why establishing the SSH session failed.
+///
+/// [`Transport`] failures mean the connection never came up (name
+/// resolution, TCP, banner or key exchange): the target may still be
+/// reachable over another transport, so `pkh put` degrades to the
+/// anonymous FTP queue — dput's default for PPAs ([`super::ftp`]).
+/// [`Refused`] failures mean the server answered but rejected the
+/// upload (host key not accepted, no matching authentication): silently
+/// switching to anonymous FTP would bypass a refusal, so they stay
+/// errors.
+#[derive(Debug)]
+pub enum ConnectFailure {
+    /// The connection itself never came up.
+    Transport(Box<dyn std::error::Error>),
+    /// The server answered but rejected the upload.
+    Refused(Box<dyn std::error::Error>),
+}
+
+impl std::fmt::Display for ConnectFailure {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            ConnectFailure::Transport(e) | ConnectFailure::Refused(e) => write!(f, "{e}"),
+        }
+    }
+}
+
+impl std::error::Error for ConnectFailure {
+    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
+        match self {
+            ConnectFailure::Transport(e) | ConnectFailure::Refused(e) => Some(e.as_ref()),
+        }
+    }
+}
 
 /// Connect to `host:port`, verify the server host key and authenticate as
 /// `login`: every ssh-agent identity first, then the configured and default
@@ -303,10 +338,13 @@ pub fn connect(
     login: &str,
     config: &SshConfig,
     prompter: &dyn Prompter,
-) -> Result<Session, Box<dyn std::error::Error>> {
+) -> Result<Session, ConnectFailure> {
-    let tcp = tcp_connect(host, port)?;
+    use ConnectFailure::*;
 
-    let mut session = Session::new()?;
+    let tcp = tcp_connect(host, port).map_err(|e| Transport(e.into()))?;
+
+    let mut session = Session::new()
+        .map_err(|e| Transport(format!("cannot initialize the SSH session: {e}").into()))?;
     // In blocking mode (the libssh2 default), a call that would block loops
     // in `_libssh2_wait_socket` (via the `BLOCK_ADJUST` macros of
     // session.h in the vendored libssh2-sys sources), which bounds the
@@ -321,14 +359,14 @@ pub fn connect(
     session.set_tcp_stream(tcp);
     session
         .handshake()
-        .map_err(|e| format!("SSH handshake with {host} failed: {e}"))?;
+        .map_err(|e| Transport(format!("SSH handshake with {host} failed: {e}").into()))?;
 
     let (key, key_type) = session
         .host_key()
-        .ok_or_else(|| format!("{host} offered no host key"))?;
+        .ok_or_else(|| Transport(format!("{host} offered no host key").into()))?;
-    verify_host_key(host, port, key, key_type, prompter)?;
+    verify_host_key(host, port, key, key_type, prompter).map_err(Refused)?;
 
-    authenticate(&session, host, login, config)?;
+    authenticate(&session, host, login, config).map_err(Refused)?;
 
     // Only SFTP open/data calls remain on this session: switch from the
     // connection-phase budget to the generous per-call transfer one