feat(transfer): SFTP connection pool for parallel file transfers (PD-SFTP-1)

SFTP folder/batch transfers in the shared engine were serialised by the
single locked provider session. Give SFTP the same pooled, file-level
parallelism FTP already has, instead of a separate ad-hoc mechanism.

SftpProvider now retains a secure SftpConnectionSpec (password and key
passphrase kept in SecretString, captured at connect() before the outer
config is zeroized: the exact posture the FTP manager already ships).
clone_for_transfer() returns an independent, not-yet-connected worker
carrying only that spec; the first transfer dials its own SSH
connection (separate socket and auth), so N files run over N
independent connections with no SSH handle or channel shared.

Host-key safety on the re-dial: pooled dials never trust unknown hosts
(the key is already in known_hosts from the first connect, a changed
key is rejected) and the freshly accepted fingerprint is verified
against the one pinned at the first connect, aborting the worker on
mismatch.

Wired through the existing shared clone-pool executor dispatch (the
same path S3 and Azure use) under a distinct SftpConnectionPool
executor kind and an Sftp session-lease label, so capacity and metrics
never mislabel the transport. The capability only flips to pool-backed
once the secure spec exists; without it SFTP stays a single locked
lease (no overclaim, non-regression).

An earlier attempt multiplexed channels over one SSH session; it was
reworked because the shared engine must use one consistent model, not
a third parallelism path.

Live-validated against the repo SFTP Docker fixture (reproducible, no
secrets): 8 x 16 MiB, byte-identical SHA-256 at concurrency 1/3/5 with
a real ~2.9x speedup, proving an actual pool rather than false
parallelism. New tests/integration_sftp_pool.rs (ignored by default).

Co-Authored-By: aeroftp[bot] <aeroftp[bot]@users.noreply.github.com>
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: 3 people

src-tauri/src/provider_transfer_executor.rs

@@ -32,6 +32,13 @@ pub enum ProviderExecutorSessionModel {
         provider_type: ProviderType,
         max_leases: usize,
     },
+    /// SFTP: N independent SSH connections re-dialled by clone workers
+    /// (PD-SFTP-1). Same clone-worker dispatch as `HttpClonePool`; the
+    /// session pool is labelled `Sftp`, never `HttpClone`.
+    SftpConnectionPool {
+        provider_type: ProviderType,
+        max_leases: usize,
+    },
 }
 
 impl ProviderExecutorSessionModel {
@@ -45,11 +52,17 @@ impl ProviderExecutorSessionModel {
             Self::HttpClonePool { max_leases, .. } => {
                 TransferSessionPoolHandle::http_clone(label, *max_leases)
             }
+            Self::SftpConnectionPool { max_leases, .. } => {
+                TransferSessionPoolHandle::sftp_connection(label, *max_leases)
+            }
         }
     }
 
     fn is_clone_pool(&self) -> bool {
-        matches!(self, Self::HttpClonePool { .. })
+        matches!(
+            self,
+            Self::HttpClonePool { .. } | Self::SftpConnectionPool { .. }
+        )
     }
 }
 
