From bd09fed0ed42599091d38c10bdb5600110340b24 Mon Sep 17 00:00:00 2001 From: Disy Date: Tue, 15 Sep 2026 21:14:18 +0800 Subject: [PATCH] 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. --- .../browse/install-job-notifications.ts | 19 +- ...nstall-job-notification-visibility.test.ts | 35 ++++ .../install-job-notification-visibility.ts | 32 +++ packages/app-lib/src/api/pack/hosted.rs | 175 ++++++++++------ packages/app-lib/src/util/fetch.rs | 194 ++++++++++++++++-- 5 files changed, 362 insertions(+), 93 deletions(-) create mode 100644 apps/app-frontend/src/helpers/install-job-notification-visibility.test.ts create mode 100644 apps/app-frontend/src/helpers/install-job-notification-visibility.ts diff --git a/apps/app-frontend/src/composables/browse/install-job-notifications.ts b/apps/app-frontend/src/composables/browse/install-job-notifications.ts index c5e58d5..74a6643 100644 --- a/apps/app-frontend/src/composables/browse/install-job-notifications.ts +++ b/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([ - 'queued', - 'running', - 'canceling', - 'waiting_for_user', - 'failed', - 'interrupted', -]) -const retainedJobStatuses = new Set(['succeeded', 'canceled']) const activeJobStatuses = new Set([ '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( diff --git a/apps/app-frontend/src/helpers/install-job-notification-visibility.test.ts b/apps/app-frontend/src/helpers/install-job-notification-visibility.test.ts new file mode 100644 index 0000000..1e31c51 --- /dev/null +++ b/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', + ]) +}) diff --git a/apps/app-frontend/src/helpers/install-job-notification-visibility.ts b/apps/app-frontend/src/helpers/install-job-notification-visibility.ts new file mode 100644 index 0000000..a942256 --- /dev/null +++ b/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([ + 'queued', + 'running', + 'canceling', + 'waiting_for_user', +]) +const failureStatuses = new Set(['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() + + 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 + } +} diff --git a/packages/app-lib/src/api/pack/hosted.rs b/packages/app-lib/src/api/pack/hosted.rs index 2a03625..f96a2cb 100644 --- a/packages/app-lib/src/api/pack/hosted.rs +++ b/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::(); 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::::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 = ¤t; + 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::>() + .await; + downloaded += results.into_iter().try_fold(0, |sum, result| { + Ok::(sum + result?) + })?; } ensure_session(&auth).await?; downloaded += tagged::download( diff --git a/packages/app-lib/src/util/fetch.rs b/packages/app-lib/src/util/fetch.rs index 75cc8cf..648e08d 100644 --- a/packages/app-lib/src/util/fetch.rs +++ b/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) { + 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::>(); + + 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 { } 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 = @@ -3341,6 +3377,21 @@ fn byte_range_header_value( }) } +fn apply_file_transport_headers( + mut request: reqwest::RequestBuilder, + range_start: Option, + range_end: Option, +) -> 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, ¤t); 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> = None; let mut alternate_probe_finished = false; let mut confirmed_switch = None; - let mut transfer_error: Option = 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();