fix: improve parallel pack downloads and suppress stale notifications

Download hosted pack files concurrently, recover file transfers from proxy and content-encoding failures, and prevent old installation failures from reappearing in notifications.

apps/app-frontend/src/composables/browse/install-job-notifications.ts

@@ -21,6 +21,7 @@ import {
 	type InstallPhaseId,
 	type InstallProgress,
 } from '@/helpers/install'
+import { createInstallJobNotificationFilter } from '@/helpers/install-job-notification-visibility'
 import { effectiveInstallProgress, hasDeterminateInstallProgress } from '@/helpers/install-progress'
 import { get_many as getInstances } from '@/helpers/instance'
 import type { DownloadManager } from '@/providers/download-manager'
@@ -258,15 +259,6 @@ const failureSummaryMessages = defineMessages({
 	},
 })
 
-const visibleJobStatuses = new Set<InstallJobStatus>([
-	'queued',
-	'running',
-	'canceling',
-	'waiting_for_user',
-	'failed',
-	'interrupted',
-])
-const retainedJobStatuses = new Set<InstallJobStatus>(['succeeded', 'canceled'])
 const activeJobStatuses = new Set<InstallJobStatus>([
 	'queued',
 	'running',
@@ -724,6 +716,8 @@ export async function useInstallJobNotifications(opts: {
 		return buttons
 	}
 
+	const filterVisibleJobs = createInstallJobNotificationFilter(opts.manager.jobs.value)
+
 	function setJobs(nextJobs: InstallJobSnapshot[]) {
 		for (const job of nextJobs) {
 			if (!jobOrder.has(job.job_id)) {
@@ -731,12 +725,7 @@ export async function useInstallJobNotifications(opts: {
 			}
 		}
 
-		const currentJobIds = new Set(jobs.value.map((job) => job.job_id))
-		const visibleJobs = nextJobs.filter(
-			(job) =>
-				visibleJobStatuses.has(job.status) ||
-				(retainedJobStatuses.has(job.status) && currentJobIds.has(job.job_id)),
-		)
+		const visibleJobs = filterVisibleJobs(nextJobs)
 		syncProgressSnapshots(visibleJobs)
 
 		jobs.value = visibleJobs.sort(

apps/app-frontend/src/helpers/install-job-notification-visibility.test.ts

@@ -0,0 +1,35 @@
+import assert from 'node:assert/strict'
+import test from 'node:test'
+
+import type { InstallJobSnapshot, InstallJobStatus } from './install.ts'
+import { createInstallJobNotificationFilter } from './install-job-notification-visibility.ts'
+
+function job(jobId: string, status: InstallJobStatus): InstallJobSnapshot {
+	return { job_id: jobId, status } as InstallJobSnapshot
+}
+
+test('does not resurrect failures that finished before the notification surface started', () => {
+	const oldFailure = job('old-failure', 'failed')
+	const filter = createInstallJobNotificationFilter([oldFailure, job('old-success', 'succeeded')])
+
+	assert.deepEqual(filter([oldFailure, job('old-success', 'succeeded')]), [])
+	assert.deepEqual(filter([oldFailure, job('current', 'running')]).map((item) => item.job_id), [
+		'current',
+	])
+})
+
+test('keeps an observed task visible when it finishes', () => {
+	const filter = createInstallJobNotificationFilter([job('old-failure', 'failed')])
+
+	assert.deepEqual(filter([job('current', 'running')]).map((item) => item.job_id), ['current'])
+	assert.deepEqual(filter([job('current', 'failed')]).map((item) => item.job_id), ['current'])
+	assert.deepEqual(filter([job('current', 'succeeded')]).map((item) => item.job_id), ['current'])
+})
+
+test('shows a newly received failure even if its active phase completed too quickly to observe', () => {
+	const filter = createInstallJobNotificationFilter([job('old-failure', 'failed')])
+
+	assert.deepEqual(filter([job('new-failure', 'failed')]).map((item) => item.job_id), [
+		'new-failure',
+	])
+})

apps/app-frontend/src/helpers/install-job-notification-visibility.ts

@@ -0,0 +1,32 @@
+import type { InstallJobSnapshot, InstallJobStatus } from './install.ts'
+
+const activeStatuses = new Set<InstallJobStatus>([
+	'queued',
+	'running',
+	'canceling',
+	'waiting_for_user',
+])
+const failureStatuses = new Set<InstallJobStatus>(['failed', 'interrupted'])
+
+/**
+ * Keeps the popup scoped to work the user could actually have observed.
+ * Finished jobs already present when the action bar starts belong to download
+ * history; they must not be resurrected by an unrelated loading event.
+ */
+export function createInstallJobNotificationFilter(initialJobs: InstallJobSnapshot[]) {
+	const missedFinishedJobIds = new Set(
+		initialJobs.filter((job) => !activeStatuses.has(job.status)).map((job) => job.job_id),
+	)
+	let visibleJobIds = new Set<string>()
+
+	return (nextJobs: InstallJobSnapshot[]) => {
+		const visibleJobs = nextJobs.filter((job) => {
+			if (activeStatuses.has(job.status)) return true
+			if (visibleJobIds.has(job.job_id)) return true
+			return failureStatuses.has(job.status) && !missedFinishedJobIds.has(job.job_id)
+		})
+
+		visibleJobIds = new Set(visibleJobs.map((job) => job.job_id))
+		return visibleJobs
+	}
+}

packages/app-lib/src/api/pack/hosted.rs

@@ -12,14 +12,17 @@ use crate::{
         download_to_path,
     },
 };
-use futures::FutureExt;
+use futures::{FutureExt, StreamExt, stream};
 use progress::PackProgress;
 use serde::{Deserialize, Serialize};
 use sha2::{Digest, Sha256};
 use std::{
     collections::{BTreeMap, HashSet},
     path::{Path, PathBuf},
-    sync::{Arc, LazyLock},
+    sync::{
+        Arc, LazyLock,
+        atomic::{AtomicUsize, Ordering},
+    },
 };
 use tokio::{
     io::AsyncReadExt,
@@ -968,71 +971,117 @@ async fn synchronize_with_progress(
         .map(|(file, _, _)| file.size)
         .sum::<u64>();
     let total_files = pending_downloads.len();
-    let mut completed_bytes = 0;
-    for (index, (file, url, object)) in
-        pending_downloads.into_values().enumerate()
-    {
-        ensure_session(&auth).await?;
-        let url = if let Some(external) = &file.external {
-            match crate::api::curseforge::get_file(
-                external.project_id,
-                external.file_id,
-            )
+    if total_files > 0 {
+        let concurrency = crate::api::settings::get()
             .await?
-            .download_url
-            {
-                Some(url) => url,
-                None => crate::api::curseforge::get_download_url(
-                    external.project_id,
-                    external.file_id,
-                )
-                .await?
-                .ok_or_else(|| {
-                    invalid("This CurseForge file requires a manual download")
-                })?,
-            }
-        } else {
-            url
-        };
-        let label =
-            format!("文件 {}/{} · {}", index + 1, total_files, file.path);
-        progress.download(completed_bytes, total_bytes, &label, index == 0);
-        let mut transferred = 0;
-        let mut on_progress = |received: u64, _total: u64| {
-            transferred = received.min(file.size);
-            progress.download(
-                completed_bytes + transferred,
-                total_bytes,
-                &label,
-                false,
-            );
-            futures::future::ready(Ok(())).boxed()
-        };
-        let mut request = DownloadRequest::new(&url, ResourceClass::Modpack)
-            .with_integrity(Integrity {
-                size: Some(file.size),
-                sha256: Some(file.sha256),
-                ..Default::default()
-            });
-        if url.starts_with(&format!("{API}/files/")) {
-            request = request.with_header("Authorization", auth.clone());
-        }
-        download_to_path(
-            request,
-            &object,
-            &state.fetch_semaphore,
-            &state.pool,
-            Some(&mut on_progress),
-        )
-        .await?;
-        completed_bytes += file.size;
-        downloaded += file.size;
+            .effective_max_concurrent_downloads()
+            .clamp(1, 16);
+        let current = std::sync::Mutex::new(BTreeMap::<String, u64>::new());
+        let completed = AtomicUsize::new(0);
         progress.download(
-            completed_bytes,
+            0,
             total_bytes,
-            &label,
-            index + 1 == total_files,
+            &format!("文件 0/{total_files}"),
+            true,
         );
+        let results = stream::iter(pending_downloads.into_values().map(
+            |(file, url, object)| {
+                let current = &current;
+                let completed = &completed;
+                let state = &state;
+                let auth = &auth;
+                async move {
+                    ensure_session(auth).await?;
+                    let url = if let Some(external) = &file.external {
+                        match crate::api::curseforge::get_file(
+                            external.project_id,
+                            external.file_id,
+                        )
+                        .await?
+                        .download_url
+                        {
+                            Some(url) => url,
+                            None => crate::api::curseforge::get_download_url(
+                                external.project_id,
+                                external.file_id,
+                            )
+                            .await?
+                            .ok_or_else(|| {
+                                invalid(
+                                    "This CurseForge file requires a manual download",
+                                )
+                            })?,
+                        }
+                    } else {
+                        url
+                    };
+                    let file_size = file.size;
+                    let file_path = file.path.clone();
+                    let progress_key = file.path.clone();
+                    let mut on_progress = |received: u64, _total: u64| {
+                        let mut bytes = current
+                            .lock()
+                            .unwrap_or_else(|poisoned| poisoned.into_inner());
+                        bytes.insert(progress_key.clone(), received.min(file_size));
+                        progress.download(
+                            bytes.values().sum(),
+                            total_bytes,
+                            &format!(
+                                "文件 {}/{} · {}",
+                                completed.load(Ordering::Relaxed),
+                                total_files,
+                                file_path
+                            ),
+                            false,
+                        );
+                        futures::future::ready(Ok(())).boxed()
+                    };
+                    let mut request =
+                        DownloadRequest::new(&url, ResourceClass::Modpack)
+                            .with_integrity(Integrity {
+                                size: Some(file_size),
+                                sha256: Some(file.sha256),
+                                ..Default::default()
+                            });
+                    if url.starts_with(&format!("{API}/files/")) {
+                        request = request
+                            .with_header("Authorization", auth.clone());
+                    }
+                    download_to_path(
+                        request,
+                        &object,
+                        &state.fetch_semaphore,
+                        &state.pool,
+                        Some(&mut on_progress),
+                    )
+                    .await?;
+                    let completed_files =
+                        completed.fetch_add(1, Ordering::Relaxed) + 1;
+                    let completed_bytes = {
+                        let mut bytes = current
+                            .lock()
+                            .unwrap_or_else(|poisoned| poisoned.into_inner());
+                        bytes.insert(progress_key, file_size);
+                        bytes.values().sum()
+                    };
+                    progress.download(
+                        completed_bytes,
+                        total_bytes,
+                        &format!(
+                            "文件 {completed_files}/{total_files} · {file_path}"
+                        ),
+                        completed_files == total_files,
+                    );
+                    Ok::<_, crate::Error>(file_size)
+                }
+            },
+        ))
+        .buffer_unordered(concurrency)
+        .collect::<Vec<_>>()
+        .await;
+        downloaded += results.into_iter().try_fold(0, |sum, result| {
+            Ok::<u64, crate::Error>(sum + result?)
+        })?;
     }
     ensure_session(&auth).await?;
     downloaded += tagged::download(

packages/app-lib/src/util/fetch.rs

@@ -917,6 +917,32 @@ fn official_route(url: &str, resource: ResourceClass) -> DownloadRoute {
     route
 }
 
+fn add_starlight_direct_recovery_route(routes: &mut Vec<DownloadRoute>) {
+    let recovery_routes = routes
+        .iter()
+        .filter(|route| {
+            route.proxy == ProxyPolicy::System
+                && Url::parse(&route.url).ok().is_some_and(|url| {
+                    url.host_str() == Some("skin.starlight.cool")
+                        && url.path().starts_with("/starlight/mod/packs/files/")
+                })
+        })
+        .cloned()
+        .map(|mut route| {
+            route.proxy = ProxyPolicy::Direct;
+            route
+        })
+        .collect::<Vec<_>>();
+
+    for recovery in recovery_routes {
+        if !routes.iter().any(|route| {
+            route.url == recovery.url && route.proxy == recovery.proxy
+        }) {
+            routes.push(recovery);
+        }
+    }
+}
+
 fn is_official_route(route: &DownloadRoute) -> bool {
     !route.is_mirror && route.source == DownloadRouteSource::Official
 }
@@ -1375,8 +1401,18 @@ fn reqwest_client_builder() -> reqwest::ClientBuilder {
         .user_agent(crate::launcher_user_agent())
 }
 
+fn disable_file_content_decoding(
+    builder: reqwest::ClientBuilder,
+) -> reqwest::ClientBuilder {
+    // File endpoints occasionally arrive through proxies or CDNs with a stale
+    // Content-Encoding header even though the body already contains the raw
+    // file. Keep transport bytes untouched; download integrity validation is
+    // the authoritative check and every request explicitly asks for identity.
+    builder.no_gzip().no_brotli().no_deflate().no_zstd()
+}
+
 fn file_reqwest_client_builder() -> reqwest::ClientBuilder {
-    reqwest_client_builder()
+    disable_file_content_decoding(reqwest_client_builder())
         .http2_adaptive_window(true)
         .http2_keep_alive_interval(Some(time::Duration::from_secs(15)))
 }
@@ -1406,7 +1442,7 @@ pub async fn configured_client() -> crate::Result<reqwest::Client> {
 }
 
 fn http1_file_reqwest_client_builder() -> reqwest::ClientBuilder {
-    reqwest_client_builder().http1_only()
+    disable_file_content_decoding(reqwest_client_builder()).http1_only()
 }
 
 pub static INSECURE_REQWEST_CLIENT: LazyLock<reqwest::Client> =
@@ -3341,6 +3377,21 @@ fn byte_range_header_value(
     })
 }
 
+fn apply_file_transport_headers(
+    mut request: reqwest::RequestBuilder,
+    range_start: Option<u64>,
+    range_end: Option<u64>,
+) -> reqwest::RequestBuilder {
+    // Downloaded files are validated against their unencoded size and hash.
+    // Request the original bytes from the very first attempt so a proxy or CDN
+    // cannot leave reqwest retrying a broken compressed response body.
+    request = request.header(header::ACCEPT_ENCODING, "identity");
+    if let Some(range) = byte_range_header_value(range_start, range_end) {
+        request = request.header(header::RANGE, range);
+    }
+    request
+}
+
 async fn send_path_request_with_clients(
     route: &DownloadRoute,
     custom_header: Option<&(String, String)>,
@@ -3383,7 +3434,11 @@ async fn send_path_request_with_clients(
         };
         let same_as_original = same_origin(&original, &current);
         let allow_sensitive = route.allow_sensitive_headers && same_as_original;
-        let mut request = client.get(current.clone());
+        let mut request = apply_file_transport_headers(
+            client.get(current.clone()),
+            range_start,
+            range_end,
+        );
         if let Some((name, value)) = custom_header
             && (allow_sensitive || !is_sensitive_header(name))
             && (!name.eq_ignore_ascii_case("x-api-key")
@@ -3394,11 +3449,6 @@ async fn send_path_request_with_clients(
         if allow_sensitive && let Some(credentials) = credentials {
             request = request.header("Authorization", &credentials.session);
         }
-        if let Some(range) = byte_range_header_value(range_start, range_end) {
-            request = request
-                .header(header::RANGE, range)
-                .header(header::ACCEPT_ENCODING, "identity");
-        }
         let response = match request.send().await {
             Ok(response) => response,
             Err(error) => {
@@ -5522,6 +5572,13 @@ async fn download_to_path_inner(
     if routes.is_empty() {
         routes.push(official_route(&request.url, request.resource));
     }
+    // A local/system proxy can successfully return many small authenticated
+    // objects and then truncate a later response body. Add a same-origin direct
+    // transport and let observed route health choose which transport goes
+    // first. Sensitive headers remain constrained to the original origin by
+    // send_path_request_with_clients.
+    add_starlight_direct_recovery_route(&mut routes);
+    order_auto_routes(&mut routes, request.resource, false);
     let part_path = suffixed_path(destination, ".part");
 
     if !request.integrity.is_empty()
@@ -6427,7 +6484,7 @@ async fn download_to_path_inner(
                 let mut alternate_probe: Option<RouteProbeFuture<'_>> = None;
                 let mut alternate_probe_finished = false;
                 let mut confirmed_switch = None;
-                let mut transfer_error: Option<crate::Error> = None;
+                let mut transfer_error: Option<(crate::Error, bool)> = None;
                 loop {
                     tokio::select! {
                         item = stream.next() => {
@@ -6437,13 +6494,15 @@ async fn download_to_path_inner(
                             let chunk = match item {
                                 Ok(chunk) => chunk,
                                 Err(error) => {
+                                    let decode_failure = error.is_decode();
                                     if is_h2_protocol_failure(&error)
                                         && let Some(authority) =
                                             url_authority(&final_url)
                                     {
                                         record_authority_h2_failure(&authority);
                                     }
-                                    transfer_error = Some(error.into());
+                                    transfer_error =
+                                        Some((error.into(), decode_failure));
                                     break;
                                 }
                             };
@@ -6519,9 +6578,14 @@ async fn download_to_path_inner(
                                     idle_ms = elapsed.as_millis(),
                                     "Download body idle deadline exceeded"
                                 );
-                                transfer_error = Some(crate::ErrorKind::NetworkError(
-                                    format!("download body idle for {}", elapsed.as_secs()),
-                                ).into());
+                                transfer_error = Some((
+                                    crate::ErrorKind::NetworkError(format!(
+                                        "download body idle for {}",
+                                        elapsed.as_secs()
+                                    ))
+                                    .into(),
+                                    false,
+                                ));
                                 break;
                             }
                             if matches!(
@@ -6580,7 +6644,7 @@ async fn download_to_path_inner(
                     break;
                 }
 
-                if let Some(error) = transfer_error {
+                if let Some((error, decode_failure)) = transfer_error {
                     record_route_failure(route, request.resource, None);
                     record_native_transfer_failure(route, None);
                     preserve_or_remove_partial(
@@ -6589,12 +6653,22 @@ async fn download_to_path_inner(
                         any_route_can_resume(&routes),
                     )
                     .await?;
+                    if decode_failure && attempts < file_attempt_budget {
+                        busted_for_route = Some((
+                            route_index,
+                            cache_busted_download_url(&route.url, attempts),
+                        ));
+                    }
                     record_download_attempt_failure(
                         &mut attempt_history,
                         route,
                         attempts,
                         &error,
-                        "resume_or_switch",
+                        if decode_failure {
+                            "cache_bust_and_resume_or_switch"
+                        } else {
+                            "resume_or_switch"
+                        },
                         Some(status),
                         remote_addr,
                         Some(http_version),
@@ -8588,6 +8662,96 @@ mod tests {
         assert_eq!(redirected, original);
     }
 
+    #[test]
+    fn file_requests_disable_content_encoding_from_the_first_attempt() {
+        let client = reqwest::Client::new();
+        let full = apply_file_transport_headers(
+            client.get("https://example.com/file.jar"),
+            None,
+            None,
+        )
+        .build()
+        .unwrap();
+        assert_eq!(
+            full.headers().get(header::ACCEPT_ENCODING).unwrap(),
+            "identity"
+        );
+        assert!(!full.headers().contains_key(header::RANGE));
+
+        let resumed = apply_file_transport_headers(
+            client.get("https://example.com/file.jar"),
+            Some(1024),
+            None,
+        )
+        .build()
+        .unwrap();
+        assert_eq!(
+            resumed.headers().get(header::ACCEPT_ENCODING).unwrap(),
+            "identity"
+        );
+        assert_eq!(
+            resumed.headers().get(header::RANGE).unwrap(),
+            "bytes=1024-"
+        );
+    }
+
+    #[test]
+    fn starlight_hosted_files_have_a_direct_recovery_route() {
+        let mut routes = vec![official_route(
+            "https://skin.starlight.cool/starlight/mod/packs/files/release/hash",
+            ResourceClass::Modpack,
+        )];
+
+        add_starlight_direct_recovery_route(&mut routes);
+
+        assert_eq!(routes.len(), 2);
+        assert_eq!(routes[0].proxy, ProxyPolicy::System);
+        assert_eq!(routes[1].proxy, ProxyPolicy::Direct);
+        assert_eq!(routes[0].url, routes[1].url);
+        assert!(routes[1].allow_sensitive_headers);
+    }
+
+    #[test]
+    fn unrelated_downloads_do_not_bypass_the_configured_proxy() {
+        let mut routes = vec![official_route(
+            "https://libraries.minecraft.net/example.jar",
+            ResourceClass::MinecraftLibrary,
+        )];
+
+        add_starlight_direct_recovery_route(&mut routes);
+
+        assert_eq!(routes.len(), 1);
+        assert_eq!(routes[0].proxy, ProxyPolicy::System);
+    }
+
+    #[tokio::test]
+    async fn file_client_ignores_a_stale_content_encoding_header() {
+        let body = b"raw jar bytes".to_vec();
+        let (url, requests, server) = spawn_http_fixture(
+            "200 OK",
+            "Content-Encoding: gzip\r\n",
+            body.clone(),
+            Duration::ZERO,
+        )
+        .await;
+        let route = direct_test_route(url, DownloadRouteSource::Official);
+        let client = file_reqwest_client_builder()
+            .no_proxy()
+            .redirect(reqwest::redirect::Policy::none())
+            .build()
+            .unwrap();
+
+        let (response, _) = send_path_request_with_clients(
+            &route, None, None, None, None, &client, &client, None,
+        )
+        .await
+        .unwrap();
+
+        assert_eq!(response.bytes().await.unwrap().as_ref(), body.as_slice());
+        assert_eq!(requests.load(Ordering::Relaxed), 1);
+        server.abort();
+    }
+
     #[tokio::test]
     async fn verifies_streaming_integrity_algorithms() {
         let file = tempfile::NamedTempFile::new().unwrap();