@@ -101,23 +114,37 @@ pub async fn resolve_provider_executor_session_model(
 
     let provider_type = provider.provider_type();
     let caps = provider.transfer_capabilities();
-    let executor_can_clone =
-        provider.transfer_executor_kind() == ProviderTransferExecutorKind::HttpClonePool;
+    let executor_kind = provider.transfer_executor_kind();
     let scheduler_can_parallelize = caps.file_parallel == Capability::Supported
         && caps.session_pool == Capability::Supported
         && provider.clone_for_transfer().is_ok();
 
-    if executor_can_clone && scheduler_can_parallelize {
-        let advertised = caps
-            .max_file_slots
-            .unwrap_or_else(|| provider.transfer_executor_max_sessions())
-            .max(1) as usize;
-        ProviderExecutorSessionModel::HttpClonePool {
-            provider_type,
-            max_leases: advertised.min(max_concurrent.max(1)).max(1),
+    if !scheduler_can_parallelize {
+        return ProviderExecutorSessionModel::locked(Some(provider_type));
+    }
+
+    let max_leases = caps
+        .max_file_slots
+        .unwrap_or_else(|| provider.transfer_executor_max_sessions())
+        .max(1) as usize;
+    let max_leases = max_leases.min(max_concurrent.max(1)).max(1);
+
+    match executor_kind {
+        ProviderTransferExecutorKind::HttpClonePool => {
+            ProviderExecutorSessionModel::HttpClonePool {
+                provider_type,
+                max_leases,
+            }
+        }
+        ProviderTransferExecutorKind::SftpConnectionPool => {
+            ProviderExecutorSessionModel::SftpConnectionPool {
+                provider_type,
+                max_leases,
+            }
+        }
+        ProviderTransferExecutorKind::LockedSingle => {
+            ProviderExecutorSessionModel::locked(Some(provider_type))
         }
-    } else {
-        ProviderExecutorSessionModel::locked(Some(provider_type))
     }
 }
 
@@ -822,6 +849,19 @@ mod tests {
         assert_eq!(pool.capacity().max_leases, 4);
     }
 
+    #[test]
+    fn sftp_connection_model_uses_sftp_lease_kind_not_http() {
+        let model = ProviderExecutorSessionModel::SftpConnectionPool {
+            provider_type: ProviderType::Sftp,
+            max_leases: 4,
+        };
+        let pool = model.session_pool("provider-test");
+
+        assert!(model.is_clone_pool());
+        assert_eq!(pool.capacity().kind, SessionLeaseKind::Sftp);
+        assert_eq!(pool.capacity().max_leases, 4);
+    }
+
     #[test]
     fn locked_list_model_uses_single_legacy_lease() {
         let model = ProviderListSessionModel::locked(Some(ProviderType::WebDav));

src-tauri/src/providers/mod.rs

@@ -302,6 +302,11 @@ impl Default for TransferOptimizationHints {
 pub enum ProviderTransferExecutorKind {
     LockedSingle,
     HttpClonePool,
+    /// SFTP file-level parallelism via N **independent** SSH connections
+    /// re-dialled from a retained secure connection spec (PD-SFTP-1, same
+    /// model as the FTP pool). Distinct from `HttpClonePool` so the
+    /// session pool / metrics label the transport as SFTP, never HTTP.
+    SftpConnectionPool,
 }
 
 /// Execution model the Core DAG scanner may use for remote list/checker work.
@@ -795,7 +800,11 @@ pub trait StorageProvider: Send + Sync {
             self.supports_server_copy(),
         );
 
-        if self.transfer_executor_kind() == ProviderTransferExecutorKind::HttpClonePool {
+        if matches!(
+            self.transfer_executor_kind(),
+            ProviderTransferExecutorKind::HttpClonePool
+                | ProviderTransferExecutorKind::SftpConnectionPool
+        ) {
             caps.file_parallel = crate::transfer_dag::Capability::Supported;
             caps.session_pool = crate::transfer_dag::Capability::Supported;
             caps.max_file_slots = Some(self.transfer_executor_max_sessions().max(1));

src-tauri/src/providers/sftp.rs

@@ -9,7 +9,10 @@
 // Copyright (c) 2024-2026 axpnet: AI-assisted (see AI-TRANSPARENCY.md)
 
 use super::types::is_session_closed_error_message;
-use super::{ProviderError, ProviderType, RemoteEntry, SftpConfig, StorageProvider};
+use super::{
+    ProviderError, ProviderTransferExecutorKind, ProviderType, RemoteEntry, SftpConfig,
+    StorageProvider,
+};
 use crate::ssh_exec::ssh_exec_collect;
 use async_trait::async_trait;
 use russh::client::AuthResult;
@@ -194,6 +197,51 @@ impl Handler for SshHandler {
     }
 }
 
+/// Secure connection spec retained after `connect()` so the shared
+/// transfer engine can re-dial N **independent** SSH+SFTP connections for
+/// file-level parallelism (PD-SFTP-1).
+///
+/// This mirrors `FtpConnectionSpec` / `FtpManager::connection_spec()`:
+/// `provider_connect` zeroizes the outer config password after the first
+/// connect, so the provider must retain its own `SecretString` copy.
+/// Holding credentials for the provider's lifetime is the exact security
+/// posture FTP already ships. Secrets are only exposed (`ExposeSecret`)
+/// at dial time, never on IPC or in logs.
+#[derive(Clone)]
+pub struct SftpConnectionSpec {
+    pub host: String,
+    pub port: u16,
+    pub username: String,
+    pub password: Option<secrecy::SecretString>,
+    pub private_key_path: Option<String>,
+    pub key_passphrase: Option<secrecy::SecretString>,
+    pub initial_path: Option<String>,
+    pub timeout_secs: u64,
+    /// SHA-256 hex of the host key accepted by the first connect. Pool
+    /// re-dials verify the new connection's key against this (defense in
+    /// depth on top of `known_hosts`), same posture as the U-02 rsync pin.
+    pub pinned_host_key_sha256: Option<String>,
+}
+
+impl SftpConnectionSpec {
+    /// Rebuild an `SftpConfig` for an independent worker. `trust_unknown_hosts`
+    /// is forced to `false`: a pooled re-dial must never TOFU, the host key
+    /// is already in `known_hosts` from the first connect.
+    fn to_config(&self) -> SftpConfig {
+        SftpConfig {
+            host: self.host.clone(),
+            port: self.port,
+            username: self.username.clone(),
+            password: self.password.clone(),
+            private_key_path: self.private_key_path.clone(),
+            key_passphrase: self.key_passphrase.clone(),
+            initial_path: self.initial_path.clone(),
+            timeout_secs: self.timeout_secs,
+            trust_unknown_hosts: false,
+        }
+    }
+}
+
 /// SFTP Provider
 ///
 /// Provides secure file transfer over SSH using the SFTP protocol.
@@ -221,6 +269,12 @@ pub struct SftpProvider {
     /// SSH connection (U-02) so the fresh TCP socket it opens for
     /// `aerorsync_serve` does not skip host-key verification.
     host_key_sha256_hex: Arc<std::sync::OnceLock<String>>,
+    /// Secure connection spec for re-dialling independent pool sessions.
+    /// `Some` once captured: either at the end of a successful `connect()`
+    /// (before `provider_connect` zeroizes the outer config) or when this
+    /// provider was produced by `clone_for_transfer()` as a not-yet-
+    /// connected pool worker.
+    connection_spec: Option<SftpConnectionSpec>,
 }
 
 impl SftpProvider {
@@ -241,6 +295,7 @@ impl SftpProvider {
             // in practice. Override per-call with --chunk-size / --buffer-size.
             buffer_size: 256 * 1024,
             host_key_sha256_hex: Arc::new(std::sync::OnceLock::new()),
+            connection_spec: None,
         }
     }
 
@@ -255,6 +310,52 @@ impl SftpProvider {
         self.host_key_sha256_hex.get().cloned()
     }
 
+    /// Secure connection spec retained after a successful `connect()`.
+    /// Mirrors `FtpManager::connection_spec()`. `None` until connected (or
+    /// until set by `clone_for_transfer()` on a pool worker).
+    pub fn connection_spec(&self) -> Option<SftpConnectionSpec> {
+        self.connection_spec.clone()
+    }
+
+    /// Ensure this provider has its own independent, authenticated SSH+SFTP
+    /// connection (PD-SFTP-1). A `clone_for_transfer()` worker starts
+    /// unconnected and carries only the secure spec; the first transfer
+    /// dials a **separate** SSH connection (separate TCP socket, separate
+    /// auth) so N files run truly in parallel, exactly like the FTP pool.
+    ///
+    /// Host-key safety on the re-dial: `connect()` still goes through
+    /// `SshHandler` -> `known_hosts` with `trust_unknown_hosts = false`
+    /// (never TOFU on a pooled dial; the key is already known from the
+    /// first connect, `KeyChanged` is rejected). Defense in depth: the
+    /// freshly accepted fingerprint is compared against the pin captured
+    /// at the first connect and a mismatch aborts the worker.
+    async fn ensure_connected(&mut self) -> Result<(), ProviderError> {
+        if self.sftp.is_some() {
+            return Ok(());
+        }
+        let spec = self
+            .connection_spec
+            .clone()
+            .ok_or(ProviderError::NotConnected)?;
+        self.config = spec.to_config();
+        // Fresh per-connection slot so the comparison reflects this dial.
+        self.host_key_sha256_hex = Arc::new(std::sync::OnceLock::new());
+        self.connect().await?;
+        if let Some(pinned) = spec.pinned_host_key_sha256.as_deref() {
+            match self.accepted_host_key_sha256_hex().as_deref() {
+                Some(seen) if seen == pinned => {}
+                other => {
+                    let _ = self.disconnect().await;
+                    return Err(ProviderError::ConnectionFailed(format!(
+                        "SFTP pool re-dial host key mismatch (expected {}, got {:?}): aborting worker",
+                        pinned, other
+                    )));
+                }
+            }
+        }
+        Ok(())
+    }
+
     /// Return a cloneable handle to the underlying SSH session, if connected.
     ///
     /// Exposed to let sibling modules (rsync-over-SSH, port forwarding, ...)
@@ -807,6 +908,27 @@ impl StorageProvider for SftpProvider {
         self.ssh_handle = Some(Arc::new(TokioMutex::new(handle)));
         self.sftp = Some(sftp);
 
+        // PD-SFTP-1: capture a secure connection spec now, while
+        // `self.config` still holds the secrets (`provider_connect`
+        // zeroizes the outer config only after this returns). The pinned
+        // host-key fingerprint was populated by `SshHandler` during the
+        // handshake; preserve an earlier pin if this dial reused one.
+        let prior_pin = self
+            .connection_spec
+            .as_ref()
+            .and_then(|s| s.pinned_host_key_sha256.clone());
+        self.connection_spec = Some(SftpConnectionSpec {
+            host: self.config.host.clone(),
+            port: self.config.port,
+            username: self.config.username.clone(),
+            password: self.config.password.clone(),
+            private_key_path: self.config.private_key_path.clone(),
+            key_passphrase: self.config.key_passphrase.clone(),
+            initial_path: self.config.initial_path.clone(),
+            timeout_secs: self.config.timeout_secs,
+            pinned_host_key_sha256: self.host_key_sha256_hex.get().cloned().or(prior_pin),
+        });
+
         tracing::info!(
             "SFTP: Connected successfully to {} (home: {})",
             self.config.host,
@@ -984,6 +1106,7 @@ impl StorageProvider for SftpProvider {
         local_path: &str,
         on_progress: Option<Box<dyn Fn(u64, u64) + Send>>,
     ) -> Result<(), ProviderError> {
+        self.ensure_connected().await?;
         let sftp = self.get_sftp()?;
         let full_path = self.normalize_path(remote_path);
 
@@ -1100,6 +1223,7 @@ impl StorageProvider for SftpProvider {
     ) -> Result<(), ProviderError> {
         use tokio::io::AsyncWriteExt;
 
+        self.ensure_connected().await?;
         let sftp = self.get_sftp()?;
         let full_path = self.normalize_path(remote_path);
 
@@ -1581,6 +1705,47 @@ impl StorageProvider for SftpProvider {
         }
     }
 
+    /// PD-SFTP-1: advertise real file-level parallelism only once a secure
+    /// connection spec exists to re-dial independent SSH connections from.
+    /// Without it (never connected, or credentials unavailable) SFTP stays
+    /// a single locked lease: honest non-regression, no overclaim.
+    fn transfer_executor_kind(&self) -> ProviderTransferExecutorKind {
+        if self.connection_spec.is_some() {
+            ProviderTransferExecutorKind::SftpConnectionPool
+        } else {
+            ProviderTransferExecutorKind::LockedSingle
+        }
+    }
+
+    /// Conservative initial cap, mirroring the FTP pool clamp (1..8).
+    /// Each lease is a full independent SSH connection; raise only after a
+    /// live benchmark on the target server says it pays.
+    fn transfer_executor_max_sessions(&self) -> u16 {
+        4
+    }
+
+    /// Produce an independent transfer worker. It is **not connected**:
+    /// it carries only the secure `SftpConnectionSpec` and dials its own
+    /// separate SSH connection lazily on the first transfer
+    /// (`ensure_connected`). No SSH handle or channel is shared, so N
+    /// workers are N independent connections, exactly like the FTP pool.
+    fn clone_for_transfer(&self) -> Result<Box<dyn StorageProvider>, ProviderError> {
+        let spec = self.connection_spec.clone().ok_or_else(|| {
+            ProviderError::NotSupported(
+                "SFTP clone_for_transfer requires a captured connection spec".to_string(),
+            )
+        })?;
+        let mut worker = SftpProvider::new(spec.to_config());
+        worker.current_dir = self.current_dir.clone();
+        worker.home_dir = self.home_dir.clone();
+        worker.download_limit_bps = self.download_limit_bps;
+        worker.upload_limit_bps = self.upload_limit_bps;
+        worker.compression_enabled = self.compression_enabled;
+        worker.buffer_size = self.buffer_size;
+        worker.connection_spec = Some(spec);
+        Ok(Box::new(worker))
+    }
+
     fn set_chunk_sizes(&mut self, upload: Option<u64>, download: Option<u64>) {
         // Cap at 16 MB (larger buffers waste memory without improving throughput)
         let cap = 16 * 1024 * 1024;

src-tauri/src/transfer_dag/session_pool.rs

@@ -170,6 +170,21 @@ impl TransferSessionPoolHandle {
         Self::new(FtpSessionPoolAdapter::new(label, pool))
     }
 
+    /// Concurrency gate for SFTP file-level parallelism (PD-SFTP-1).
+    ///
+    /// The real resource is N **independent** SSH connections re-dialled by
+    /// `clone_for_transfer()` workers from a retained secure connection
+    /// spec (same model the FTP pool uses). This handle only bounds how
+    /// many run at once. Labelled `Sftp`, never `HttpClone`, so capacity
+    /// and metrics tell the truth about the transport.
+    pub fn sftp_connection(label: impl Into<String>, capacity: usize) -> Self {
+        Self::new(SemaphoreSessionPool::with_capacity(
+            SessionLeaseKind::Sftp,
+            label,
+            capacity,
+        ))
+    }
+
     pub fn executor_managed_ftp(label: impl Into<String>, capacity: usize) -> Self {
         Self::new(ExecutorManagedSessionPool::new(
             SessionLeaseKind::Ftp,
@@ -489,6 +504,17 @@ mod tests {
         assert_eq!(lease.info().label, "s3-list");
     }
 
+    #[tokio::test]
+    async fn sftp_connection_pool_reports_sftp_kind_not_http() {
+        let pool = TransferSessionPoolHandle::sftp_connection("axpbuntu-remote", 4);
+        let lease = pool.acquire().await.unwrap();
+
+        assert_eq!(pool.capacity().kind, SessionLeaseKind::Sftp);
+        assert_eq!(pool.capacity().max_leases, 4);
+        assert_eq!(lease.kind(), SessionLeaseKind::Sftp);
+        assert_eq!(lease.info().label, "axpbuntu-remote");
+    }
+
     #[tokio::test]
     async fn executor_managed_ftp_lease_preserves_capacity_without_double_acquire() {
         let pool = TransferSessionPoolHandle::executor_managed_ftp("ftp-batch", 3);