feat(dl): segmented multi-connection downloads

crates/soar-cli/src/cli.rs

@@ -3,6 +3,13 @@ use clap_complete::Shell;
 
 use crate::utils::parse_default_repos_arg;
 
+fn parse_connections(value: &str) -> Result<usize, String> {
+    value
+        .parse::<std::num::NonZeroUsize>()
+        .map(std::num::NonZeroUsize::get)
+        .map_err(|_| "connections must be a positive integer".to_string())
+}
+
 #[derive(Parser)]
 #[command(
     author,
@@ -456,6 +463,11 @@ pub enum Commands {
         /// Overwrite existing download with same file
         #[arg(required = false, long)]
         force_overwrite: bool,
+
+        /// Connections per HTTP or OCI blob download (must be positive).
+        /// OCI manifest layers use separate single-stream downloads.
+        #[arg(required = false, long, value_parser = parse_connections)]
+        connections: Option<usize>,
     },
 
     /// Health check
@@ -590,3 +602,41 @@ pub enum Commands {
         repo: Option<String>,
     },
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn download_connections_must_be_positive() {
+        for value in ["0", "-1", "invalid"] {
+            assert!(
+                Args::try_parse_from([
+                    "soar",
+                    "download",
+                    "--connections",
+                    value,
+                    "https://example.com/file"
+                ])
+                .is_err()
+            );
+        }
+        for (value, expected) in [("1", 1), ("8", 8)] {
+            let args = Args::try_parse_from([
+                "soar",
+                "download",
+                "--connections",
+                value,
+                "https://example.com/file",
+            ])
+            .unwrap();
+            let Commands::Download {
+                connections, ..
+            } = args.command
+            else {
+                panic!("expected download command");
+            };
+            assert_eq!(connections, Some(expected));
+        }
+    }
+}

crates/soar-cli/src/download.rs

