merge(transfer): GUI segmented downloads (GTC-0..GTC-2)

Wires intra-file range parallelism into the GUI command surface,
matching the CLI pget path. Five commits:

- bootstrap GUI<->CLI transfer parity harness (GTC-0)
- ProviderDownloadExecutor segmented (GTC-1)
- provider_download_file segmented (GTC-2)
- read_range self-dial on SFTP/FTP cloned workers (clone-pool
  workers were unconnected so read_range failed; download() already
  self-dialled, lifted the same guard into read_range)
- live-WAN integration test against axpbuntu (SFTP 2.52x, FTP 1.71x,
  S3 2.15x; byte-identical; .aerotmp cleanup verified)

PD-SFTP-1 6/6 + PD-FTP-1 1/1 Docker regression green after the
read_range fix. Frontend wiring is GTC-5 scope; downstream callers
that pass download_segments=None see no behaviour change.

Author: axpnet

src-tauri/src/bin/aeroftp_cli.rs

@@ -4963,6 +4963,9 @@ async fn run_shared_provider_download_batch(
         max_concurrent: None,
         retry_count: None,
         timeout_seconds: None,
+        // CLI segmented downloads use the dedicated `pget` path, not
+        // the GUI provider executor, so the executor stays single-stream.
+        download_segments: None,
     });
 
     let entries: Vec<TransferEntry> = files
@@ -5164,6 +5167,9 @@ async fn run_shared_provider_upload_batch(
         max_concurrent: None,
         retry_count: None,
         timeout_seconds: None,
+        // CLI segmented downloads use the dedicated `pget` path, not
+        // the GUI provider executor, so the executor stays single-stream.
+        download_segments: None,
     });
 
     let entries: Vec<TransferEntry> = files

src-tauri/src/lib.rs

@@ -3997,6 +3997,10 @@ async fn download_files_batch(
             max_concurrent: params.max_concurrent,
             retry_count: params.retry_count,
             timeout_seconds: params.timeout_seconds,
+            // GTC-1: FTP GUI batch stays on `FtpDownloadExecutor`
+            // (no-double-pool invariant); the segments knob only
+            // matters on the `ProviderDownloadExecutor` path.
+            download_segments: None,
         },
     );
 
@@ -4225,6 +4229,10 @@ async fn upload_files_batch(
             max_concurrent: params.max_concurrent,
             retry_count: params.retry_count,
             timeout_seconds: params.timeout_seconds,
+            // GTC-1: FTP GUI batch stays on `FtpDownloadExecutor`
+            // (no-double-pool invariant); the segments knob only
+            // matters on the `ProviderDownloadExecutor` path.
+            download_segments: None,
         },
     );
 
