diff --git a/Cargo.lock b/Cargo.lock index 4d43cbf..d02c88e 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -6841,7 +6841,7 @@ dependencies = [ [[package]] name = "theseus" -version = "1.0.0-beta15" +version = "1.0.2-beta4" dependencies = [ "ariadne", "async-minecraft-ping", @@ -6945,7 +6945,7 @@ dependencies = [ [[package]] name = "theseus_gui" -version = "1.0.0-beta15" +version = "1.0.2-beta4" dependencies = [ "ashpd", "async_zip", diff --git a/apps/app-frontend/package.json b/apps/app-frontend/package.json index 47adfd0..0f14e95 100644 --- a/apps/app-frontend/package.json +++ b/apps/app-frontend/package.json @@ -1,7 +1,7 @@ { "name": "@modrinth/app-frontend", "private": true, - "version": "1.0.0-beta15", + "version": "1.0.2-beta4", "type": "module", "scripts": { "dev": "vite", diff --git a/apps/app-frontend/src/announcements/catalog.ts b/apps/app-frontend/src/announcements/catalog.ts index 136f426..86d81b5 100644 --- a/apps/app-frontend/src/announcements/catalog.ts +++ b/apps/app-frontend/src/announcements/catalog.ts @@ -32,6 +32,33 @@ export const ANNOUNCEMENT_CHANGE_TYPES: readonly AnnouncementChangeType[] = [ ] export const launcherAnnouncements: readonly LauncherAnnouncement[] = [ + { + id: 'launcher-1.0.2-beta4', + version: '1.0.2-beta4', + publishedAt: '2026-10-08', + title: { + 'en-US': 'Starlight Launcher 1.0.2-beta4', + 'zh-CN': 'Starlight Launcher 1.0.2-beta4', + }, + changes: { + changed: [ + { + 'en-US': 'Improved download scheduling to reduce concurrency automatically when tasks are waiting for capacity, making large Minecraft installations more reliable on constrained connections.', + 'zh-CN': '优化下载任务调度：任务因容量不足排队时会自动降低并发，提升大型 Minecraft 版本在受限网络环境下的安装稳定性。', + }, + ], + fixed: [ + { + 'en-US': 'Fixed downloads timing out while waiting for internal connection capacity, which could cause Minecraft installations to fail even when the network was available.', + 'zh-CN': '修复下载任务等待内部连接容量时过早超时，导致网络正常的情况下 Minecraft 安装仍可能失败的问题。', + }, + { + 'en-US': 'Fixed instance group changes not saving for some instances.', + 'zh-CN': '修复部分实例的分组修改无法保存的问题。', + }, + ], + }, + }, { id: 'launcher-1.0.2-beta3', version: '1.0.2-beta3', diff --git a/apps/app/Cargo.toml b/apps/app/Cargo.toml index 8a45c99..91750d1 100644 --- a/apps/app/Cargo.toml +++ b/apps/app/Cargo.toml @@ -1,7 +1,7 @@ [package] name = "theseus_gui" # The actual version is set by the theseus-build workflow on tagging -version = "1.0.0-beta15" +version = "1.0.2-beta4" edition.workspace = true description = "Axolotl Launcher is a cross-platform Minecraft launcher" homepage = "https://www.ghs.red" diff --git a/packages/app-lib/Cargo.toml b/packages/app-lib/Cargo.toml index 570aab0..826aceb 100644 --- a/packages/app-lib/Cargo.toml +++ b/packages/app-lib/Cargo.toml @@ -1,7 +1,7 @@ [package] name = "theseus" # The actual version is set by the theseus-build workflow on tagging -version = "1.0.0-beta15" +version = "1.0.2-beta4" edition.workspace = true [dependencies] diff --git a/packages/app-lib/src/launcher/download.rs b/packages/app-lib/src/launcher/download.rs index e53e743..08d987d 100644 --- a/packages/app-lib/src/launcher/download.rs +++ b/packages/app-lib/src/launcher/download.rs @@ -1947,9 +1947,9 @@ pub async fn download_assets( // failures. Runs concurrently (same budget as the original scheduler) so // import flows are not serialised. if !fallback_assets.is_empty() { - let limit = crate::util::download::task_concurrency_limit(st) - .map(|limit| limit.saturating_mul(2)) - .unwrap_or(ASSET_BATCH_CONCURRENCY); + let limit = + crate::util::download::single_authority_task_concurrency_limit(st) + .unwrap_or(ASSET_BATCH_CONCURRENCY); futures::stream::iter(fallback_assets) .map(Ok::) .try_for_each_concurrent(limit, |item| { @@ -2123,7 +2123,7 @@ pub async fn download_libraries( let num_files = tasks.len(); loading_try_for_each_concurrent( stream::iter(tasks).map(Ok::, crate::Error>), - crate::util::download::task_concurrency_limit(&st).map(|limit| limit.saturating_mul(2)), + crate::util::download::task_concurrency_limit(&st), loading_bar, loading_amount, num_files, diff --git a/packages/app-lib/src/state/mod.rs b/packages/app-lib/src/state/mod.rs index dde048c..d0e689c 100644 --- a/packages/app-lib/src/state/mod.rs +++ b/packages/app-lib/src/state/mod.rs @@ -118,6 +118,7 @@ pub struct State { download_sample_requests: AtomicU64, download_sample_errors: AtomicU64, download_sample_throttles: AtomicU64, + download_sample_queue_waits: AtomicU64, pub(crate) install_job_semaphore: Semaphore, pub(crate) install_db_semaphore: Semaphore, pub(crate) install_job_cancellations: DashMap, @@ -196,6 +197,7 @@ struct AutoConcurrencySample { requests: u64, errors: u64, throttles: u64, + queue_waits: u64, } /// Per-instance lock registry with task-local reentrancy. @@ -399,6 +401,15 @@ impl AutoConcurrencyController { } else { sample.errors as f64 / sample.requests as f64 }; + if sample.queue_waits > 0 { + self.high_utilization_windows = 0; + self.pressure_windows = 0; + self.probe = None; + self.cooldown_until = Some(now + AUTO_DOWNLOAD_PROBE_COOLDOWN); + return current + .saturating_sub(current.div_ceil(4)) + .max(AUTO_DOWNLOAD_CONCURRENCY_MIN); + } // A single throttled window is handled by the per-host limiter; the // global concurrency only backs off when pressure persists across // several samples, and then in small steps instead of a big cut. @@ -677,6 +688,11 @@ impl State { self.download_sample_errors.fetch_add(1, Ordering::AcqRel); } + pub(crate) fn record_download_queue_wait(&self) { + self.download_sample_queue_waits + .fetch_add(1, Ordering::AcqRel); + } + pub(crate) fn update_download_settings( self: &Arc, settings: &Settings, @@ -735,6 +751,7 @@ impl State { self.download_sample_requests.swap(0, Ordering::AcqRel); self.download_sample_errors.swap(0, Ordering::AcqRel); self.download_sample_throttles.swap(0, Ordering::AcqRel); + self.download_sample_queue_waits.swap(0, Ordering::AcqRel); continue; } let sample = AutoConcurrencySample { @@ -749,6 +766,9 @@ impl State { throttles: self .download_sample_throttles .swap(0, Ordering::AcqRel), + queue_waits: self + .download_sample_queue_waits + .swap(0, Ordering::AcqRel), }; let current = self.download_concurrency(); let ceiling = @@ -771,6 +791,7 @@ impl State { requests = sample.requests, errors = sample.errors, throttles = sample.throttles, + queue_waits = sample.queue_waits, "Adjusted automatic download concurrency" ); self.resize_download_concurrency(target); @@ -935,6 +956,7 @@ impl State { download_sample_requests: AtomicU64::new(0), download_sample_errors: AtomicU64::new(0), download_sample_throttles: AtomicU64::new(0), + download_sample_queue_waits: AtomicU64::new(0), install_job_semaphore: Semaphore::new(MAX_CONCURRENT_INSTALL_JOBS), install_db_semaphore: Semaphore::new(1), install_job_cancellations: DashMap::new(), @@ -1024,6 +1046,7 @@ pub(crate) async fn test_state( download_sample_requests: AtomicU64::new(0), download_sample_errors: AtomicU64::new(0), download_sample_throttles: AtomicU64::new(0), + download_sample_queue_waits: AtomicU64::new(0), install_job_semaphore: Semaphore::new(1), install_db_semaphore: Semaphore::new(1), install_job_cancellations: DashMap::new(), @@ -1053,9 +1076,24 @@ mod auto_concurrency_tests { requests: 100, errors: 0, throttles: 0, + queue_waits: 0, } } + #[test] + fn queue_pressure_backs_off_immediately_and_repeatedly() { + let mut controller = AutoConcurrencyController::default(); + let now = Instant::now(); + let congested = AutoConcurrencySample { + queue_waits: 1, + ..healthy(8, 100) + }; + + assert_eq!(controller.next_target(now, 128, 128, congested), 96); + assert_eq!(controller.next_target(now, 96, 128, congested), 72); + assert_eq!(controller.next_target(now, 16, 128, congested), 16); + } + #[test] fn probes_up_after_two_saturated_windows_and_respects_ceiling() { let mut controller = AutoConcurrencyController::default(); @@ -1130,6 +1168,7 @@ mod auto_concurrency_tests { requests: 100, errors: 5, throttles: 0, + queue_waits: 0, }; assert_eq!(controller.next_target(Instant::now(), 16, 128, sample), 16); } diff --git a/packages/app-lib/src/util/download/h2_download.rs b/packages/app-lib/src/util/download/h2_download.rs index bc582ac..d8a46ad 100644 --- a/packages/app-lib/src/util/download/h2_download.rs +++ b/packages/app-lib/src/util/download/h2_download.rs @@ -62,6 +62,7 @@ pub(crate) enum H2DownloadOutcome { #[derive(Clone, Copy, Debug, Eq, PartialEq)] pub(crate) enum H2DownloadFailure { Ineligible(&'static str), + Capacity, Connect, Tls, Protocol, @@ -76,6 +77,7 @@ impl H2DownloadFailure { pub(crate) const fn as_str(self) -> &'static str { match self { Self::Ineligible(reason) => reason, + Self::Capacity => "HTTP/2 connection capacity is unavailable", Self::Connect => "HTTP/2 TCP connection failed", Self::Tls => "HTTP/2 TLS connection failed", Self::Protocol => "HTTP/2 protocol failed", @@ -104,6 +106,7 @@ impl H2DownloadFailure { pub(crate) async fn try_download_via_h2( request: &DownloadRequest, route: &DownloadRoute, + semaphore: &fetch::FetchSemaphore, destination: &Path, part_path: &Path, policy: super::native::NativeH2Policy, @@ -122,13 +125,19 @@ pub(crate) async fn try_download_via_h2( preserve_partial: false, }; } - let connection = match connect_authority( - route, - true, - policy.allow_cold_connection, - ) - .await + let connection_wait = + connect_authority(route, true, policy.allow_cold_connection); + let connection_result = if let Some(cancellation) = + request.cancellation.as_ref() { + tokio::select! { + _ = cancellation.cancelled() => return H2DownloadOutcome::Canceled, + result = connection_wait => result, + } + } else { + connection_wait.await + }; + let connection = match connection_result { Ok(connection) => connection, Err(failure) => { return H2DownloadOutcome::Fallback { @@ -137,6 +146,30 @@ pub(crate) async fn try_download_via_h2( }; } }; + // The shared connection reserves native capacity when it is created. + // Acquire the per-download permit afterwards so every native path uses + // the same native-then-fetch order and cannot form a circular wait. + let fetch_permit_wait = + fetch::acquire_fetch_permit_with_pressure(semaphore); + let fetch_permit_result = if let Some(cancellation) = + request.cancellation.as_ref() + { + tokio::select! { + _ = cancellation.cancelled() => return H2DownloadOutcome::Canceled, + result = fetch_permit_wait => result, + } + } else { + fetch_permit_wait.await + }; + let _fetch_permit = match fetch_permit_result { + Ok(permit) => permit, + Err(_) => { + return H2DownloadOutcome::Fallback { + failure: H2DownloadFailure::Connect, + preserve_partial: false, + }; + } + }; let Ok(uri) = route.url.parse::() else { return H2DownloadOutcome::Fallback { failure: H2DownloadFailure::Http, @@ -156,14 +189,20 @@ pub(crate) async fn try_download_via_h2( let total_size = if let Some(size) = expected_size { size } else { - let _probe_stream_permit = match tokio::time::timeout( - ASSET_RESOURCE_WAIT_TIMEOUT, - super::h2_stream_budget::acquire(route), - ) - .await + let probe_wait = super::h2_stream_budget::acquire_with_pressure(route); + let probe_result = if let Some(cancellation) = + request.cancellation.as_ref() { - Ok(Ok(permit)) => permit, - Ok(Err(_)) | Err(_) => { + tokio::select! { + _ = cancellation.cancelled() => return H2DownloadOutcome::Canceled, + result = probe_wait => result, + } + } else { + probe_wait.await + }; + let _probe_stream_permit = match probe_result { + Ok(permit) => permit, + Err(_) => { return H2DownloadOutcome::Fallback { failure: H2DownloadFailure::Connect, preserve_partial: false, @@ -239,10 +278,7 @@ pub(crate) async fn try_download_via_h2( crate::install::DownloadItemStatus::WaitingForResource, ) .await; - let stream_wait = tokio::time::timeout( - ASSET_RESOURCE_WAIT_TIMEOUT, - super::h2_stream_budget::acquire(route), - ); + let stream_wait = super::h2_stream_budget::acquire_with_pressure(route); let stream_wait_started = Instant::now(); let stream_result = if let Some(cancellation) = request.cancellation.as_ref() @@ -255,8 +291,8 @@ pub(crate) async fn try_download_via_h2( stream_wait.await }; let _stream_permit = match stream_result { - Ok(Ok(permit)) => permit, - Ok(Err(_)) | Err(_) => { + Ok(permit) => permit, + Err(_) => { return H2DownloadOutcome::Fallback { failure: H2DownloadFailure::Connect, preserve_partial: false, @@ -327,6 +363,7 @@ async fn connect_authority( "Failed to establish shared HTTP/2 connection" ); Err(match error.kind { + H2ConnectFailureKind::Capacity => H2DownloadFailure::Capacity, H2ConnectFailureKind::Tcp => H2DownloadFailure::Connect, H2ConnectFailureKind::Tls => H2DownloadFailure::Tls, H2ConnectFailureKind::Protocol => H2DownloadFailure::Protocol, @@ -771,6 +808,12 @@ mod tests { use super::*; use bytes::Bytes; + #[test] + fn capacity_wait_is_not_classified_as_a_transport_failure() { + assert!(!H2DownloadFailure::Capacity.is_transfer_failure()); + assert!(!H2DownloadFailure::Capacity.should_cooldown_authority()); + } + #[tokio::test] async fn single_stream_reports_bytes_before_the_response_finishes() { let size = 512 * 1024; @@ -1183,28 +1226,6 @@ async fn download_asset_item( "timed out waiting for asset destination lock".to_string(), ) })?; - let fetch_permit = if apply_native_policy { - let Some(semaphore) = native_semaphore else { - return Err(crate::ErrorKind::OtherError( - "native asset batch is missing fetch budget".to_string(), - ) - .into()); - }; - Some( - tokio::time::timeout( - ASSET_RESOURCE_WAIT_TIMEOUT, - semaphore.0.acquire(), - ) - .await - .map_err(|_| { - crate::ErrorKind::NetworkError( - "timed out waiting for asset fetch permit".to_string(), - ) - })??, - ) - } else { - None - }; // A different downloader may have committed the object while this item // waited for the destination lock. Reuse it instead of opening another // stream, which also prevents cross-engine `.part`/rename races. @@ -1243,24 +1264,26 @@ async fn download_asset_item( return Ok(AssetBatchItemOutcome::LocalObjectFailed { error }); } }; - let _stream_permit = if apply_native_policy { - Some( - tokio::time::timeout( - ASSET_RESOURCE_WAIT_TIMEOUT, - super::h2_stream_budget::acquire_asset(route), + // Resolve the primary or lazily expanded sibling before taking the fetch + // permit. Shared connections reserve native capacity, so this preserves + // the native-then-fetch order used by every other native path. + let connection = connections.connection(rescue).await; + let fetch_permit = if apply_native_policy { + let Some(semaphore) = native_semaphore else { + return Err(crate::ErrorKind::OtherError( + "native asset batch is missing fetch budget".to_string(), ) - .await - .map_err(|_| { - crate::ErrorKind::NetworkError( - "timed out waiting for asset HTTP/2 stream permit" - .to_string(), - ) - })??, - ) + .into()); + }; + Some(fetch::acquire_fetch_permit_with_pressure(semaphore).await?) + } else { + None + }; + let _stream_permit = if apply_native_policy { + Some(super::h2_stream_budget::acquire_asset_with_pressure(route).await?) } else { None }; - let connection = connections.connection(rescue).await; let _connection_stream = connection.track_stream(); let mut headers = HeaderMap::new(); headers.insert( diff --git a/packages/app-lib/src/util/download/h2_pool.rs b/packages/app-lib/src/util/download/h2_pool.rs index 0cc405d..5966031 100644 --- a/packages/app-lib/src/util/download/h2_pool.rs +++ b/packages/app-lib/src/util/download/h2_pool.rs @@ -34,6 +34,7 @@ const CONNECTION_WAIT_TIMEOUT: Duration = Duration::from_secs(45); #[derive(Debug, Clone, Copy, Eq, PartialEq)] pub(crate) enum H2ConnectFailureKind { + Capacity, Tcp, Tls, Protocol, @@ -357,9 +358,41 @@ async fn connect_tcp(host: &str, port: u16) -> std::io::Result { } /// Connects a new shared HTTP/2 connection to `authority` (host[:port]). -async fn establish( +async fn reserve_physical_budget( route: &DownloadRoute, reserve_native_budget: bool, +) -> Result, H2ConnectError> { + Ok(if reserve_native_budget { + Some( + crate::util::fetch::wait_for_download_resource( + super::native_budget::acquire(route), + ) + .await + .map_err(|_| { + H2ConnectError::new( + H2ConnectFailureKind::Capacity, + "HTTP/2 connection capacity remained stalled for 5 minutes" + .to_string(), + ) + })? + .map_err(|error| { + H2ConnectError::new( + H2ConnectFailureKind::Capacity, + format!( + "failed to reserve HTTP/2 connection capacity: {error}" + ), + ) + })?, + ) + } else { + None + }) +} + +/// Connects a new shared HTTP/2 connection to `authority` (host[:port]). +async fn establish( + route: &DownloadRoute, + physical_budget: Option, ) -> Result, H2ConnectError> { let authority = crate::util::fetch::url_authority(&route.url).ok_or_else(|| { @@ -368,16 +401,6 @@ async fn establish( "HTTP/2 route has no authority".to_string(), ) })?; - let physical_budget = if reserve_native_budget { - Some(super::native_budget::acquire(route).await.map_err(|error| { - H2ConnectError::new( - H2ConnectFailureKind::Tcp, - format!("failed to reserve HTTP/2 connection capacity: {error}"), - ) - })?) - } else { - None - }; let (host, port) = authority .rsplit_once(':') .map(|(host, port)| (host, port.parse::().unwrap_or(443))) @@ -487,16 +510,16 @@ pub(crate) async fn shared_connection( ) })?; let slot = connection_slot(&authority).await; - let mut cached = tokio::time::timeout(CONNECTION_WAIT_TIMEOUT, slot.lock()) - .await - .map_err(|_| { - H2ConnectError::new( - H2ConnectFailureKind::Protocol, - format!( - "timed out waiting for HTTP/2 connection slot {authority}" - ), - ) - })?; + let mut cached = crate::util::fetch::wait_for_download_resource( + slot.lock(), + ) + .await + .map_err(|_| { + H2ConnectError::new( + H2ConnectFailureKind::Capacity, + format!("HTTP/2 connection slot remained stalled for {authority}"), + ) + })?; if let Some(connection) = cached.as_ref().filter(|connection| { !connection.is_dead() && !connection.is_idle_expired() }) { @@ -525,9 +548,11 @@ pub(crate) async fn shared_connection( )); } tracing::debug!(authority, "Establishing cold shared HTTP/2 connection"); + let physical_budget = + reserve_physical_budget(route, reserve_native_budget).await?; let connection = tokio::time::timeout( CONNECTION_WAIT_TIMEOUT, - establish(route, reserve_native_budget), + establish(route, physical_budget), ) .await .map_err(|_| { @@ -555,15 +580,16 @@ pub(crate) async fn shared_batch_connection( ) })?; let slot = batch_connection_slot(&authority).await; - let mut cached = tokio::time::timeout( - CONNECTION_WAIT_TIMEOUT, + let mut cached = crate::util::fetch::wait_for_download_resource( slot.lock(), ) .await .map_err(|_| { H2ConnectError::new( - H2ConnectFailureKind::Protocol, - format!("timed out waiting for asset HTTP/2 connection slot {authority}"), + H2ConnectFailureKind::Capacity, + format!( + "asset HTTP/2 connection slot remained stalled for {authority}" + ), ) })?; if let Some(connection) = cached.as_ref().filter(|connection| { @@ -591,9 +617,11 @@ pub(crate) async fn shared_batch_connection( *cached = None; } tracing::debug!(authority, "Establishing sibling HTTP/2 asset connection"); + let physical_budget = + reserve_physical_budget(route, reserve_native_budget).await?; let connection = tokio::time::timeout( CONNECTION_WAIT_TIMEOUT, - establish(route, reserve_native_budget), + establish(route, physical_budget), ) .await .map_err(|_| { @@ -614,10 +642,13 @@ pub(crate) async fn has_live_connection(authority: &str) -> bool { return false; }; drop(connections); - let live = slot - .lock() - .await + let Ok(cached) = slot.try_lock() else { + // This is only an eligibility probe. A busy slot means another task + // is establishing or refreshing the connection; do not queue this + // caller behind that work before its cancellation-aware download path. + return false; + }; + cached .as_ref() - .is_some_and(|connection| !connection.is_dead()); - live + .is_some_and(|connection| !connection.is_dead()) } diff --git a/packages/app-lib/src/util/download/h2_range.rs b/packages/app-lib/src/util/download/h2_range.rs index 532fbe4..bc6b1d4 100644 --- a/packages/app-lib/src/util/download/h2_range.rs +++ b/packages/app-lib/src/util/download/h2_range.rs @@ -218,13 +218,9 @@ async fn download_range( crate::install::DownloadItemStatus::WaitingForResource, ) .await; - let _permit = tokio::time::timeout( - Duration::from_secs(45), - super::h2_stream_budget::acquire(route), - ) - .await - .map_err(|_| H2DownloadFailure::Connect)? - .map_err(|_| H2DownloadFailure::Connect)?; + let _permit = super::h2_stream_budget::acquire_with_pressure(route) + .await + .map_err(|_| H2DownloadFailure::Connect)?; super::h2_download::record_install_stage( request, crate::install::DownloadItemStatus::Downloading, diff --git a/packages/app-lib/src/util/download/h2_stream_budget.rs b/packages/app-lib/src/util/download/h2_stream_budget.rs index d85edf8..5ad9cba 100644 --- a/packages/app-lib/src/util/download/h2_stream_budget.rs +++ b/packages/app-lib/src/util/download/h2_stream_budget.rs @@ -104,6 +104,20 @@ pub(crate) async fn acquire( } } +pub(crate) async fn acquire_with_pressure( + route: &DownloadRoute, +) -> crate::Result { + crate::util::fetch::wait_for_download_resource(acquire(route)) + .await + .map_err(|_| { + crate::Error::from(crate::ErrorKind::NetworkError( + "HTTP/2 stream permit remained stalled for 5 minutes" + .to_string(), + )) + })? + .map_err(Into::into) +} + /// Acquires a stream for the Minecraft asset batch. Assets retain a large /// logical worker window, but use a separate physical stream pool so they /// cannot consume the ordinary native content share. @@ -134,10 +148,25 @@ pub(crate) async fn acquire_asset( } } +pub(crate) async fn acquire_asset_with_pressure( + route: &DownloadRoute, +) -> crate::Result { + crate::util::fetch::wait_for_download_resource(acquire_asset(route)) + .await + .map_err(|_| { + crate::Error::from(crate::ErrorKind::NetworkError( + "asset HTTP/2 stream permit remained stalled for 5 minutes" + .to_string(), + )) + })? + .map_err(Into::into) +} + #[cfg(test)] mod tests { use super::*; use crate::util::fetch::DownloadRouteSource; + use std::sync::atomic::{AtomicUsize, Ordering}; fn route() -> DownloadRoute { DownloadRoute { @@ -190,4 +219,40 @@ mod tests { MAX_H2_STREAMS_PER_AUTHORITY ); } + + #[tokio::test] + async fn contended_stream_wait_reports_pressure_and_resumes() { + let route = DownloadRoute { + url: "https://h2-pressure.example/file".to_string(), + source: DownloadRouteSource::Official, + is_mirror: false, + allow_sensitive_headers: true, + supports_range: true, + proxy: ProxyPolicy::Direct, + }; + let mut held = Vec::new(); + for _ in 0..MAX_H2_STREAMS_PER_AUTHORITY { + held.push(acquire(&route).await.unwrap()); + } + let pressure = Arc::new(AtomicUsize::new(0)); + let observed = Arc::clone(&pressure); + let release = tokio::spawn(async move { + tokio::time::sleep(Duration::from_millis(35)).await; + drop(held); + }); + + let permit = crate::util::fetch::wait_for_resource_with_pressure( + acquire(&route), + Duration::from_millis(5), + Duration::from_millis(100), + move || { + observed.fetch_add(1, Ordering::Relaxed); + }, + ) + .await; + + release.await.unwrap(); + assert!(permit.is_ok()); + assert!(pressure.load(Ordering::Relaxed) >= 1); + } } diff --git a/packages/app-lib/src/util/download/mod.rs b/packages/app-lib/src/util/download/mod.rs index 743b7bf..34f895c 100644 --- a/packages/app-lib/src/util/download/mod.rs +++ b/packages/app-lib/src/util/download/mod.rs @@ -73,6 +73,19 @@ pub(crate) fn task_concurrency_limit(state: &crate::State) -> Option { if active_engine() == DownloadEngine::XmclCompat { None } else { - Some(state.download_concurrency()) + Some(native_budget::task_limit( + state.download_concurrency(), + usize::MAX, + )) + } +} + +pub(crate) fn single_authority_task_concurrency_limit( + state: &crate::State, +) -> Option { + if active_engine() == DownloadEngine::XmclCompat { + None + } else { + Some(native_budget::task_limit(state.download_concurrency(), 1)) } } diff --git a/packages/app-lib/src/util/download/native_budget.rs b/packages/app-lib/src/util/download/native_budget.rs index 7eb3340..6f9a35c 100644 --- a/packages/app-lib/src/util/download/native_budget.rs +++ b/packages/app-lib/src/util/download/native_budget.rs @@ -144,6 +144,15 @@ pub(crate) fn available(route: &DownloadRoute) -> usize { .unwrap_or(MAX_CONNECTIONS_PER_AUTHORITY) } +pub(crate) fn task_limit( + configured_limit: usize, + authority_count: usize, +) -> usize { + configured_limit.max(1).min(MAX_NATIVE_CONNECTIONS).min( + MAX_CONNECTIONS_PER_AUTHORITY.saturating_mul(authority_count.max(1)), + ) +} + #[cfg(test)] mod tests { use super::*; @@ -174,4 +183,11 @@ mod tests { drop(permits); assert!(try_acquire(&route).is_ok()); } + + #[test] + fn task_limit_matches_native_connection_capacity() { + assert_eq!(task_limit(128, 1), 8); + assert_eq!(task_limit(128, 4), 32); + assert_eq!(task_limit(4, 1), 4); + } } diff --git a/packages/app-lib/src/util/fetch.rs b/packages/app-lib/src/util/fetch.rs index c174677..72df1ba 100644 --- a/packages/app-lib/src/util/fetch.rs +++ b/packages/app-lib/src/util/fetch.rs @@ -85,6 +85,8 @@ const MAX_REDIRECT_LOCATION_BYTES: usize = 8 * 1024; const FILE_TRANSFER_CONNECT_TIMEOUT: time::Duration = time::Duration::from_secs(20); const RESOURCE_WAIT_TIMEOUT: time::Duration = time::Duration::from_secs(45); +const RESOURCE_PRESSURE_INTERVAL: time::Duration = time::Duration::from_secs(3); +const RESOURCE_STALL_TIMEOUT: time::Duration = time::Duration::from_secs(300); #[cfg(not(test))] const FILE_TRANSFER_READ_TIMEOUT: time::Duration = time::Duration::from_secs(60); @@ -3894,6 +3896,73 @@ struct NativeConnectionPermit<'a> { _native: crate::util::download::native_budget::NativeBudgetPermit, } +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub(crate) enum ResourceWaitError { + TimedOut, +} + +pub(crate) async fn wait_for_resource_with_pressure( + future: F, + pressure_interval: time::Duration, + stall_timeout: time::Duration, + mut on_pressure: P, +) -> Result +where + F: Future, + P: FnMut(), +{ + let pressure_interval = + pressure_interval.max(time::Duration::from_millis(1)); + let pressure_timer = tokio::time::sleep(pressure_interval); + let stall_timer = tokio::time::sleep(stall_timeout); + tokio::pin!(future, pressure_timer, stall_timer); + + loop { + tokio::select! { + result = &mut future => return Ok(result), + _ = &mut stall_timer => return Err(ResourceWaitError::TimedOut), + _ = &mut pressure_timer => { + on_pressure(); + pressure_timer.as_mut().reset( + tokio::time::Instant::now() + pressure_interval, + ); + } + } + } +} + +pub(crate) async fn wait_for_download_resource( + future: F, +) -> Result +where + F: Future, +{ + wait_for_resource_with_pressure( + future, + RESOURCE_PRESSURE_INTERVAL, + RESOURCE_STALL_TIMEOUT, + || { + if let Some(state) = crate::State::get_if_initialized() { + state.record_download_queue_wait(); + } + }, + ) + .await +} + +pub(crate) async fn acquire_fetch_permit_with_pressure( + semaphore: &FetchSemaphore, +) -> crate::Result> { + wait_for_download_resource(semaphore.0.acquire()) + .await + .map_err(|_| { + crate::Error::from(ErrorKind::NetworkError( + "download permit remained stalled for 5 minutes".to_string(), + )) + })? + .map_err(Into::into) +} + async fn acquire_native_connection<'a>( route: &DownloadRoute, semaphore: &'a FetchSemaphore, @@ -3906,6 +3975,20 @@ async fn acquire_native_connection<'a>( }) } +async fn acquire_native_connection_with_pressure<'a>( + route: &DownloadRoute, + semaphore: &'a FetchSemaphore, +) -> crate::Result> { + wait_for_download_resource(acquire_native_connection(route, semaphore)) + .await + .map_err(|_| { + crate::Error::from(ErrorKind::NetworkError( + "native download resources remained stalled for 5 minutes" + .to_string(), + )) + })? +} + fn try_acquire_native_connection<'a>( route: &DownloadRoute, semaphore: &'a FetchSemaphore, @@ -5703,19 +5786,11 @@ async fn download_to_path_inner( None }; if let Some((h2_route, h2_policy)) = h2_selection { - let h2_permit = - tokio::time::timeout(RESOURCE_WAIT_TIMEOUT, semaphore.0.acquire()) - .await - .map_err(|_| { - ErrorKind::NetworkError( - "timed out waiting for HTTP/2 download permit" - .to_string(), - ) - })??; let h2_started = Instant::now(); match crate::util::download::h2_download::try_download_via_h2( &request, &h2_route, + semaphore, destination, &part_path, h2_policy, @@ -5808,7 +5883,6 @@ async fn download_to_path_inner( .await?; } } - drop(h2_permit); } let mut official_integrity_retry = h2_failed_nonofficial.is_some(); @@ -6147,8 +6221,6 @@ async fn download_to_path_inner( } else { 0 }; - let mut activity = crate::State::get_if_initialized() - .map(|state| state.begin_download_connection()); record_install_download_started( &request, route, @@ -6161,25 +6233,21 @@ async fn download_to_path_inner( DownloadItemStatus::WaitingForResource, ) .await; - let permit_wait = tokio::time::timeout( - RESOURCE_WAIT_TIMEOUT, - acquire_native_connection(route, semaphore), - ); + let permit_wait = + acquire_native_connection_with_pressure(route, semaphore); let resource_wait_started = Instant::now(); - let permit = if let Some(cancellation) = request.cancellation.as_ref() { + let permit = if let Some(cancellation) = + request.cancellation.as_ref() + { tokio::select! { _ = cancellation.cancelled() => return Err(ErrorKind::OtherError("download canceled while waiting for native resources".to_string()).into()), result = permit_wait => result, } } else { permit_wait.await - } - .map_err(|_| { - ErrorKind::NetworkError( - "timed out waiting for native download resources" - .to_string(), - ) - })??; + }?; + let mut activity = crate::State::get_if_initialized() + .map(|state| state.begin_download_connection()); tracing::debug!( route = %sanitize_url_for_log(&route.url), resource = "native_connection_and_fetch", @@ -7135,6 +7203,40 @@ mod tests { static H2_FALLBACK_TEST_LOCK: LazyLock> = LazyLock::new(|| std::sync::Mutex::new(())); + #[tokio::test] + async fn resource_wait_reports_pressure_without_losing_queue_position() { + let pressure_events = Arc::new(AtomicUsize::new(0)); + let observed = Arc::clone(&pressure_events); + let result = wait_for_resource_with_pressure( + async { + tokio::time::sleep(Duration::from_millis(35)).await; + 7 + }, + Duration::from_millis(5), + Duration::from_millis(100), + move || { + observed.fetch_add(1, Ordering::Relaxed); + }, + ) + .await; + + assert_eq!(result, Ok(7)); + assert!(pressure_events.load(Ordering::Relaxed) >= 1); + } + + #[tokio::test] + async fn resource_wait_keeps_a_final_stall_guard() { + let result = wait_for_resource_with_pressure( + std::future::pending::<()>(), + Duration::from_millis(5), + Duration::from_millis(25), + || {}, + ) + .await; + + assert_eq!(result, Err(ResourceWaitError::TimedOut)); + } + async fn spawn_range_server( data: Arc>, wrong_content_range: bool,