@@ -35,6 +35,9 @@ pub struct DownloadContext {
     pub extract_dir: Option<String>,
     pub skip_existing: bool,
     pub force_overwrite: bool,
+    /// Connections each download may open; more than one downloads a
+    /// range-capable server's file in parallel segments.
+    pub connections: usize,
 }
 
 impl DownloadContext {
@@ -124,7 +127,8 @@ async fn download_link(ctx: &DownloadContext, link: &str) -> SoarResult<()> {
 
             let mut dl = Download::new(url)
                 .overwrite(ctx.get_overwrite_mode())
-                .extract(ctx.extract);
+                .extract(ctx.extract)
+                .connections(ctx.connections);
 
             if let Some(ref out) = ctx.output {
                 dl = dl.output(out);
@@ -253,7 +257,9 @@ async fn download_link(ctx: &DownloadContext, link: &str) -> SoarResult<()> {
 
             info!("Downloading package: {}", package.pkg_name);
             if let Some(ref url) = package.ghcr_blob {
-                let mut dl = OciDownload::new(url.as_str()).overwrite(ctx.get_overwrite_mode());
+                let mut dl = OciDownload::new(url.as_str())
+                    .connections(ctx.connections)
+                    .overwrite(ctx.get_overwrite_mode());
 
                 if let Some(ref out) = ctx.output {
                     dl = dl.output(out);
@@ -266,8 +272,9 @@ async fn download_link(ctx: &DownloadContext, link: &str) -> SoarResult<()> {
 
                 dl.execute()?;
             } else {
-                let mut dl =
-                    Download::new(&package.download_url).overwrite(ctx.get_overwrite_mode());
+                let mut dl = Download::new(&package.download_url)
+                    .overwrite(ctx.get_overwrite_mode())
+                    .connections(ctx.connections);
 
                 if let Some(ref out) = ctx.output {
                     dl = dl.output(out);
@@ -289,6 +296,7 @@ async fn handle_oci_download(ctx: &DownloadContext, reference: &str) -> SoarResu
     info!("Downloading using OCI reference: {}", reference);
 
     let mut dl = OciDownload::new(reference)
+        .connections(ctx.connections)
         .filter(ctx.create_filter())
         .parallel(get_config().ghcr_concurrency.unwrap_or(8))
         .overwrite(ctx.get_overwrite_mode());
@@ -384,7 +392,8 @@ fn handle_forge_release(
 
     let mut dl = Download::new(selected_asset.url())
         .overwrite(ctx.get_overwrite_mode())
-        .extract(ctx.extract);
+        .extract(ctx.extract)
+        .connections(ctx.connections);
 
     if let Some(ref out) = ctx.output {
         dl = dl.output(out);

crates/soar-cli/src/main.rs

@@ -582,6 +582,7 @@ async fn handle_cli() -> SoarResult<()> {
                     extract_dir,
                     skip_existing,
                     force_overwrite,
+                    connections,
                 } => {
                     let progress_callback: Arc<dyn Fn(soar_dl::types::Progress) + Send + Sync> = {
                         let pb: Arc<Mutex<Option<ProgressBar>>> = Arc::new(Mutex::new(None));
@@ -596,6 +597,9 @@ async fn handle_cli() -> SoarResult<()> {
                     let match_keywords = match_keywords.unwrap_or_default();
                     let exclude_keywords = exclude_keywords.unwrap_or_default();
 
+                    let connections = connections
+                        .or(get_config().download_concurrency)
+                        .unwrap_or(8);
                     let context = DownloadContext {
                         regexes,
                         globs,
@@ -609,6 +613,7 @@ async fn handle_cli() -> SoarResult<()> {
                         extract_dir,
                         skip_existing,
                         force_overwrite,
+                        connections,
                     };
 
                     download(context, links, github, gitlab, codeberg, gitea, ghcr).await?;

crates/soar-cli/src/plugin_manifest.rs

@@ -47,6 +47,12 @@ label = "Parallel limit"
 type = "number"
 default = 4
 
+[[config]]
+key = "download_concurrency"
+label = "Connections per download"
+type = "number"
+default = 8
+
 [[config]]
 key = "search_limit"
 label = "Search result limit"
@@ -260,6 +266,13 @@ mod tests {
             toml::from_str(&manifest(&["default".into()])).expect("manifest must parse");
         assert_eq!(parsed["schema_version"].as_integer(), Some(1));
         assert_eq!(parsed["id"].as_str(), Some("soar"));
+        let connections = parsed["config"]
+            .as_array()
+            .unwrap()
+            .iter()
+            .find(|entry| entry["key"].as_str() == Some("download_concurrency"))
+            .expect("download concurrency setting");
+        assert_eq!(connections["default"].as_integer(), Some(8));
     }
 
     #[test]

crates/soar-cli/src/self_actions.rs

@@ -4,6 +4,7 @@ use std::{
 };
 
 use semver::Version;
+use soar_config::config::get_config;
 use soar_core::{
     SoarResult,
     error::{ErrorContext, SoarError},
@@ -163,6 +164,7 @@ pub async fn process_self_action(action: &SelfAction) -> SoarResult<()> {
                     })?;
 
                 let dl = Download::new(asset.url())
+                    .connections(get_config().download_concurrency.unwrap_or(8))
                     .output(staging.to_string_lossy())
                     .overwrite(OverwriteMode::Force)
                     .checksum(expected_checksum)

crates/soar-config/src/config.rs

@@ -75,6 +75,11 @@ pub struct Config {
     /// Default: 8
     pub ghcr_concurrency: Option<usize>,
 
+    /// Number of connections each download may open. More than one splits a
+    /// range-capable server's file into parallel segments.
+    /// Default: 8
+    pub download_concurrency: Option<usize>,
+
     /// Limits the number of results returned by a search.
     /// Default: 20
     pub search_limit: Option<usize>,
@@ -373,6 +378,7 @@ impl Config {
             parallel_limit: Some(4),
             search_limit: Some(20),
             ghcr_concurrency: Some(8),
+            download_concurrency: Some(8),
             cross_repo_updates: Some(false),
             install_patterns: Some(default_install_patterns()),
             completions: None,
@@ -463,6 +469,7 @@ impl Config {
             parallel_limit: Some(4),
             search_limit: Some(20),
             ghcr_concurrency: Some(8),
+            download_concurrency: Some(8),
             cross_repo_updates: Some(false),
             install_patterns: Some(default_install_patterns()),
             completions: None,
@@ -530,6 +537,7 @@ impl Config {
         }
 
         self.ghcr_concurrency.get_or_insert(8);
+        self.download_concurrency.get_or_insert(8);
         self.search_limit.get_or_insert(20);
         self.cross_repo_updates.get_or_insert(false);
 
@@ -539,6 +547,9 @@ impl Config {
         if self.ghcr_concurrency == Some(0) {
             return Err(ConfigError::ZeroLimit("ghcr_concurrency"));
         }
+        if self.download_concurrency == Some(0) {
+            return Err(ConfigError::ZeroLimit("download_concurrency"));
+        }
         if self.search_limit == Some(0) {
             return Err(ConfigError::ZeroLimit("search_limit"));
         }
@@ -839,6 +850,7 @@ mod tests {
         assert_eq!(config.parallel_limit, Some(4));
         assert_eq!(config.search_limit, Some(20));
         assert_eq!(config.ghcr_concurrency, Some(8));
+        assert_eq!(config.download_concurrency, Some(8));
         assert_eq!(config.cross_repo_updates, Some(false));
     }
 
@@ -955,11 +967,17 @@ mod tests {
 
     #[test]
     fn test_config_resolve_rejects_zero_limits() {
-        for field in ["parallel_limit", "ghcr_concurrency", "search_limit"] {
+        for field in [
+            "parallel_limit",
+            "ghcr_concurrency",
+            "download_concurrency",
+            "search_limit",
+        ] {
             let mut config = Config::default_config::<&str>(&[]);
             match field {
                 "parallel_limit" => config.parallel_limit = Some(0),
                 "ghcr_concurrency" => config.ghcr_concurrency = Some(0),
+                "download_concurrency" => config.download_concurrency = Some(0),
                 _ => config.search_limit = Some(0),
             }
             assert!(
@@ -1049,13 +1067,15 @@ mod tests {
     fn test_config_resolve_sets_defaults() {
         let mut config = Config::default_config::<&str>(&[]);
         config.ghcr_concurrency = None;
+        config.download_concurrency = None;
         config.search_limit = None;
         config.cross_repo_updates = None;
         config.install_patterns = None;
 
         config.resolve().unwrap();
 
         assert_eq!(config.ghcr_concurrency, Some(8));
+        assert_eq!(config.download_concurrency, Some(8));
         assert_eq!(config.search_limit, Some(20));
         assert_eq!(config.cross_repo_updates, Some(false));
         assert!(config.install_patterns.is_some());

crates/soar-core/src/package/install.rs

@@ -52,7 +52,11 @@ use crate::{
 /// the artifact. A licence publishes none, because it is served from a branch
 /// and is documentation rather than something that runs: pinning it would turn
 /// an upstream copyright-year edit into a failed download.
-async fn install_extras(package: &Package, install_dir: &Path) -> SoarResult<()> {
+async fn install_extras(
+    package: &Package,
+    install_dir: &Path,
+    connections: usize,
+) -> SoarResult<()> {
     let Some(extras) = &package.extra else {
         return Ok(());
     };
@@ -85,7 +89,8 @@ async fn install_extras(package: &Package, install_dir: &Path) -> SoarResult<()>
         }
         let mut dl = Download::new(&e.url)
             .output(dest.to_string_lossy())
-            .overwrite(OverwriteMode::Skip);
+            .overwrite(OverwriteMode::Skip)
+            .connections(connections);
         if let Some(sum) = e.blake3.as_ref() {
             dl = dl.checksum(sum);
         }
@@ -1085,6 +1090,7 @@ impl PackageInstaller {
             let mut dl = OciDownload::new(url.as_str())
                 .output(output_path.to_string_lossy())
                 .parallel(self.config.ghcr_concurrency.unwrap_or(8))
+                .connections(self.config.download_concurrency.unwrap_or(8))
                 .overwrite(OverwriteMode::Skip);
 
             if let Some(ref cb) = self.progress_callback {
@@ -1205,7 +1211,8 @@ impl PackageInstaller {
                     .output(output_path.to_string_lossy())
                     .overwrite(OverwriteMode::Skip)
                     .extract(should_extract)
-                    .extract_to(&extract_dir);
+                    .extract_to(&extract_dir)
+                    .connections(self.config.download_concurrency.unwrap_or(8));
 
                 if let Some(ref bsum) = self.package.bsum {
                     dl = dl.checksum(bsum);
@@ -1353,7 +1360,12 @@ impl PackageInstaller {
                 mark_elfs_executable(&self.install_dir);
             }
 
-            install_extras(&self.package, &self.install_dir).await?;
+            install_extras(
+                &self.package,
+                &self.install_dir,
+                self.config.download_concurrency.unwrap_or(8),
+            )
+            .await?;
 
             // Handle nested_extract: extract an archive within the package
             if let Some(ref nested_archive) = self.nested_extract {
@@ -1619,7 +1631,7 @@ mod tests {
 
     /// Runs `install_extras`, which never waits on anything, to completion.
     fn install_side_files(package: &Package, dir: &Path) {
-        let future = std::pin::pin!(install_extras(package, dir));
+        let future = std::pin::pin!(install_extras(package, dir, 8));
         let mut cx = std::task::Context::from_waker(std::task::Waker::noop());
         match std::future::Future::poll(future, &mut cx) {
             std::task::Poll::Ready(result) => result.unwrap(),

crates/soar-dl/Cargo.toml

@@ -13,6 +13,7 @@ categories.workspace = true
 [dependencies]
 compak = { workspace = true }
 fast-glob = { workspace = true }
+httpdate = "1"
 miette = { workspace = true }
 nix = { workspace = true }
 percent-encoding = { workspace = true }