@@ -4750,6 +4758,10 @@ async fn download_folder(
             max_concurrent: params.max_concurrent,
             retry_count: params.retry_count,
             timeout_seconds: params.timeout_seconds,
+            // GTC-1: FTP GUI batch stays on `FtpDownloadExecutor`
+            // (no-double-pool invariant); the segments knob only
+            // matters on the `ProviderDownloadExecutor` path.
+            download_segments: None,
         },
     );
     info!(
@@ -5280,6 +5292,10 @@ async fn upload_folder(
             max_concurrent: params.max_concurrent,
             retry_count: params.retry_count,
             timeout_seconds: params.timeout_seconds,
+            // GTC-1: FTP GUI batch stays on `FtpDownloadExecutor`
+            // (no-double-pool invariant); the segments knob only
+            // matters on the `ProviderDownloadExecutor` path.
+            download_segments: None,
         },
     );
     info!(

src-tauri/src/provider_commands.rs

@@ -906,6 +906,7 @@ pub async fn provider_pwd(state: State<'_, ProviderState>) -> Result<String, Str
 }
 
 /// Download a file from the remote server
+#[allow(clippy::too_many_arguments)]
 #[tauri::command]
 pub async fn provider_download_file(
     app: AppHandle,
@@ -914,6 +915,7 @@ pub async fn provider_download_file(
     local_path: String,
     modified: Option<String>,
     use_delta: Option<bool>,
+    download_segments: Option<u32>,
 ) -> Result<String, String> {
     let mut provider_lock = state.provider.lock().await;
 
@@ -960,7 +962,7 @@ pub async fn provider_download_file(
     let fname_progress = filename.clone();
 
     let dl_start_time = std::time::Instant::now();
-    let progress_cb: Option<Box<dyn Fn(u64, u64) + Send>> = if file_size > 0 {
+    let mut progress_cb: Option<Box<dyn Fn(u64, u64) + Send>> = if file_size > 0 {
         Some(Box::new(move |transferred: u64, total: u64| {
             let pct = if total > 0 {
                 ((transferred as f64 / total as f64) * 100.0) as u8
@@ -1090,25 +1092,122 @@ pub async fn provider_download_file(
     // Resume-aware download: if provider supports resume and a partial .aerotmp exists,
     // use resume_download to continue from where we left off. This avoids re-downloading
     // data on S3/Azure (pay-per-GB) and all other HTTP-based providers.
-    let result = if provider.supports_resume() {
-        let tmp_path = format!("{}.aerotmp", local_path);
-        let offset = tokio::fs::metadata(&tmp_path)
+    let tmp_path = format!("{}.aerotmp", local_path);
+    let partial_offset = if provider.supports_resume() {
+        tokio::fs::metadata(&tmp_path)
             .await
             .map(|m| m.len())
-            .unwrap_or(0);
-        if offset > 0 {
-            info!(
-                "Resuming download from offset {} bytes: {}",
-                offset, filename
-            );
-            provider
-                .resume_download(&remote_path, &local_path, offset, progress_cb)
-                .await
-        } else {
-            provider
-                .download(&remote_path, &local_path, progress_cb)
-                .await
+            .unwrap_or(0)
+    } else {
+        0
+    };
+
+    // GTC-2: opportunistic intra-file range parallelism on the single-file
+    // path. Gated on no-resume (`partial_offset == 0`) because the segmented
+    // engine pre-allocates and overwrites its own `.aerotmp`; a partial
+    // legacy resume must not be silently dropped. On hard failure we fall
+    // through to the legacy single-stream branch below.
+    let mut segmented_result: Option<Result<(), String>> = None;
+    if partial_offset == 0 {
+        if let Some(requested) = download_segments {
+            if let Some(segments) =
+                crate::provider_transfer_executor::provider_segmented_download_eligible(
+                    provider.as_ref(),
+                    file_size,
+                    requested,
+                    requested as usize,
+                )
+            {
+                info!(
+                    "Segmented download: {} segments on {} ({} bytes)",
+                    segments, filename, file_size
+                );
+                let cancel = tokio_util::sync::CancellationToken::new();
+                let outcome =
+                    crate::provider_transfer_executor::run_provider_segmented_download(
+                        provider.as_ref(),
+                        &remote_path,
+                        &local_path,
+                        file_size,
+                        segments,
+                        progress_cb.take(),
+                        cancel,
+                    )
+                    .await;
+                if let Err(ref e) = outcome {
+                    warn!(
+                        "Segmented download failed, falling back to single-stream: {}",
+                        e
+                    );
+                }
+                segmented_result = Some(outcome);
+            }
         }
+    }
+
+    // If segmented ran (success or hard error) the original progress_cb has
+    // been moved into it. Build a fresh callback for the legacy fallback so
+    // the user still sees per-byte progress.
+    if segmented_result.is_some() && progress_cb.is_none() && file_size > 0 {
+        let app_progress_fb = app.clone();
+        let tid_fb = transfer_id.clone();
+        let fname_fb = filename.clone();
+        let start_fb = std::time::Instant::now();
+        progress_cb = Some(Box::new(move |transferred: u64, total: u64| {
+            let pct = if total > 0 {
+                ((transferred as f64 / total as f64) * 100.0) as u8
+            } else {
+                0
+            };
+            let elapsed = start_fb.elapsed().as_secs_f64();
+            let speed = if elapsed > 0.1 {
+                (transferred as f64 / elapsed) as u64
+            } else {
+                0
+            };
+            let eta = if speed > 0 && transferred < total {
+                ((total - transferred) as f64 / speed as f64) as u64
+            } else {
+                0
+            };
+            let _ = app_progress_fb.emit(
+                "transfer_event",
+                crate::TransferEvent {
+                    event_type: "progress".to_string(),
+                    transfer_id: tid_fb.clone(),
+                    filename: fname_fb.clone(),
+                    direction: "download".to_string(),
+                    message: None,
+                    progress: Some(crate::TransferProgress {
+                        transfer_id: tid_fb.clone(),
+                        filename: fname_fb.clone(),
+                        direction: "download".to_string(),
+                        percentage: pct,
+                        transferred,
+                        total,
+                        speed_bps: speed,
+                        eta_seconds: eta as u32,
+                        total_files: None,
+                        path: None,
+                    }),
+                    path: None,
+                    delta_stats: None,
+                    fallback_reason: None,
+                },
+            );
+        }));
+    }
+
+    let result = if let Some(Ok(())) = &segmented_result {
+        Ok(())
+    } else if partial_offset > 0 {
+        info!(
+            "Resuming download from offset {} bytes: {}",
+            partial_offset, filename
+        );
+        provider
+            .resume_download(&remote_path, &local_path, partial_offset, progress_cb)
+            .await
     } else {
         provider
             .download(&remote_path, &local_path, progress_cb)
@@ -1179,11 +1278,13 @@ pub async fn provider_download_folder(
     max_concurrent: Option<u32>,
     retry_count: Option<u32>,
     timeout_seconds: Option<u64>,
+    download_segments: Option<u32>,
 ) -> Result<String, String> {
     let runtime_settings = resolve_provider_transfer_settings(TransferSettingsInput {
         max_concurrent,
         retry_count,
         timeout_seconds,
+        download_segments,
     });
 
     // Capture current pwd so we can restore it after folder scan changes it
@@ -1236,6 +1337,10 @@ pub async fn provider_upload_folder(
         max_concurrent,
         retry_count,
         timeout_seconds,
+        // Upload-side intra-file parallelism is a separate slice (out
+        // of scope for GTC-1); upload paths keep single-stream legacy
+        // behaviour regardless of the requested segments knob.
+        download_segments: None,
     });
 
     // Capture current pwd so we can restore it after upload