fix: 优化下载并发调度，发布 1.0.2-beta4

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::<Uri>() 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(

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<TcpStream> {
 }
 
 /// 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<Option<super::native_budget::NativeBudgetPermit>, 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<super::native_budget::NativeBudgetPermit>,
 ) -> Result<Arc<SharedH2Connection>, 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::<u16>().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())
 }

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,

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<H2StreamPermit> {
+    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<H2StreamPermit> {
+    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);
+    }
 }

packages/app-lib/src/util/download/mod.rs

@@ -73,6 +73,19 @@ pub(crate) fn task_concurrency_limit(state: &crate::State) -> Option<usize> {
     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<usize> {
+    if active_engine() == DownloadEngine::XmclCompat {
+        None
+    } else {
+        Some(native_budget::task_limit(state.download_concurrency(), 1))
     }
 }

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);
+    }
 }