feat：移除了弹窗，服务器添加sls

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

@@ -0,0 +1,623 @@
+//! Shared per-authority HTTP/2 connections for file downloads.
+//!
+//! `reqwest`'s connection pool opens a fresh TCP+TLS connection for every
+//! request that arrives while no idle connection is available, so a batch of
+//! concurrent downloads to one CDN costs one handshake per file. This module
+//! instead maintains a long-lived HTTP/2 connection per authority and
+//! multiplexes every download as a separate stream over it (`SendRequest` is
+//! cheap to clone and each clone opens an independent stream). Asset batches
+//! may lazily add one sibling under sustained saturation; large files can also
+//! split into range streams over their shared connection.
+
+use bytes::Bytes;
+use h2::client::SendRequest;
+use rustls::ClientConfig;
+use rustls_pki_types::ServerName;
+use std::collections::HashMap;
+use std::net::IpAddr;
+use std::sync::atomic::{AtomicUsize, Ordering};
+use std::sync::{Arc, Mutex, Once};
+use std::time::Duration;
+use tokio::net::TcpStream;
+use tokio::sync::Mutex as AsyncMutex;
+use tokio::sync::Notify;
+use tokio_rustls::TlsConnector;
+
+use crate::util::fetch::DownloadRoute;
+
+const CONNECT_TIMEOUT: Duration = Duration::from_secs(10);
+const TLS_HANDSHAKE_TIMEOUT: Duration = Duration::from_secs(10);
+const DNS_PREWARM_TIMEOUT: Duration = Duration::from_secs(10);
+const STREAM_READY_TIMEOUT: Duration = Duration::from_secs(30);
+const IDLE_EVICTION_TIMEOUT: Duration = Duration::from_secs(5 * 60);
+const CONNECTION_WAIT_TIMEOUT: Duration = Duration::from_secs(45);
+
+#[derive(Debug, Clone, Copy, Eq, PartialEq)]
+pub(crate) enum H2ConnectFailureKind {
+    Tcp,
+    Tls,
+    Protocol,
+}
+
+#[derive(Debug)]
+pub(crate) struct H2ConnectError {
+    pub(crate) kind: H2ConnectFailureKind,
+    detail: String,
+}
+
+impl std::fmt::Display for H2ConnectError {
+    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        formatter.write_str(&self.detail)
+    }
+}
+
+impl std::error::Error for H2ConnectError {}
+
+impl H2ConnectError {
+    fn new(kind: H2ConnectFailureKind, detail: String) -> Self {
+        Self { kind, detail }
+    }
+}
+
+/// A live shared HTTP/2 connection to one authority.
+pub struct SharedH2Connection {
+    authority: String,
+    sender: Mutex<SendRequest<Bytes>>,
+    // One permit represents the shared TCP/TLS connection, not every H2
+    // stream opened through it.
+    physical_budget: Mutex<Option<super::native_budget::NativeBudgetPermit>>,
+    /// Set to true by the driver task when the connection terminates.
+    dead: Arc<std::sync::atomic::AtomicBool>,
+    /// Number of application streams currently assigned to this connection.
+    /// This is deliberately separate from HTTP/2's peer stream accounting: it
+    /// lets an asset batch distribute work across sibling TCP connections.
+    active_streams: Arc<AtomicUsize>,
+    last_activity: Mutex<std::time::Instant>,
+    evict: Arc<Notify>,
+}
+
+pub(crate) struct H2StreamActivity {
+    active_streams: Arc<AtomicUsize>,
+}
+
+impl Drop for H2StreamActivity {
+    fn drop(&mut self) {
+        self.active_streams.fetch_sub(1, Ordering::Release);
+    }
+}
+
+impl SharedH2Connection {
+    fn new(
+        authority: String,
+        sender: SendRequest<Bytes>,
+        physical_budget: Option<super::native_budget::NativeBudgetPermit>,
+    ) -> Self {
+        Self {
+            authority,
+            sender: Mutex::new(sender),
+            physical_budget: Mutex::new(physical_budget),
+            dead: Arc::new(std::sync::atomic::AtomicBool::new(false)),
+            active_streams: Arc::new(AtomicUsize::new(0)),
+            last_activity: Mutex::new(std::time::Instant::now()),
+            evict: Arc::new(Notify::new()),
+        }
+    }
+
+    #[cfg(test)]
+    pub(crate) fn for_test(sender: SendRequest<Bytes>) -> Self {
+        Self::new("test.invalid:443".to_string(), sender, None)
+    }
+
+    pub fn is_dead(&self) -> bool {
+        self.dead.load(std::sync::atomic::Ordering::Acquire)
+    }
+
+    pub(crate) fn active_streams(&self) -> usize {
+        self.active_streams.load(Ordering::Acquire)
+    }
+
+    pub(crate) fn track_stream(&self) -> H2StreamActivity {
+        *self
+            .last_activity
+            .lock()
+            .unwrap_or_else(|poisoned| poisoned.into_inner()) =
+            std::time::Instant::now();
+        self.active_streams.fetch_add(1, Ordering::AcqRel);
+        H2StreamActivity {
+            active_streams: Arc::clone(&self.active_streams),
+        }
+    }
+
+    #[cfg(test)]
+    fn mark_idle_for_test(&self) {
+        *self
+            .last_activity
+            .lock()
+            .unwrap_or_else(|poisoned| poisoned.into_inner()) =
+            std::time::Instant::now() - IDLE_EVICTION_TIMEOUT;
+    }
+
+    fn is_idle_expired(&self) -> bool {
+        self.active_streams() == 0
+            && self
+                .last_activity
+                .lock()
+                .unwrap_or_else(|poisoned| poisoned.into_inner())
+                .elapsed()
+                >= IDLE_EVICTION_TIMEOUT
+    }
+
+    fn evict(&self) {
+        self.evict.notify_waiters();
+    }
+
+    fn has_physical_budget(&self) -> bool {
+        self.physical_budget
+            .lock()
+            .unwrap_or_else(|poisoned| poisoned.into_inner())
+            .is_some()
+    }
+
+    fn release_physical_budget(&self) {
+        self.physical_budget
+            .lock()
+            .unwrap_or_else(|poisoned| poisoned.into_inner())
+            .take();
+    }
+
+    /// Sends a request on the shared connection and awaits the response
+    /// headers, yielding the response and its receive stream. Each call
+    /// opens an independent multiplexed stream.
+    pub async fn open(
+        &self,
+        request: http::Request<()>,
+    ) -> Result<http::Response<h2::RecvStream>, h2::Error> {
+        let ready_started = std::time::Instant::now();
+        let sender = self
+            .sender
+            .lock()
+            .unwrap_or_else(|poisoned| poisoned.into_inner())
+            .clone();
+        *self
+            .last_activity
+            .lock()
+            .unwrap_or_else(|poisoned| poisoned.into_inner()) =
+            std::time::Instant::now();
+        let mut sender =
+            tokio::time::timeout(STREAM_READY_TIMEOUT, sender.ready())
+                .await
+                .map_err(|_| {
+                    h2::Error::from(h2::Reason::ENHANCE_YOUR_CALM)
+                })??;
+        let ready_wait = ready_started.elapsed();
+        if ready_wait >= Duration::from_millis(25) {
+            tracing::debug!(
+                authority = %self.authority,
+                ready_wait_ms = ready_wait.as_millis(),
+                "HTTP/2 stream admission waited for peer or connection capacity"
+            );
+        }
+        let (response, _) = sender.send_request(request, true)?;
+        tokio::time::timeout(STREAM_READY_TIMEOUT, response)
+            .await
+            .map_err(|_| h2::Error::from(h2::Reason::ENHANCE_YOUR_CALM))?
+    }
+}
+
+type ConnectionSlot = Arc<AsyncMutex<Option<Arc<SharedH2Connection>>>>;
+
+/// Registry of live shared connections, keyed by authority.
+static CONNECTIONS: std::sync::LazyLock<
+    AsyncMutex<HashMap<String, ConnectionSlot>>,
+> = std::sync::LazyLock::new(|| AsyncMutex::new(HashMap::new()));
+
+/// A bounded sibling connection for saturated asset batches. Normal file
+/// downloads always use `CONNECTIONS`; a second TCP congestion domain is only
+/// created by the asset scheduler after it observes sustained pressure.
+static BATCH_CONNECTIONS: std::sync::LazyLock<
+    AsyncMutex<HashMap<String, ConnectionSlot>>,
+> = std::sync::LazyLock::new(|| AsyncMutex::new(HashMap::new()));
+
+async fn connection_slot(authority: &str) -> ConnectionSlot {
+    let mut connections = CONNECTIONS.lock().await;
+    connections
+        .entry(authority.to_string())
+        .or_insert_with(|| Arc::new(AsyncMutex::new(None)))
+        .clone()
+}
+
+async fn batch_connection_slot(authority: &str) -> ConnectionSlot {
+    let mut connections = BATCH_CONNECTIONS.lock().await;
+    connections
+        .entry(authority.to_string())
+        .or_insert_with(|| Arc::new(AsyncMutex::new(None)))
+        .clone()
+}
+
+fn tls_config() -> Arc<ClientConfig> {
+    static INSTALL_PROVIDER: Once = Once::new();
+    INSTALL_PROVIDER.call_once(|| {
+        let _ = rustls::crypto::ring::default_provider().install_default();
+    });
+    static CONFIG: std::sync::LazyLock<Mutex<Option<Arc<ClientConfig>>>> =
+        std::sync::LazyLock::new(|| Mutex::new(None));
+    let mut guard = CONFIG
+        .lock()
+        .unwrap_or_else(|poisoned| poisoned.into_inner());
+    if let Some(config) = guard.as_ref() {
+        return Arc::clone(config);
+    }
+    let mut config = ClientConfig::builder()
+        .with_root_certificates(platform_root_certs())
+        .with_no_client_auth();
+    config.enable_early_data = true;
+    config.alpn_protocols = vec![b"h2".to_vec()];
+    let config = Arc::new(config);
+    *guard = Some(Arc::clone(&config));
+    config
+}
+
+fn platform_root_certs() -> rustls::RootCertStore {
+    let mut store = rustls::RootCertStore::empty();
+    let certs = rustls_native_certs::load_native_certs();
+    for cert in certs.certs {
+        let _ = store.add(cert);
+    }
+    if certs.errors.is_empty() {
+        return store;
+    }
+    store.extend(webpki_roots::TLS_SERVER_ROOTS.iter().cloned());
+    store
+}
+
+async fn connect_addresses(
+    host: &str,
+    port: u16,
+    addresses: &[IpAddr],
+) -> std::io::Result<TcpStream> {
+    let mut last_error = None;
+    for address in addresses {
+        match tokio::time::timeout(
+            CONNECT_TIMEOUT,
+            TcpStream::connect((*address, port)),
+        )
+        .await
+        {
+            Ok(Ok(stream)) => {
+                stream.set_nodelay(true).ok();
+                return Ok(stream);
+            }
+            Ok(Err(error)) => last_error = Some(error),
+            Err(_) => {
+                last_error = Some(std::io::Error::new(
+                    std::io::ErrorKind::TimedOut,
+                    format!("connection to {host}:{port} timed out"),
+                ));
+            }
+        }
+    }
+    Err(last_error.unwrap_or_else(|| {
+        std::io::Error::new(
+            std::io::ErrorKind::NotFound,
+            format!("no addresses available for {host}"),
+        )
+    }))
+}
+
+async fn connect_tcp(host: &str, port: u16) -> std::io::Result<TcpStream> {
+    // Prefer the ordered address list from the shared download resolver
+    // (IPv4/IPv6 preference and per-IP reliability), falling back to the
+    // system resolver when no list is cached yet.
+    let resolver = &crate::util::fetch::DOWNLOAD_DNS_RESOLVER;
+    let addresses = resolver.resolved_addresses(host);
+    let mut last_error = None;
+    if !addresses.is_empty() {
+        match connect_addresses(host, port, &addresses).await {
+            Ok(stream) => return Ok(stream),
+            Err(error) => last_error = Some(error),
+        }
+        if resolver.record_connection_failure(host) {
+            let _ = tokio::time::timeout(
+                CONNECTION_WAIT_TIMEOUT,
+                resolver.pre_resolve(host),
+            )
+            .await;
+            let refreshed = resolver.resolved_addresses(host);
+            if !refreshed.is_empty() && refreshed != addresses {
+                match connect_addresses(host, port, &refreshed).await {
+                    Ok(stream) => return Ok(stream),
+                    Err(error) => last_error = Some(error),
+                }
+            }
+        }
+    }
+    let stream = tokio::time::timeout(
+        CONNECT_TIMEOUT,
+        tokio::net::TcpStream::connect((host, port)),
+    )
+    .await
+    .map_err(|_| {
+        last_error.take().unwrap_or_else(|| {
+            std::io::Error::new(
+                std::io::ErrorKind::TimedOut,
+                format!("connection to {host}:{port} timed out"),
+            )
+        })
+    })?
+    .map_err(|error| {
+        last_error.take().unwrap_or_else(|| {
+            std::io::Error::new(
+                error.kind(),
+                format!("connection to {host}:{port} failed: {error}"),
+            )
+        })
+    })?;
+    stream.set_nodelay(true).ok();
+    Ok(stream)
+}
+
+/// Connects a new shared HTTP/2 connection to `authority` (host[:port]).
+async fn establish(
+    route: &DownloadRoute,
+    reserve_native_budget: bool,
+) -> Result<Arc<SharedH2Connection>, H2ConnectError> {
+    let authority =
+        crate::util::fetch::url_authority(&route.url).ok_or_else(|| {
+            H2ConnectError::new(
+                H2ConnectFailureKind::Protocol,
+                "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)))
+        .unwrap_or((&authority, 443));
+
+    // Pre-resolve so `connect_tcp` gets the ordered, reliability-ranked
+    // address list shared with the legacy reqwest path.
+    let _ = tokio::time::timeout(
+        DNS_PREWARM_TIMEOUT,
+        crate::util::fetch::DOWNLOAD_DNS_RESOLVER.pre_resolve(host),
+    )
+    .await;
+
+    let tcp = connect_tcp(host, port).await.map_err(|error| {
+		H2ConnectError::new(
+			H2ConnectFailureKind::Tcp,
+			format!(
+				"failed to establish shared HTTP/2 connection to {authority}: {error}"
+			),
+		)
+	})?;
+
+    let server_name =
+        ServerName::try_from(host.to_string()).map_err(|error| {
+            H2ConnectError::new(
+                H2ConnectFailureKind::Tls,
+                format!("invalid server name for {host}: {error}"),
+            )
+        })?;
+    let connector = TlsConnector::from(tls_config());
+    let tls = tokio::time::timeout(
+        TLS_HANDSHAKE_TIMEOUT,
+        connector.connect(server_name, tcp),
+    )
+    .await
+    .map_err(|_| {
+        H2ConnectError::new(
+            H2ConnectFailureKind::Tls,
+            format!("TLS handshake with {authority} timed out"),
+        )
+    })?
+    .map_err(|error| {
+        H2ConnectError::new(
+            H2ConnectFailureKind::Tls,
+            format!("TLS handshake with {authority} failed: {error}"),
+        )
+    })?;
+
+    let mut builder = h2::client::Builder::new();
+    builder
+        .initial_window_size(1024 * 1024)
+        .initial_connection_window_size(64 * 1024 * 1024);
+    let (sender, mut connection) =
+        builder.handshake(Box::pin(tls)).await.map_err(|error| {
+            H2ConnectError::new(
+                H2ConnectFailureKind::Protocol,
+                format!("HTTP/2 handshake with {authority} failed: {error}"),
+            )
+        })?;
+
+    // Tune flow-control windows for high-stream multiplexing (e.g. hundreds
+    // of concurrent asset downloads over one connection). With the default
+    // 64 KiB connection window, concurrent streams would stall waiting for
+    // connection-level window updates; a larger per-stream window also lets
+    // the peer send a whole small file without round trips.
+    // Keep the adaptive connection target aligned with the initial connection
+    // window configured before the handshake.
+    connection.set_target_window_size(64 * 1024 * 1024);
+
+    let shared = Arc::new(SharedH2Connection::new(
+        authority.clone(),
+        sender,
+        physical_budget,
+    ));
+
+    let dead = Arc::clone(&shared.dead);
+    let connection_budget = Arc::clone(&shared);
+    let evict = Arc::clone(&shared.evict);
+    let authority = authority.to_string();
+    tokio::spawn(async move {
+        tokio::select! {
+            _ = connection => {}
+            _ = evict.notified() => {
+                tracing::debug!(authority, "Evicting idle shared HTTP/2 connection");
+            }
+        }
+        dead.store(true, std::sync::atomic::Ordering::Release);
+        connection_budget.release_physical_budget();
+        tracing::debug!(authority, "Shared HTTP/2 connection closed");
+    });
+
+    Ok(shared)
+}
+
+/// Returns the live shared connection for `authority`, establishing one on
+/// first use or after a previous connection died.
+pub(crate) async fn shared_connection(
+    route: &DownloadRoute,
+    reserve_native_budget: bool,
+    allow_cold_connection: bool,
+) -> Result<Arc<SharedH2Connection>, H2ConnectError> {
+    let authority =
+        crate::util::fetch::url_authority(&route.url).ok_or_else(|| {
+            H2ConnectError::new(
+                H2ConnectFailureKind::Protocol,
+                "HTTP/2 route has no authority".to_string(),
+            )
+        })?;
+    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}"
+                ),
+            )
+        })?;
+    if let Some(connection) = cached.as_ref().filter(|connection| {
+        !connection.is_dead() && !connection.is_idle_expired()
+    }) {
+        if !reserve_native_budget || connection.has_physical_budget() {
+            tracing::debug!(authority, "Reusing shared HTTP/2 connection");
+            return Ok(Arc::clone(connection));
+        }
+        return Err(H2ConnectError::new(
+            H2ConnectFailureKind::Protocol,
+            "shared HTTP/2 connection is not covered by the native connection budget"
+                .to_string(),
+        ));
+    }
+    if cached.as_ref().is_some_and(|connection| {
+        connection.is_dead() || connection.is_idle_expired()
+    }) {
+        if let Some(connection) = cached.as_ref() {
+            connection.evict();
+        }
+        *cached = None;
+    }
+    if !allow_cold_connection {
+        return Err(H2ConnectError::new(
+            H2ConnectFailureKind::Protocol,
+            "HTTP/2 policy requires an existing shared connection".to_string(),
+        ));
+    }
+    tracing::debug!(authority, "Establishing cold shared HTTP/2 connection");
+    let connection = tokio::time::timeout(
+        CONNECTION_WAIT_TIMEOUT,
+        establish(route, reserve_native_budget),
+    )
+    .await
+    .map_err(|_| {
+        H2ConnectError::new(
+            H2ConnectFailureKind::Tcp,
+            format!("timed out establishing HTTP/2 connection to {authority}"),
+        )
+    })??;
+    *cached = Some(Arc::clone(&connection));
+    Ok(connection)
+}
+
+/// Returns the optional second connection used exclusively by a busy asset
+/// batch. It is stored independently so ordinary file downloads retain their
+/// stable primary connection and never create extra TCP connections.
+pub(crate) async fn shared_batch_connection(
+    route: &DownloadRoute,
+    reserve_native_budget: bool,
+) -> Result<Arc<SharedH2Connection>, H2ConnectError> {
+    let authority =
+        crate::util::fetch::url_authority(&route.url).ok_or_else(|| {
+            H2ConnectError::new(
+                H2ConnectFailureKind::Protocol,
+                "HTTP/2 route has no authority".to_string(),
+            )
+        })?;
+    let slot = batch_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 asset HTTP/2 connection slot {authority}"),
+        )
+    })?;
+    if let Some(connection) = cached.as_ref().filter(|connection| {
+        !connection.is_dead() && !connection.is_idle_expired()
+    }) {
+        if !reserve_native_budget || connection.has_physical_budget() {
+            tracing::debug!(
+                authority,
+                "Reusing sibling HTTP/2 asset connection"
+            );
+            return Ok(Arc::clone(connection));
+        }
+        return Err(H2ConnectError::new(
+            H2ConnectFailureKind::Protocol,
+            "sibling HTTP/2 connection is not covered by the native connection budget"
+                .to_string(),
+        ));
+    }
+    if cached.as_ref().is_some_and(|connection| {
+        connection.is_dead() || connection.is_idle_expired()
+    }) {
+        if let Some(connection) = cached.as_ref() {
+            connection.evict();
+        }
+        *cached = None;
+    }
+    tracing::debug!(authority, "Establishing sibling HTTP/2 asset connection");
+    let connection = tokio::time::timeout(
+        CONNECTION_WAIT_TIMEOUT,
+        establish(route, reserve_native_budget),
+    )
+    .await
+    .map_err(|_| {
+        H2ConnectError::new(
+            H2ConnectFailureKind::Tcp,
+            format!(
+                "timed out establishing asset HTTP/2 connection to {authority}"
+            ),
+        )
+    })??;
+    *cached = Some(Arc::clone(&connection));
+    Ok(connection)
+}
+
+pub(crate) async fn has_live_connection(authority: &str) -> bool {
+    let connections = CONNECTIONS.lock().await;
+    let Some(slot) = connections.get(authority).cloned() else {
+        return false;
+    };
+    drop(connections);
+    let live = slot
+        .lock()
+        .await
+        .as_ref()
+        .is_some_and(|connection| !connection.is_dead());
+    live
+}

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

@@ -0,0 +1,533 @@
+//! Multi-range download of one file over a shared HTTP/2 connection.
+
+use super::h2_download::{H2DownloadFailure, H2DownloadOutcome};
+use super::h2_pool::SharedH2Connection;
+use crate::util::fetch::{
+    self, DownloadRequest, DownloadResult, DownloadRoute,
+};
+use futures::stream::{FuturesUnordered, StreamExt};
+use http::header::{ACCEPT_ENCODING, RANGE};
+use http::{HeaderValue, StatusCode, Uri};
+use std::path::Path;
+use std::path::PathBuf;
+use std::sync::Arc;
+use std::sync::atomic::{AtomicU64, Ordering};
+use std::time::Duration;
+
+struct H2Range {
+    start: u64,
+    end: u64,
+}
+
+struct PartCleanupGuard {
+    path: PathBuf,
+    armed: bool,
+}
+
+impl PartCleanupGuard {
+    fn new(path: &Path) -> Self {
+        Self {
+            path: path.to_path_buf(),
+            armed: true,
+        }
+    }
+
+    fn disarm(&mut self) {
+        self.armed = false;
+    }
+}
+
+impl Drop for PartCleanupGuard {
+    fn drop(&mut self) {
+        if self.armed {
+            let _ = std::fs::remove_file(&self.path);
+        }
+    }
+}
+
+pub(crate) async fn download(
+    connection: &Arc<SharedH2Connection>,
+    uri: &Uri,
+    request: &DownloadRequest,
+    route: &DownloadRoute,
+    destination: &Path,
+    part_path: &Path,
+    total_size: u64,
+    concurrency: usize,
+) -> H2DownloadOutcome {
+    if request
+        .cancellation
+        .as_ref()
+        .is_some_and(tokio_util::sync::CancellationToken::is_cancelled)
+    {
+        return H2DownloadOutcome::Canceled;
+    }
+    if total_size == 0 {
+        return H2DownloadOutcome::Fallback {
+            failure: H2DownloadFailure::Content,
+            preserve_partial: false,
+        };
+    }
+    let count = concurrency
+        .max(1)
+        .min(usize::try_from(total_size).unwrap_or(usize::MAX).max(1));
+    let mut cleanup = PartCleanupGuard::new(part_path);
+    let output =
+        match super::range_output::RangeOutput::create(part_path, total_size)
+            .await
+        {
+            Ok(output) => output,
+            Err(_) => {
+                return H2DownloadOutcome::Fallback {
+                    failure: H2DownloadFailure::Io,
+                    preserve_partial: false,
+                };
+            }
+        };
+    let ranges = split_ranges(total_size, count);
+    let downloaded = Arc::new(AtomicU64::new(0));
+    let reported_bucket = Arc::new(AtomicU64::new(0));
+    let progress_delta = (total_size / 200).max(256 * 1024);
+    let mut tasks = FuturesUnordered::new();
+    for range in ranges {
+        tasks.push(download_range(
+            Arc::clone(connection),
+            uri.clone(),
+            request,
+            route,
+            Arc::clone(&output),
+            range,
+            total_size,
+            Arc::clone(&downloaded),
+            Arc::clone(&reported_bucket),
+            progress_delta,
+        ));
+    }
+    loop {
+        let next = if let Some(cancellation) = request.cancellation.as_ref() {
+            tokio::select! {
+                biased;
+                _ = cancellation.cancelled() => {
+                    drop(tasks);
+                    drop(output);
+                    return H2DownloadOutcome::Canceled;
+                }
+                result = tasks.next() => result,
+            }
+        } else {
+            tasks.next().await
+        };
+        let Some(result) = next else {
+            break;
+        };
+        if let Err(failure) = result {
+            drop(tasks);
+            drop(output);
+            return H2DownloadOutcome::Fallback {
+                failure,
+                preserve_partial: false,
+            };
+        }
+    }
+    drop(output);
+    let verification = if let Some(cancellation) = request.cancellation.as_ref()
+    {
+        tokio::select! {
+            biased;
+            _ = cancellation.cancelled() => return H2DownloadOutcome::Canceled,
+            result = fetch::verify_file(part_path, &request.integrity) => result,
+        }
+    } else {
+        fetch::verify_file(part_path, &request.integrity).await
+    };
+    if let Err(error) = verification {
+        let failure = if fetch::is_integrity_error(&error) {
+            H2DownloadFailure::Integrity
+        } else {
+            H2DownloadFailure::Content
+        };
+        return H2DownloadOutcome::Fallback {
+            failure,
+            preserve_partial: false,
+        };
+    }
+    let finalized = if let Some(cancellation) = request.cancellation.as_ref() {
+        tokio::select! {
+            biased;
+            _ = cancellation.cancelled() => return H2DownloadOutcome::Canceled,
+            result = fetch::finalize_download(part_path, destination) => result,
+        }
+    } else {
+        fetch::finalize_download(part_path, destination).await
+    };
+    if finalized.is_err() {
+        return H2DownloadOutcome::Fallback {
+            failure: H2DownloadFailure::Io,
+            preserve_partial: false,
+        };
+    }
+    cleanup.disarm();
+    H2DownloadOutcome::Completed(DownloadResult {
+        path: destination.to_path_buf(),
+        url: uri.to_string(),
+        source: route.source,
+        size: total_size,
+        attempts: 1,
+        fallback_count: 0,
+    })
+}
+
+#[allow(clippy::too_many_arguments)]
+async fn download_range(
+    connection: Arc<SharedH2Connection>,
+    uri: Uri,
+    request: &DownloadRequest,
+    route: &DownloadRoute,
+    output: Arc<super::range_output::RangeOutput>,
+    range: H2Range,
+    total_size: u64,
+    downloaded: Arc<AtomicU64>,
+    reported_bucket: Arc<AtomicU64>,
+    progress_delta: u64,
+) -> Result<(), H2DownloadFailure> {
+    super::h2_download::record_install_stage(
+        request,
+        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)?;
+    super::h2_download::record_install_stage(
+        request,
+        crate::install::DownloadItemStatus::Downloading,
+    )
+    .await;
+    let mut writer = output
+        .open_range(range.start, range.end + 1)
+        .await
+        .map_err(|_| H2DownloadFailure::Io)?;
+    let mut headers = super::h2_download::request_headers(request, route);
+    headers.insert(ACCEPT_ENCODING, HeaderValue::from_static("identity"));
+    headers.insert(
+        RANGE,
+        HeaderValue::from_str(&format!("bytes={}-{}", range.start, range.end))
+            .map_err(|_| H2DownloadFailure::Http)?,
+    );
+    let (response, mut stream) =
+        super::h2_download::open_stream(&connection, &uri, headers)
+            .await
+            .map_err(|_| H2DownloadFailure::Protocol)?;
+    if response.status() != StatusCode::PARTIAL_CONTENT
+        || !content_range_matches(response.headers(), &range, total_size)
+    {
+        return Err(H2DownloadFailure::Http);
+    }
+    let activity = super::h2_receive::H2TransferActivity::begin();
+    let mut offset = range.start;
+    while let Some(chunk) =
+        super::h2_receive::receive_chunk(&mut stream, "range")
+            .await
+            .map_err(|_| H2DownloadFailure::Protocol)?
+    {
+        let remaining = range.end.saturating_add(1).saturating_sub(offset);
+        if chunk.len() as u64 > remaining {
+            return Err(H2DownloadFailure::Protocol);
+        }
+        let accepted = chunk.len();
+        writer
+            .write_next(&chunk[..accepted])
+            .await
+            .map_err(|_| H2DownloadFailure::Io)?;
+        activity.record_bytes(accepted);
+        super::h2_receive::release_capacity(&mut stream, chunk.len())
+            .map_err(|_| H2DownloadFailure::Protocol)?;
+        offset += accepted as u64;
+        let current = downloaded.fetch_add(accepted as u64, Ordering::Relaxed)
+            + accepted as u64;
+        let bucket = current / progress_delta;
+        let previous = reported_bucket.fetch_max(bucket, Ordering::Relaxed);
+        if current >= total_size || bucket > previous {
+            super::h2_download::record_install_progress(
+                request,
+                current.min(total_size),
+                total_size,
+            )
+            .await;
+        }
+    }
+    (offset == range.end + 1)
+        .then_some(())
+        .ok_or(H2DownloadFailure::Protocol)?;
+    writer.flush().await.map_err(|_| H2DownloadFailure::Io)
+}
+
+fn split_ranges(size: u64, count: usize) -> Vec<H2Range> {
+    let base = size / count as u64;
+    let remainder = size % count as u64;
+    let mut start = 0;
+    (0..count)
+        .map(|index| {
+            let length = base + u64::from(index < remainder as usize);
+            let range = H2Range {
+                start,
+                end: start + length - 1,
+            };
+            start += length;
+            range
+        })
+        .collect()
+}
+
+fn content_range_matches(
+    headers: &http::HeaderMap,
+    range: &H2Range,
+    total_size: u64,
+) -> bool {
+    headers
+        .get(http::header::CONTENT_RANGE)
+        .and_then(|value| value.to_str().ok())
+        .is_some_and(|value| {
+            value
+                == format!("bytes {}-{}/{}", range.start, range.end, total_size)
+        })
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use bytes::Bytes;
+    use futures::future::poll_fn;
+    use http::{Response, StatusCode};
+    use sha1_smol::Sha1;
+
+    #[test]
+    fn splits_file_into_eight_contiguous_ranges() {
+        let ranges = split_ranges(83, 8);
+        assert_eq!(ranges.len(), 8);
+        assert_eq!(ranges.first().unwrap().start, 0);
+        assert_eq!(ranges.last().unwrap().end, 82);
+        assert!(
+            ranges
+                .windows(2)
+                .all(|pair| pair[0].end + 1 == pair[1].start)
+        );
+    }
+
+    #[test]
+    fn splits_file_into_sixteen_contiguous_ranges() {
+        let ranges = split_ranges(1024 * 1024 + 7, 16);
+        assert_eq!(ranges.len(), 16);
+        assert_eq!(ranges.first().unwrap().start, 0);
+        assert_eq!(ranges.last().unwrap().end, 1024 * 1024 + 6);
+        assert!(
+            ranges
+                .windows(2)
+                .all(|pair| pair[0].end + 1 == pair[1].start)
+        );
+    }
+
+    #[tokio::test]
+    async fn downloads_one_file_over_eight_h2_range_streams() {
+        let data = Arc::new(
+            (0..2 * 1024 * 1024)
+                .map(|index| (index % 251) as u8)
+                .collect::<Vec<_>>(),
+        );
+        let request_count = Arc::new(AtomicU64::new(0));
+        let (client_io, server_io) = tokio::io::duplex(256 * 1024);
+        let server_data = Arc::clone(&data);
+        let server_requests = Arc::clone(&request_count);
+        let server = tokio::spawn(async move {
+            let mut connection =
+                h2::server::handshake(server_io).await.unwrap();
+            while let Some(result) = connection.accept().await {
+                let (request, mut respond) = result.unwrap();
+                let data = Arc::clone(&server_data);
+                let requests = Arc::clone(&server_requests);
+                tokio::spawn(async move {
+                    requests.fetch_add(1, Ordering::Relaxed);
+                    let value = request.headers()[RANGE].to_str().unwrap();
+                    let value = value.strip_prefix("bytes=").unwrap();
+                    let (start, end) = value.split_once('-').unwrap();
+                    let start = start.parse::<usize>().unwrap();
+                    let end = end.parse::<usize>().unwrap();
+                    let response = Response::builder()
+                        .status(StatusCode::PARTIAL_CONTENT)
+                        .header(
+                            http::header::CONTENT_RANGE,
+                            format!("bytes {start}-{end}/{}", data.len()),
+                        )
+                        .body(())
+                        .unwrap();
+                    let mut stream =
+                        respond.send_response(response, false).unwrap();
+                    let mut offset = start;
+                    while offset <= end {
+                        let wanted = (end - offset + 1).min(16 * 1024);
+                        stream.reserve_capacity(wanted);
+                        let capacity = poll_fn(|cx| stream.poll_capacity(cx))
+                            .await
+                            .unwrap()
+                            .unwrap();
+                        let length = capacity.min(wanted);
+                        let finished = offset + length > end;
+                        stream
+                            .send_data(
+                                Bytes::copy_from_slice(
+                                    &data[offset..offset + length],
+                                ),
+                                finished,
+                            )
+                            .unwrap();
+                        offset += length;
+                    }
+                });
+            }
+        });
+
+        let mut builder = h2::client::Builder::new();
+        builder
+            .initial_window_size(1024 * 1024)
+            .initial_connection_window_size(64 * 1024 * 1024);
+        let (sender, mut driver) =
+            builder.handshake::<_, Bytes>(client_io).await.unwrap();
+        driver.set_target_window_size(64 * 1024 * 1024);
+        let client_driver = tokio::spawn(async move { driver.await });
+        let connection = Arc::new(SharedH2Connection::for_test(sender));
+        let directory = tempfile::tempdir().unwrap();
+        let destination = directory.path().join("pack.mrpack");
+        let part_path = directory.path().join("pack.mrpack.part");
+        let hash = Sha1::from(&data[..]).hexdigest();
+        let request = DownloadRequest::new(
+            "https://h2-range.test/pack.mrpack",
+            fetch::ResourceClass::Modpack,
+        )
+        .with_integrity(
+            fetch::Integrity::sha1(hash).with_size(data.len() as u64),
+        );
+        let route = DownloadRoute {
+            url: request.url.clone(),
+            source: fetch::DownloadRouteSource::Official,
+            is_mirror: false,
+            allow_sensitive_headers: true,
+            supports_range: true,
+            proxy: fetch::ProxyPolicy::Direct,
+        };
+        let uri = route.url.parse().unwrap();
+
+        let result = download(
+            &connection,
+            &uri,
+            &request,
+            &route,
+            &destination,
+            &part_path,
+            data.len() as u64,
+            8,
+        )
+        .await;
+
+        assert!(matches!(result, H2DownloadOutcome::Completed(_)));
+        assert_eq!(request_count.load(Ordering::Relaxed), 8);
+        assert_eq!(tokio::fs::read(destination).await.unwrap(), *data);
+        client_driver.abort();
+        server.abort();
+    }
+
+    #[tokio::test]
+    async fn cancellation_stops_all_ranges_and_removes_partial_output() {
+        let request_count = Arc::new(AtomicU64::new(0));
+        let (client_io, server_io) = tokio::io::duplex(256 * 1024);
+        let server_requests = Arc::clone(&request_count);
+        let server = tokio::spawn(async move {
+            let mut connection =
+                h2::server::handshake(server_io).await.unwrap();
+            while let Some(result) = connection.accept().await {
+                let (request, mut respond) = result.unwrap();
+                let requests = Arc::clone(&server_requests);
+                tokio::spawn(async move {
+                    requests.fetch_add(1, Ordering::Relaxed);
+                    let value = request.headers()[RANGE].to_str().unwrap();
+                    let value = value.strip_prefix("bytes=").unwrap();
+                    let (start, end) = value.split_once('-').unwrap();
+                    let response = Response::builder()
+                        .status(StatusCode::PARTIAL_CONTENT)
+                        .header(
+                            http::header::CONTENT_RANGE,
+                            format!("bytes {start}-{end}/2097152"),
+                        )
+                        .body(())
+                        .unwrap();
+                    let _stream =
+                        respond.send_response(response, false).unwrap();
+                    futures::future::pending::<()>().await;
+                });
+            }
+        });
+
+        let (sender, driver) = h2::client::handshake(client_io).await.unwrap();
+        let client_driver = tokio::spawn(async move { driver.await });
+        let connection = Arc::new(SharedH2Connection::for_test(sender));
+        let directory = tempfile::tempdir().unwrap();
+        let destination = directory.path().join("pack.zip");
+        let part_path = directory.path().join("pack.zip.part");
+        let cancellation = tokio_util::sync::CancellationToken::new();
+        let mut request = DownloadRequest::new(
+            "https://h2-range.test/pack.zip",
+            fetch::ResourceClass::CurseForge,
+        )
+        .with_integrity(fetch::Integrity::default().with_size(2 * 1024 * 1024));
+        request.cancellation = Some(cancellation.clone());
+        let route = DownloadRoute {
+            url: request.url.clone(),
+            source: fetch::DownloadRouteSource::Official,
+            is_mirror: false,
+            allow_sensitive_headers: true,
+            supports_range: true,
+            proxy: fetch::ProxyPolicy::Direct,
+        };
+        let uri = route.url.parse().unwrap();
+        let part_for_task = part_path.clone();
+        let destination_for_task = destination.clone();
+        let download_task = tokio::spawn(async move {
+            download(
+                &connection,
+                &uri,
+                &request,
+                &route,
+                &destination_for_task,
+                &part_for_task,
+                2 * 1024 * 1024,
+                16,
+            )
+            .await
+        });
+        tokio::time::timeout(std::time::Duration::from_secs(2), async {
+            while request_count.load(Ordering::Relaxed) < 16 {
+                tokio::task::yield_now().await;
+            }
+        })
+        .await
+        .unwrap();
+
+        cancellation.cancel();
+        let outcome = tokio::time::timeout(
+            std::time::Duration::from_secs(2),
+            download_task,
+        )
+        .await
+        .unwrap()
+        .unwrap();
+
+        assert!(matches!(outcome, H2DownloadOutcome::Canceled));
+        assert!(!part_path.exists());
+        assert!(!destination.exists());
+        client_driver.abort();
+        server.abort();
+    }
+}

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

@@ -0,0 +1,173 @@
+//! Shared receive-side accounting for native HTTP/2 streams.
+
+use bytes::Bytes;
+use std::time::Duration;
+
+const STREAM_RECV_TIMEOUT: Duration = Duration::from_secs(30);
+const MIN_PROGRESS_BYTES: u64 = 256 * 1024;
+
+/// Tracks one active H2 stream in the native automatic-concurrency metrics.
+pub(crate) struct H2TransferActivity {
+    _activity: Option<crate::state::DownloadConnectionActivity>,
+}
+
+impl H2TransferActivity {
+    pub(crate) fn begin() -> Self {
+        Self {
+            _activity: crate::State::get_if_initialized()
+                .map(|state| state.begin_download_connection()),
+        }
+    }
+
+    pub(crate) fn record_bytes(&self, bytes: usize) {
+        if let Some(state) = crate::State::get_if_initialized() {
+            state.record_download_bytes(bytes as u64);
+        }
+    }
+}
+
+/// Limits install-progress work before it reaches the shared reporter lock.
+pub(crate) struct H2ProgressGate {
+    last_reported: u64,
+    threshold: u64,
+}
+
+impl H2ProgressGate {
+    pub(crate) fn new(total_size: u64) -> Self {
+        Self {
+            last_reported: 0,
+            threshold: MIN_PROGRESS_BYTES.max(total_size / 200),
+        }
+    }
+
+    pub(crate) fn should_report(
+        &mut self,
+        downloaded: u64,
+        total_size: u64,
+    ) -> bool {
+        if downloaded < total_size
+            && downloaded.saturating_sub(self.last_reported) < self.threshold
+        {
+            return false;
+        }
+        if downloaded == self.last_reported {
+            return false;
+        }
+        self.last_reported = downloaded;
+        true
+    }
+}
+
+pub(crate) async fn receive_chunk(
+    stream: &mut h2::RecvStream,
+    context: &str,
+) -> crate::Result<Option<Bytes>> {
+    tokio::time::timeout(STREAM_RECV_TIMEOUT, stream.data())
+        .await
+        .map_err(|_| {
+            crate::ErrorKind::NetworkError(format!(
+                "HTTP/2 {context} stream receive timed out"
+            ))
+        })?
+        .transpose()
+        .map_err(|error| {
+            crate::ErrorKind::NetworkError(format!(
+                "HTTP/2 {context} stream error: {error}"
+            ))
+            .into()
+        })
+}
+
+/// Returns receive-window capacity after the caller has consumed the frame.
+pub(crate) fn release_capacity(
+    stream: &mut h2::RecvStream,
+    bytes: usize,
+) -> crate::Result<()> {
+    stream
+        .flow_control()
+        .release_capacity(bytes)
+        .map_err(|error| {
+            crate::ErrorKind::NetworkError(format!(
+                "HTTP/2 flow-control release failed: {error}"
+            ))
+            .into()
+        })
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use futures::future::poll_fn;
+    use http::{Request, Response};
+
+    #[test]
+    fn progress_gate_throttles_intermediate_updates_and_reports_completion() {
+        let total = 2 * 1024 * 1024;
+        let mut gate = H2ProgressGate::new(total);
+
+        assert!(!gate.should_report(64 * 1024, total));
+        assert!(gate.should_report(256 * 1024, total));
+        assert!(!gate.should_report(320 * 1024, total));
+        assert!(gate.should_report(total, total));
+        assert!(!gate.should_report(total, total));
+    }
+
+    #[tokio::test]
+    async fn receive_capacity_allows_body_larger_than_initial_window() {
+        const BODY_SIZE: usize = 2 * 1024 * 1024;
+        const FRAME_SIZE: usize = 16 * 1024;
+
+        let (client_io, server_io) = tokio::io::duplex(64 * 1024);
+        let server = tokio::spawn(async move {
+            let mut connection =
+                h2::server::handshake(server_io).await.unwrap();
+            let (_, mut respond) = connection.accept().await.unwrap().unwrap();
+            // Keep polling the connection while the response body waits for
+            // client WINDOW_UPDATE frames.
+            tokio::spawn(async move {
+                while connection.accept().await.is_some() {}
+            });
+            let mut stream =
+                respond.send_response(Response::new(()), false).unwrap();
+            let mut remaining = BODY_SIZE;
+            while remaining > 0 {
+                stream.reserve_capacity(remaining.min(FRAME_SIZE));
+                let capacity = poll_fn(|cx| stream.poll_capacity(cx))
+                    .await
+                    .unwrap()
+                    .unwrap();
+                let length = capacity.min(remaining).min(FRAME_SIZE);
+                remaining -= length;
+                stream
+                    .send_data(Bytes::from(vec![0x5a; length]), remaining == 0)
+                    .unwrap();
+            }
+        });
+
+        let mut builder = h2::client::Builder::new();
+        builder
+            .initial_window_size(1024 * 1024)
+            .initial_connection_window_size(64 * 1024 * 1024);
+        let (mut sender, mut connection) =
+            builder.handshake::<_, Bytes>(client_io).await.unwrap();
+        connection.set_target_window_size(64 * 1024 * 1024);
+        let client_driver = tokio::spawn(async move { connection.await });
+
+        let request =
+            Request::get("https://h2-flow.test/file").body(()).unwrap();
+        sender = sender.ready().await.unwrap();
+        let (response, _) = sender.send_request(request, true).unwrap();
+        let mut body = response.await.unwrap().into_body();
+        let mut received = 0;
+        while let Some(chunk) = receive_chunk(&mut body, "test").await.unwrap()
+        {
+            assert!(chunk.iter().all(|byte| *byte == 0x5a));
+            received += chunk.len();
+            release_capacity(&mut body, chunk.len()).unwrap();
+        }
+
+        assert_eq!(received, BODY_SIZE);
+        server.await.unwrap();
+        client_driver.abort();
+    }
+}

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

@@ -0,0 +1,193 @@
+//! Logical stream budgets for shared native HTTP/2 connections.
+//!
+//! HTTP/2 streams share one physical connection per authority, so they must
+//! not consume the HTTP/1 connection permits held by `native_budget`.
+
+use crate::util::fetch::{DownloadRoute, ProxyPolicy};
+use parking_lot::Mutex;
+use std::collections::HashMap;
+use std::sync::{Arc, LazyLock};
+use tokio::sync::{AcquireError, OwnedSemaphorePermit, Semaphore};
+use tokio::time::{Duration, sleep};
+
+const MAX_H2_STREAMS: usize = 128;
+const MAX_H2_STREAMS_PER_AUTHORITY: usize = 32;
+const MAX_ASSET_H2_STREAMS: usize = 96;
+const MAX_ASSET_H2_STREAMS_PER_AUTHORITY: usize = 24;
+
+#[derive(Clone, Debug, Eq, Hash, PartialEq)]
+struct AuthorityKey {
+    authority: String,
+    proxy: ProxyPolicy,
+}
+
+static AUTHORITY_BUDGETS: LazyLock<
+    Mutex<HashMap<AuthorityKey, Arc<Semaphore>>>,
+> = LazyLock::new(|| Mutex::new(HashMap::new()));
+static GLOBAL_BUDGET: LazyLock<Arc<Semaphore>> =
+    LazyLock::new(|| Arc::new(Semaphore::new(MAX_H2_STREAMS)));
+static ASSET_AUTHORITY_BUDGETS: LazyLock<
+    Mutex<HashMap<AuthorityKey, Arc<Semaphore>>>,
+> = LazyLock::new(|| Mutex::new(HashMap::new()));
+static ASSET_GLOBAL_BUDGET: LazyLock<Arc<Semaphore>> =
+    LazyLock::new(|| Arc::new(Semaphore::new(MAX_ASSET_H2_STREAMS)));
+
+pub(crate) struct H2StreamPermit {
+    _global: OwnedSemaphorePermit,
+    _authority: Option<OwnedSemaphorePermit>,
+}
+
+fn budget(route: &DownloadRoute) -> Option<Arc<Semaphore>> {
+    let authority = crate::util::fetch::url_authority(&route.url)?;
+    let key = AuthorityKey {
+        authority,
+        proxy: route.proxy,
+    };
+    let mut budgets = AUTHORITY_BUDGETS.lock();
+    if budgets.len() >= 256 {
+        budgets.retain(|_, budget| Arc::strong_count(budget) > 1);
+    }
+    Some(
+        budgets
+            .entry(key)
+            .or_insert_with(|| {
+                Arc::new(Semaphore::new(MAX_H2_STREAMS_PER_AUTHORITY))
+            })
+            .clone(),
+    )
+}
+
+fn asset_budget(route: &DownloadRoute) -> Option<Arc<Semaphore>> {
+    let authority = crate::util::fetch::url_authority(&route.url)?;
+    let key = AuthorityKey {
+        authority,
+        proxy: route.proxy,
+    };
+    let mut budgets = ASSET_AUTHORITY_BUDGETS.lock();
+    if budgets.len() >= 256 {
+        budgets.retain(|_, budget| Arc::strong_count(budget) > 1);
+    }
+    Some(
+        budgets
+            .entry(key)
+            .or_insert_with(|| {
+                Arc::new(Semaphore::new(MAX_ASSET_H2_STREAMS_PER_AUTHORITY))
+            })
+            .clone(),
+    )
+}
+
+pub(crate) async fn acquire(
+    route: &DownloadRoute,
+) -> Result<H2StreamPermit, AcquireError> {
+    let authority_budget = budget(route);
+    loop {
+        let global = Arc::clone(&GLOBAL_BUDGET).acquire_owned().await?;
+        let authority = match authority_budget.as_ref() {
+            Some(budget) => match Arc::clone(budget).try_acquire_owned() {
+                Ok(permit) => Some(permit),
+                Err(tokio::sync::TryAcquireError::NoPermits) => {
+                    drop(global);
+                    sleep(Duration::from_millis(5)).await;
+                    continue;
+                }
+                Err(tokio::sync::TryAcquireError::Closed) => {
+                    unreachable!("H2 authority budgets are never closed")
+                }
+            },
+            None => None,
+        };
+        return Ok(H2StreamPermit {
+            _global: global,
+            _authority: authority,
+        });
+    }
+}
+
+/// 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.
+pub(crate) async fn acquire_asset(
+    route: &DownloadRoute,
+) -> Result<H2StreamPermit, AcquireError> {
+    let authority_budget = asset_budget(route);
+    loop {
+        let global = Arc::clone(&ASSET_GLOBAL_BUDGET).acquire_owned().await?;
+        let authority = match authority_budget.as_ref() {
+            Some(budget) => match Arc::clone(budget).try_acquire_owned() {
+                Ok(permit) => Some(permit),
+                Err(tokio::sync::TryAcquireError::NoPermits) => {
+                    drop(global);
+                    sleep(Duration::from_millis(5)).await;
+                    continue;
+                }
+                Err(tokio::sync::TryAcquireError::Closed) => {
+                    unreachable!("asset authority budgets are never closed")
+                }
+            },
+            None => None,
+        };
+        return Ok(H2StreamPermit {
+            _global: global,
+            _authority: authority,
+        });
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::util::fetch::DownloadRouteSource;
+
+    fn route() -> DownloadRoute {
+        DownloadRoute {
+            url: "https://h2-budget.example/file".to_string(),
+            source: DownloadRouteSource::Official,
+            is_mirror: false,
+            allow_sensitive_headers: true,
+            supports_range: true,
+            proxy: ProxyPolicy::Direct,
+        }
+    }
+
+    #[tokio::test]
+    async fn authority_stream_budget_is_independent_from_connection_budget() {
+        let route = route();
+        let mut permits = Vec::new();
+        for _ in 0..MAX_H2_STREAMS_PER_AUTHORITY {
+            permits.push(acquire(&route).await.unwrap());
+        }
+        assert_eq!(budget(&route).unwrap().available_permits(), 0,);
+        drop(permits);
+        assert_eq!(
+            budget(&route).unwrap().available_permits(),
+            MAX_H2_STREAMS_PER_AUTHORITY,
+        );
+    }
+
+    #[tokio::test]
+    async fn asset_stream_budget_is_separate_and_bounded() {
+        let route = DownloadRoute {
+            url: "https://asset-h2-budget.example/file".to_string(),
+            source: DownloadRouteSource::Bmclapi,
+            is_mirror: true,
+            allow_sensitive_headers: false,
+            supports_range: true,
+            proxy: ProxyPolicy::Direct,
+        };
+        let mut permits = Vec::new();
+        for _ in 0..MAX_ASSET_H2_STREAMS_PER_AUTHORITY {
+            permits.push(acquire_asset(&route).await.unwrap());
+        }
+        assert_eq!(asset_budget(&route).unwrap().available_permits(), 0,);
+        drop(permits);
+        assert_eq!(
+            asset_budget(&route).unwrap().available_permits(),
+            MAX_ASSET_H2_STREAMS_PER_AUTHORITY,
+        );
+        assert_eq!(
+            budget(&route).unwrap().available_permits(),
+            MAX_H2_STREAMS_PER_AUTHORITY
+        );
+    }
+}

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

@@ -0,0 +1,5 @@
+//! Legacy adaptive download engine.
+//!
+//! The native implementation currently lives in `fetch.rs`; this module is
+//! retained because the download module still exposes the legacy engine as a
+//! selectable implementation boundary.

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

@@ -0,0 +1,73 @@
+//! Compact local download-event log.
+//!
+//! Only stall/error events are written. Each line is intentionally
+//! short and numeric so the file stays tiny; `scripts/axolotl/decode-download-log.mjs`
+//! can turn it back into readable text.
+
+use super::slow::{SlowEvent, SlowRule};
+use std::io::Write;
+use std::path::Path;
+use std::time::{SystemTime, UNIX_EPOCH};
+
+const MAX_LOG_BYTES: u64 = 5 * 1024 * 1024;
+
+pub fn engine_code(engine: &str) -> u8 {
+    match engine {
+        "xmcl" => 1,
+        _ => 0,
+    }
+}
+
+pub fn source_code(source: &str) -> u8 {
+    match source {
+        "official" => 0,
+        "bmclapi" => 1,
+        "mcim" => 2,
+        "alternate" => 3,
+        "tianpao" => 5,
+        _ => 4,
+    }
+}
+
+pub fn encode_stall(event: &SlowEvent) -> String {
+    let rule = match event.rule {
+        SlowRule::R1NoProgress => 1,
+        SlowRule::R2BelowExpectation => 2,
+        SlowRule::R3SegmentWaste => 3,
+        SlowRule::R4FrequentSwitches => 4,
+    };
+    let timestamp = SystemTime::now()
+        .duration_since(UNIX_EPOCH)
+        .map(|duration| duration.as_secs())
+        .unwrap_or(0);
+    format!(
+        "{timestamp}|{}|{rule}|{}|{}",
+        engine_code(&event.engine),
+        source_code(&event.source),
+        event.detail
+    )
+}
+
+pub fn append_stall(log_path: &Path, event: &SlowEvent) -> std::io::Result<()> {
+    rotate_if_needed(log_path)?;
+    let mut file = std::fs::OpenOptions::new()
+        .create(true)
+        .append(true)
+        .open(log_path)?;
+    writeln!(file, "{}", encode_stall(event))
+}
+
+fn rotate_if_needed(log_path: &Path) -> std::io::Result<()> {
+    let Ok(metadata) = std::fs::metadata(log_path) else {
+        return Ok(());
+    };
+    if metadata.len() < MAX_LOG_BYTES {
+        return Ok(());
+    }
+    let rotated = log_path.with_extension("log.1");
+    if rotated.exists() {
+        std::fs::remove_file(&rotated)?;
+    }
+    std::fs::rename(log_path, rotated)?;
+    Ok(())
+}

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

@@ -0,0 +1,78 @@
+//! Modular download engine selection.
+//!
+//! The launcher can use either the native adaptive engine (HTTP/2
+//! multiplexing with shared per-authority connections, falling back to
+//! HTTP/1.1 single-stream) or the XMCL-compatible engine. The native engine
+//! is the default; the XMCL engine remains available for users who prefer it.
+
+use serde::{Deserialize, Serialize};
+use std::sync::atomic::{AtomicU8, Ordering};
+
+pub mod h2_download;
+pub mod h2_pool;
+pub(crate) mod h2_range;
+pub(crate) mod h2_receive;
+pub(crate) mod h2_stream_budget;
+pub mod legacy;
+pub mod log;
+pub(crate) mod modrinth_redirect;
+pub(crate) mod native;
+pub(crate) mod native_breaker;
+pub(crate) mod native_budget;
+pub(crate) mod native_reputation;
+pub(crate) mod native_slow;
+pub(crate) mod range_output;
+pub mod shared;
+pub mod slow;
+pub mod xmcl;
+
+#[derive(
+    Debug, Clone, Copy, Default, Eq, PartialEq, Serialize, Deserialize,
+)]
+#[repr(u8)]
+#[serde(rename_all = "snake_case")]
+pub enum DownloadEngine {
+    #[default]
+    Legacy,
+    #[serde(rename = "xmcl", alias = "xmcl_compat")]
+    XmclCompat,
+}
+
+impl DownloadEngine {
+    pub const fn as_str(self) -> &'static str {
+        match self {
+            Self::Legacy => "legacy",
+            Self::XmclCompat => "xmcl",
+        }
+    }
+
+    pub fn from_str(value: &str) -> Self {
+        match value {
+            "xmcl" | "xmcl_compat" => Self::XmclCompat,
+            _ => Self::Legacy,
+        }
+    }
+}
+
+static ACTIVE_ENGINE: AtomicU8 = AtomicU8::new(0);
+
+/// Returns the engine the launcher should use for new downloads.
+pub fn active_engine() -> DownloadEngine {
+    match ACTIVE_ENGINE.load(Ordering::Relaxed) {
+        1 => DownloadEngine::XmclCompat,
+        _ => DownloadEngine::Legacy,
+    }
+}
+
+/// Sets the engine used by new downloads.
+pub fn set_active_engine(engine: DownloadEngine) {
+    ACTIVE_ENGINE.store(engine as u8, Ordering::Relaxed);
+}
+
+pub(crate) fn task_concurrency_limit(state: &crate::State) -> Option<usize> {
+    if active_engine() == DownloadEngine::XmclCompat {
+        None
+    } else {
+        Some(state.download_concurrency())
+    }
+}

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

@@ -0,0 +1,71 @@
+//! Modrinth CDN redirect normalization for native download requests.
+
+use url::Url;
+
+use crate::util::fetch::{
+    MODRINTH_CDN_LEGACY_HOST, MODRINTH_CDN_OFFICIAL_HOST, TIANPAO_HOST,
+};
+
+/// Identifies a Tianpao response that points at one of Modrinth's real CDN
+/// hosts. The caller uses this only to abandon the mirror request; it must
+/// follow the server-provided host without rewriting it.
+pub(crate) fn is_tianpao_official_redirect(
+    current: &Url,
+    location: Option<&str>,
+) -> bool {
+    current
+        .host_str()
+        .is_some_and(|host| host.eq_ignore_ascii_case(TIANPAO_HOST))
+        && location.is_some_and(|location| {
+            let Ok(redirect) = current.join(location) else {
+                return false;
+            };
+            redirect.host_str().is_some_and(|host| {
+                host.eq_ignore_ascii_case(MODRINTH_CDN_LEGACY_HOST)
+                    || host.eq_ignore_ascii_case(MODRINTH_CDN_OFFICIAL_HOST)
+            })
+        })
+}
+
+pub(crate) fn repair_official_redirect(
+    original: &Url,
+    redirect: &Url,
+    location: &str,
+) -> Option<Url> {
+    if location.is_ascii()
+        || !redirect.host_str().is_some_and(|host| {
+            host.eq_ignore_ascii_case(MODRINTH_CDN_LEGACY_HOST)
+                || host.eq_ignore_ascii_case(MODRINTH_CDN_OFFICIAL_HOST)
+        })
+        || original.path().is_empty()
+    {
+        return None;
+    }
+    let mut repaired = redirect.clone();
+    repaired.set_path(original.path());
+    repaired.set_query(original.query());
+    repaired.set_fragment(original.fragment());
+    Some(repaired)
+}
+
+pub(crate) fn is_official_redirect(location: Option<&str>) -> bool {
+    let Some(location) = location.filter(|location| {
+        location.len() <= 8 * 1024
+            && location.is_ascii()
+            && location
+                .get(..8)
+                .is_some_and(|scheme| scheme.eq_ignore_ascii_case("https://"))
+    }) else {
+        return false;
+    };
+    let authority = location[8..]
+        .split(['/', '?', '#'])
+        .next()
+        .unwrap_or_default();
+    authority.eq_ignore_ascii_case(MODRINTH_CDN_OFFICIAL_HOST)
+        || authority
+            .eq_ignore_ascii_case(&format!("{MODRINTH_CDN_OFFICIAL_HOST}:443"))
+        || authority.eq_ignore_ascii_case(MODRINTH_CDN_LEGACY_HOST)
+        || authority
+            .eq_ignore_ascii_case(&format!("{MODRINTH_CDN_LEGACY_HOST}:443"))
+}

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

@@ -0,0 +1,225 @@
+//! Policy boundary for the native download engine.
+
+use crate::util::fetch::{DownloadRoute, ProxyPolicy};
+
+const SMALL_FILE_LIMIT: u64 = 4 * 1024 * 1024;
+const MEDIUM_FILE_LIMIT: u64 = 16 * 1024 * 1024;
+const LARGE_FILE_LIMIT: u64 = 64 * 1024 * 1024;
+
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+pub(crate) struct NativeH2Policy {
+    /// Whether this selection may establish the authority's first H2
+    /// connection instead of requiring an already-warm multiplexed session.
+    pub(crate) allow_cold_connection: bool,
+    pub(crate) abort_if_slow: bool,
+    pub(crate) expected_speed: Option<u64>,
+}
+
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+pub(crate) enum NativeH2IneligibleReason {
+    Http1Fallback,
+    SystemProxy,
+}
+
+impl NativeH2IneligibleReason {
+    pub(crate) const fn as_str(self) -> &'static str {
+        match self {
+            Self::Http1Fallback => "authority is temporarily using HTTP/1.1",
+            Self::SystemProxy => "system proxy requires the reqwest transport",
+        }
+    }
+}
+
+pub(crate) fn h2_ineligible_reason(
+    route: &DownloadRoute,
+) -> Option<NativeH2IneligibleReason> {
+    let authority = crate::util::fetch::url_authority(&route.url)?;
+    if crate::util::fetch::authority_uses_http1_fallback(&authority) {
+        return Some(NativeH2IneligibleReason::Http1Fallback);
+    }
+    if route.proxy == ProxyPolicy::System && system_proxy_configured() {
+        return Some(NativeH2IneligibleReason::SystemProxy);
+    }
+    None
+}
+
+pub(crate) fn explicit_h2_policy(
+    route: &DownloadRoute,
+) -> Option<NativeH2Policy> {
+    h2_ineligible_reason(route)
+        .is_none()
+        .then_some(NativeH2Policy {
+            allow_cold_connection: true,
+            abort_if_slow: true,
+            expected_speed: None,
+        })
+}
+
+pub(crate) async fn h2_policy(
+    route: &DownloadRoute,
+    size: Option<u64>,
+) -> Option<NativeH2Policy> {
+    if h2_ineligible_reason(route).is_some() {
+        return None;
+    }
+    let Some(size) = size else {
+        return Some(NativeH2Policy {
+            allow_cold_connection: true,
+            abort_if_slow: false,
+            expected_speed: None,
+        });
+    };
+    let authority = crate::util::fetch::url_authority(&route.url)?;
+    let h2 = super::native_reputation::get_transport(
+        &authority,
+        route.proxy,
+        super::native_reputation::NativeTransport::H2Single,
+    );
+    let multi_range = super::native_reputation::get_transport(
+        &authority,
+        route.proxy,
+        super::native_reputation::NativeTransport::Http1MultiRange,
+    );
+    let expected_speed = h2
+        .and_then(|health| health.throughput_bps)
+        .map(|speed| speed.min(u64::MAX as f64) as u64);
+    if size < SMALL_FILE_LIMIT {
+        return Some(NativeH2Policy {
+            allow_cold_connection: true,
+            abort_if_slow: false,
+            expected_speed,
+        });
+    }
+    if size < MEDIUM_FILE_LIMIT {
+        // The connection slot serializes the first handshake, so a batch of
+        // medium files shares one cold-start attempt instead of stampeding the
+        // CDN with independent TCP/TLS connections.
+        return Some(NativeH2Policy {
+            allow_cold_connection: true,
+            abort_if_slow: true,
+            expected_speed,
+        });
+    }
+    if !super::h2_pool::has_live_connection(&authority).await {
+        return None;
+    }
+    let h2 = h2.filter(|health| health.success_samples >= 2)?;
+    let h2_speed = h2.throughput_bps?;
+    let range_speed = multi_range.and_then(|health| health.throughput_bps);
+    if size < LARGE_FILE_LIMIT {
+        let h2_preferred = range_speed
+            .map(|range_speed| h2_speed >= range_speed * 1.1)
+            .unwrap_or(true);
+        return h2_preferred.then_some(NativeH2Policy {
+            allow_cold_connection: false,
+            abort_if_slow: true,
+            expected_speed: Some(h2_speed.min(u64::MAX as f64) as u64),
+        });
+    }
+    let range = multi_range.filter(|health| health.success_samples >= 2)?;
+    let range_speed = range.throughput_bps?;
+    (h2.success_samples >= 3 && h2_speed >= range_speed * 1.25).then_some(
+        NativeH2Policy {
+            allow_cold_connection: false,
+            abort_if_slow: true,
+            expected_speed: Some(h2_speed.min(u64::MAX as f64) as u64),
+        },
+    )
+}
+
+fn system_proxy_configured() -> bool {
+    let environment_proxy = [
+        "HTTPS_PROXY",
+        "https_proxy",
+        "HTTP_PROXY",
+        "http_proxy",
+        "ALL_PROXY",
+        "all_proxy",
+    ]
+    .into_iter()
+    .any(|name| std::env::var_os(name).is_some_and(|value| !value.is_empty()));
+    environment_proxy || platform_proxy_configured()
+}
+
+#[cfg(windows)]
+fn platform_proxy_configured() -> bool {
+    use winreg::RegKey;
+    use winreg::enums::HKEY_CURRENT_USER;
+
+    let Ok(settings) = RegKey::predef(HKEY_CURRENT_USER).open_subkey(
+        "Software\\Microsoft\\Windows\\CurrentVersion\\Internet Settings",
+    ) else {
+        return false;
+    };
+    let enabled = settings
+        .get_value::<u32, _>("ProxyEnable")
+        .is_ok_and(|value| value != 0);
+    let auto_configured = settings
+        .get_value::<String, _>("AutoConfigURL")
+        .is_ok_and(|value| !value.trim().is_empty());
+    enabled || auto_configured
+}
+
+#[cfg(not(windows))]
+fn platform_proxy_configured() -> bool {
+    false
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::util::fetch::{DownloadRouteSource, ProxyPolicy};
+
+    fn route(proxy: ProxyPolicy) -> DownloadRoute {
+        DownloadRoute {
+            url: "https://native-policy.example/file".to_string(),
+            source: DownloadRouteSource::Official,
+            is_mirror: false,
+            allow_sensitive_headers: true,
+            supports_range: true,
+            proxy,
+        }
+    }
+
+    #[test]
+    fn direct_routes_ignore_system_proxy_environment() {
+        assert_ne!(
+            h2_ineligible_reason(&route(ProxyPolicy::Direct)),
+            Some(NativeH2IneligibleReason::SystemProxy)
+        );
+    }
+
+    #[tokio::test]
+    async fn medium_file_can_establish_a_cold_h2_connection() {
+        assert_eq!(
+            h2_policy(&route(ProxyPolicy::Direct), Some(8 * 1024 * 1024),)
+                .await,
+            Some(NativeH2Policy {
+                allow_cold_connection: true,
+                abort_if_slow: true,
+                expected_speed: None,
+            })
+        );
+    }
+
+    #[tokio::test]
+    async fn small_file_can_establish_h2_without_history() {
+        assert_eq!(
+            h2_policy(&route(ProxyPolicy::Direct), Some(1024 * 1024),).await,
+            Some(NativeH2Policy {
+                allow_cold_connection: true,
+                abort_if_slow: false,
+                expected_speed: None,
+            })
+        );
+    }
+
+    #[tokio::test]
+    async fn large_file_still_requires_a_warm_h2_connection_and_history() {
+        assert!(
+            h2_policy(&route(ProxyPolicy::Direct), Some(32 * 1024 * 1024),)
+                .await
+                .is_none()
+        );
+    }
+}

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

@@ -0,0 +1,130 @@
+//! Short-lived route circuit breaker for native transfers.
+
+use crate::util::fetch::{DownloadRoute, ProxyPolicy};
+use parking_lot::Mutex;
+use std::collections::HashMap;
+use std::sync::LazyLock;
+use std::time::{Duration, Instant};
+
+const FAILURE_THRESHOLD: u32 = 3;
+const MAX_FAILURE_COOLDOWN: Duration = Duration::from_secs(1);
+const DEFAULT_COOLDOWN: Duration = MAX_FAILURE_COOLDOWN;
+
+#[derive(Clone, Debug, Eq, Hash, PartialEq)]
+struct BreakerKey {
+    authority: String,
+    proxy: ProxyPolicy,
+}
+
+#[derive(Default)]
+struct BreakerState {
+    consecutive_failures: u32,
+    open_until: Option<Instant>,
+}
+
+static BREAKERS: LazyLock<Mutex<HashMap<BreakerKey, BreakerState>>> =
+    LazyLock::new(|| Mutex::new(HashMap::new()));
+
+fn key(route: &DownloadRoute) -> Option<BreakerKey> {
+    Some(BreakerKey {
+        authority: crate::util::fetch::url_authority(&route.url)?,
+        proxy: route.proxy,
+    })
+}
+
+pub(crate) fn is_open(route: &DownloadRoute) -> bool {
+    let Some(key) = key(route) else {
+        return false;
+    };
+    let mut breakers = BREAKERS.lock();
+    let Some(state) = breakers.get_mut(&key) else {
+        return false;
+    };
+    match state.open_until {
+        Some(until) if until > Instant::now() => true,
+        Some(_) => {
+            state.open_until = None;
+            state.consecutive_failures = 0;
+            false
+        }
+        None => false,
+    }
+}
+
+pub(crate) fn should_skip(
+    route: &DownloadRoute,
+    has_healthy_alternate: bool,
+) -> bool {
+    if !has_healthy_alternate {
+        return false;
+    }
+    is_open(route)
+}
+
+pub(crate) fn record_success(route: &DownloadRoute) {
+    let Some(key) = key(route) else {
+        return;
+    };
+    if let Some(state) = BREAKERS.lock().get_mut(&key) {
+        state.consecutive_failures = 0;
+        state.open_until = None;
+    }
+}
+
+pub(crate) fn record_failure(route: &DownloadRoute) {
+    record_failure_with_cooldown(route, DEFAULT_COOLDOWN);
+}
+
+pub(crate) fn record_failure_with_cooldown(
+    route: &DownloadRoute,
+    cooldown: Duration,
+) {
+    let Some(key) = key(route) else {
+        return;
+    };
+    let mut breakers = BREAKERS.lock();
+    let state = breakers.entry(key).or_default();
+    state.consecutive_failures = state.consecutive_failures.saturating_add(1);
+    if state.consecutive_failures >= FAILURE_THRESHOLD {
+        state.open_until =
+            Some(Instant::now() + cooldown.min(MAX_FAILURE_COOLDOWN));
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::util::fetch::DownloadRouteSource;
+
+    fn route(host: &str) -> DownloadRoute {
+        DownloadRoute {
+            url: format!("https://{host}/file"),
+            source: DownloadRouteSource::Official,
+            is_mirror: false,
+            allow_sensitive_headers: true,
+            supports_range: true,
+            proxy: ProxyPolicy::Direct,
+        }
+    }
+
+    #[test]
+    fn unique_route_is_never_skipped() {
+        let route = route("unique-breaker.example");
+        for _ in 0..FAILURE_THRESHOLD {
+            record_failure(&route);
+        }
+        assert!(should_skip(&route, true));
+        assert!(!should_skip(&route, false));
+        record_success(&route);
+    }
+
+    #[test]
+    fn externally_requested_cooldown_is_limited_to_one_second() {
+        let route = route("clamped-breaker.example");
+        for _ in 0..FAILURE_THRESHOLD {
+            record_failure_with_cooldown(&route, Duration::from_secs(60));
+        }
+        assert!(is_open(&route));
+        record_success(&route);
+    }
+}

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

@@ -0,0 +1,177 @@
+//! Per-authority connection budget for the native download engine.
+
+use crate::util::fetch::{DownloadRoute, ProxyPolicy};
+use parking_lot::Mutex;
+use std::collections::HashMap;
+use std::sync::{Arc, LazyLock};
+use tokio::sync::{OwnedSemaphorePermit, Semaphore, TryAcquireError};
+use tokio::time::{Duration, sleep};
+
+const MAX_NATIVE_CONNECTIONS: usize = 32;
+const MAX_CONNECTIONS_PER_AUTHORITY: usize = 8;
+
+#[derive(Clone, Debug, Eq, Hash, PartialEq)]
+struct AuthorityKey {
+    authority: String,
+    proxy: ProxyPolicy,
+}
+
+static AUTHORITY_BUDGETS: LazyLock<
+    Mutex<HashMap<AuthorityKey, Arc<Semaphore>>>,
+> = LazyLock::new(|| Mutex::new(HashMap::new()));
+static GLOBAL_BUDGET: LazyLock<Arc<Semaphore>> =
+    LazyLock::new(|| Arc::new(Semaphore::new(MAX_NATIVE_CONNECTIONS)));
+
+pub(crate) struct NativeBudgetPermit {
+    _global: OwnedSemaphorePermit,
+    _authority: Option<OwnedSemaphorePermit>,
+}
+
+fn budget(route: &DownloadRoute) -> Option<Arc<Semaphore>> {
+    let authority = crate::util::fetch::url_authority(&route.url)?;
+    let key = AuthorityKey {
+        authority,
+        proxy: route.proxy,
+    };
+    let mut budgets = AUTHORITY_BUDGETS.lock();
+    if budgets.len() >= 256 {
+        budgets.retain(|_, budget| Arc::strong_count(budget) > 1);
+    }
+    Some(
+        budgets
+            .entry(key)
+            .or_insert_with(|| {
+                Arc::new(Semaphore::new(MAX_CONNECTIONS_PER_AUTHORITY))
+            })
+            .clone(),
+    )
+}
+
+pub(crate) async fn acquire(
+    route: &DownloadRoute,
+) -> Result<NativeBudgetPermit, tokio::sync::AcquireError> {
+    let authority_budget = budget(route);
+    loop {
+        let global = Arc::clone(&GLOBAL_BUDGET).acquire_owned().await?;
+        let authority = match authority_budget.as_ref() {
+            Some(budget) => match Arc::clone(budget).try_acquire_owned() {
+                Ok(permit) => Some(permit),
+                Err(TryAcquireError::NoPermits) => {
+                    drop(global);
+                    sleep(Duration::from_millis(5)).await;
+                    continue;
+                }
+                Err(TryAcquireError::Closed) => {
+                    unreachable!("native authority budgets are never closed")
+                }
+            },
+            None => None,
+        };
+        return Ok(NativeBudgetPermit {
+            _global: global,
+            _authority: authority,
+        });
+    }
+}
+
+pub(crate) async fn acquire_many(
+    route: &DownloadRoute,
+    count: usize,
+) -> Result<Vec<NativeBudgetPermit>, tokio::sync::AcquireError> {
+    let authority_budget = budget(route);
+    let (global, authority) = loop {
+        let global = Arc::clone(&GLOBAL_BUDGET)
+            .acquire_many_owned(count as u32)
+            .await?;
+        let authority = match authority_budget.as_ref() {
+            Some(budget) => {
+                match Arc::clone(budget).try_acquire_many_owned(count as u32) {
+                    Ok(permit) => Some(permit),
+                    Err(TryAcquireError::NoPermits) => {
+                        drop(global);
+                        sleep(Duration::from_millis(5)).await;
+                        continue;
+                    }
+                    Err(TryAcquireError::Closed) => unreachable!(
+                        "native authority budgets are never closed"
+                    ),
+                }
+            }
+            None => None,
+        };
+        break (global, authority);
+    };
+    let mut global = global;
+    let mut authority = authority;
+    let mut permits = Vec::with_capacity(count);
+    for _ in 0..count {
+        let global = global
+            .split(1)
+            .expect("native global permit batch has enough permits");
+        let authority = authority.as_mut().map(|authority| {
+            authority
+                .split(1)
+                .expect("native authority permit batch has enough permits")
+        });
+        permits.push(NativeBudgetPermit {
+            _global: global,
+            _authority: authority,
+        });
+    }
+    Ok(permits)
+}
+
+pub(crate) fn try_acquire(
+    route: &DownloadRoute,
+) -> Result<NativeBudgetPermit, TryAcquireError> {
+    let global = Arc::clone(&GLOBAL_BUDGET).try_acquire_owned()?;
+    let authority = match budget(route) {
+        Some(budget) => match budget.try_acquire_owned() {
+            Ok(permit) => Some(permit),
+            Err(error) => return Err(error),
+        },
+        None => None,
+    };
+    Ok(NativeBudgetPermit {
+        _global: global,
+        _authority: authority,
+    })
+}
+
+pub(crate) fn available(route: &DownloadRoute) -> usize {
+    budget(route)
+        .map(|budget| budget.available_permits())
+        .unwrap_or(MAX_CONNECTIONS_PER_AUTHORITY)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::util::fetch::DownloadRouteSource;
+
+    fn route() -> DownloadRoute {
+        DownloadRoute {
+            url: "https://budget.example/file".to_string(),
+            source: DownloadRouteSource::Official,
+            is_mirror: false,
+            allow_sensitive_headers: true,
+            supports_range: true,
+            proxy: ProxyPolicy::Direct,
+        }
+    }
+
+    #[tokio::test]
+    async fn authority_budget_is_bounded() {
+        let route = route();
+        let mut permits = Vec::new();
+        for _ in 0..MAX_CONNECTIONS_PER_AUTHORITY {
+            permits.push(acquire(&route).await.unwrap());
+        }
+        assert!(matches!(
+            try_acquire(&route),
+            Err(TryAcquireError::NoPermits)
+        ));
+        drop(permits);
+        assert!(try_acquire(&route).is_ok());
+    }
+}

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

@@ -0,0 +1,429 @@
+//! Persisted route reputation for the native download engine.
+
+use crate::util::fetch::ProxyPolicy;
+use parking_lot::Mutex;
+use serde::{Deserialize, Serialize};
+use std::collections::HashMap;
+use std::path::{Path, PathBuf};
+use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
+use std::sync::{LazyLock, OnceLock};
+use std::time::{Duration, SystemTime, UNIX_EPOCH};
+
+const FILE_NAME: &str = "native-download-reputation.json";
+const SCHEMA_VERSION: u32 = 1;
+const MAX_ENTRIES: usize = 256;
+const MAX_AGE_SECS: u64 = 7 * 24 * 60 * 60;
+const WRITE_DELAY: Duration = Duration::from_secs(5);
+
+#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq, Serialize, Deserialize)]
+#[serde(rename_all = "snake_case")]
+pub(crate) enum NativeTransport {
+    H2Single,
+    H2MultiRange,
+    Http1MultiRange,
+}
+
+#[derive(Clone, Debug, Eq, Hash, PartialEq)]
+struct ReputationKey {
+    family: String,
+    authority: String,
+    proxy: ProxyPolicy,
+}
+
+#[derive(Clone, Copy, Debug, Default, Serialize, Deserialize)]
+pub(crate) struct NativeRouteReputation {
+    pub(crate) success_samples: u32,
+    pub(crate) failure_samples: u32,
+    #[serde(default)]
+    pub(crate) consecutive_failures: u32,
+    pub(crate) ttfb_ms: Option<f64>,
+    pub(crate) throughput_bps: Option<f64>,
+    updated_at: u64,
+}
+
+#[derive(Serialize, Deserialize)]
+struct PersistedStore {
+    version: u32,
+    routes: Vec<PersistedRoute>,
+    #[serde(default)]
+    transports: Vec<PersistedTransport>,
+}
+
+#[derive(Serialize, Deserialize)]
+struct PersistedRoute {
+    family: String,
+    authority: String,
+    proxy: ProxyPolicy,
+    #[serde(flatten)]
+    health: NativeRouteReputation,
+}
+
+#[derive(Clone, Copy, Debug, Default, Serialize, Deserialize)]
+pub(crate) struct NativeTransportReputation {
+    pub(crate) success_samples: u32,
+    pub(crate) throughput_bps: Option<f64>,
+    updated_at: u64,
+}
+
+#[derive(Clone, Debug, Eq, Hash, PartialEq)]
+struct TransportKey {
+    authority: String,
+    proxy: ProxyPolicy,
+    transport: NativeTransport,
+}
+
+#[derive(Serialize, Deserialize)]
+struct PersistedTransport {
+    authority: String,
+    proxy: ProxyPolicy,
+    transport: NativeTransport,
+    #[serde(flatten)]
+    health: NativeTransportReputation,
+}
+
+static REPUTATION: LazyLock<
+    Mutex<HashMap<ReputationKey, NativeRouteReputation>>,
+> = LazyLock::new(|| Mutex::new(HashMap::new()));
+static TRANSPORT_REPUTATION: LazyLock<
+    Mutex<HashMap<TransportKey, NativeTransportReputation>>,
+> = LazyLock::new(|| Mutex::new(HashMap::new()));
+static LOAD: tokio::sync::OnceCell<()> = tokio::sync::OnceCell::const_new();
+static PATH: OnceLock<PathBuf> = OnceLock::new();
+static GENERATION: AtomicU64 = AtomicU64::new(0);
+static WRITE_SCHEDULED: AtomicBool = AtomicBool::new(false);
+
+pub(crate) async fn load_if_needed() {
+    LOAD.get_or_init(|| async {
+		let Some(path) = reputation_path() else {
+			return;
+		};
+		let _ = PATH.set(path.clone());
+		let Ok(contents) = tokio::fs::read(&path).await else {
+			return;
+		};
+		let Ok(store) = serde_json::from_slice::<PersistedStore>(&contents) else {
+			tracing::warn!(path = %path.display(), "Ignoring invalid native download reputation");
+			return;
+		};
+		if store.version != SCHEMA_VERSION {
+			return;
+		}
+		let cutoff = now_secs().saturating_sub(MAX_AGE_SECS);
+		let mut reputation = REPUTATION.lock();
+		for route in store.routes.into_iter().take(MAX_ENTRIES) {
+			if route.health.updated_at < cutoff {
+				continue;
+			}
+			reputation.insert(
+				ReputationKey {
+					family: route.family,
+					authority: route.authority,
+					proxy: route.proxy,
+				},
+				route.health,
+			);
+		}
+		drop(reputation);
+		let mut transports = TRANSPORT_REPUTATION.lock();
+		for transport in store.transports.into_iter().take(MAX_ENTRIES) {
+			if transport.health.updated_at < cutoff {
+				continue;
+			}
+			transports.insert(
+				TransportKey {
+					authority: transport.authority,
+					proxy: transport.proxy,
+					transport: transport.transport,
+				},
+				transport.health,
+			);
+		}
+	})
+	.await;
+}
+
+pub(crate) fn get_transport(
+    authority: &str,
+    proxy: ProxyPolicy,
+    transport: NativeTransport,
+) -> Option<NativeTransportReputation> {
+    TRANSPORT_REPUTATION
+        .lock()
+        .get(&TransportKey {
+            authority: authority.to_string(),
+            proxy,
+            transport,
+        })
+        .copied()
+}
+
+pub(crate) fn record_transport_success(
+    authority: &str,
+    proxy: ProxyPolicy,
+    transport: NativeTransport,
+    throughput_bps: f64,
+) {
+    if !throughput_bps.is_finite() || throughput_bps <= 0.0 {
+        return;
+    }
+    let mut reputation = TRANSPORT_REPUTATION.lock();
+    let entry = reputation
+        .entry(TransportKey {
+            authority: authority.to_string(),
+            proxy,
+            transport,
+        })
+        .or_default();
+    entry.success_samples = entry.success_samples.saturating_add(1);
+    entry.throughput_bps =
+        Some(update_ewma(entry.throughput_bps, throughput_bps));
+    entry.updated_at = now_secs();
+    drop(reputation);
+    schedule_write();
+}
+
+pub(crate) fn get(
+    family: &str,
+    authority: &str,
+    proxy: ProxyPolicy,
+) -> Option<NativeRouteReputation> {
+    REPUTATION
+        .lock()
+        .get(&ReputationKey {
+            family: family.to_string(),
+            authority: authority.to_string(),
+            proxy,
+        })
+        .copied()
+}
+
+pub(crate) fn record_success(
+    family: &str,
+    authority: &str,
+    proxy: ProxyPolicy,
+    ttfb_ms: f64,
+    throughput_bps: Option<f64>,
+) {
+    let mut reputation = REPUTATION.lock();
+    let entry = reputation
+        .entry(ReputationKey {
+            family: family.to_string(),
+            authority: authority.to_string(),
+            proxy,
+        })
+        .or_default();
+    entry.success_samples = entry.success_samples.saturating_add(1);
+    entry.consecutive_failures = 0;
+    entry.ttfb_ms = Some(update_ewma(entry.ttfb_ms, ttfb_ms));
+    if let Some(throughput_bps) = throughput_bps {
+        entry.throughput_bps =
+            Some(update_ewma(entry.throughput_bps, throughput_bps));
+    }
+    entry.updated_at = now_secs();
+    drop(reputation);
+    schedule_write();
+}
+
+pub(crate) fn record_transfer_success(
+    family: &str,
+    authority: &str,
+    proxy: ProxyPolicy,
+    throughput_bps: f64,
+) {
+    if !throughput_bps.is_finite() || throughput_bps <= 0.0 {
+        return;
+    }
+    let mut reputation = REPUTATION.lock();
+    let entry = reputation
+        .entry(ReputationKey {
+            family: family.to_string(),
+            authority: authority.to_string(),
+            proxy,
+        })
+        .or_default();
+    entry.success_samples = entry.success_samples.saturating_add(1);
+    entry.consecutive_failures = 0;
+    entry.throughput_bps =
+        Some(update_ewma(entry.throughput_bps, throughput_bps));
+    entry.updated_at = now_secs();
+    drop(reputation);
+    schedule_write();
+}
+
+pub(crate) fn record_failure(
+    family: &str,
+    authority: &str,
+    proxy: ProxyPolicy,
+) {
+    let mut reputation = REPUTATION.lock();
+    let entry = reputation
+        .entry(ReputationKey {
+            family: family.to_string(),
+            authority: authority.to_string(),
+            proxy,
+        })
+        .or_default();
+    entry.failure_samples = entry.failure_samples.saturating_add(1);
+    entry.consecutive_failures = entry.consecutive_failures.saturating_add(1);
+    entry.updated_at = now_secs();
+    drop(reputation);
+    schedule_write();
+}
+
+fn update_ewma(current: Option<f64>, sample: f64) -> f64 {
+    current.map_or(sample, |current| current * 0.75 + sample * 0.25)
+}
+
+fn schedule_write() {
+    GENERATION.fetch_add(1, Ordering::AcqRel);
+    let Ok(runtime) = tokio::runtime::Handle::try_current() else {
+        return;
+    };
+    if WRITE_SCHEDULED.swap(true, Ordering::AcqRel) {
+        return;
+    }
+    runtime.spawn(async {
+		loop {
+			let generation = GENERATION.load(Ordering::Acquire);
+			tokio::time::sleep(WRITE_DELAY).await;
+			if let Err(error) = write_snapshot().await {
+				tracing::warn!(%error, "Failed to persist native download reputation");
+			}
+			if GENERATION.load(Ordering::Acquire) == generation {
+				WRITE_SCHEDULED.store(false, Ordering::Release);
+				if GENERATION.load(Ordering::Acquire) == generation
+					|| WRITE_SCHEDULED.swap(true, Ordering::AcqRel)
+				{
+					break;
+				}
+			}
+		}
+	});
+}
+
+async fn write_snapshot() -> std::io::Result<()> {
+    let Some(path) = PATH.get().cloned().or_else(reputation_path) else {
+        return Ok(());
+    };
+    let mut routes = REPUTATION
+        .lock()
+        .iter()
+        .map(|(key, health)| PersistedRoute {
+            family: key.family.clone(),
+            authority: key.authority.clone(),
+            proxy: key.proxy,
+            health: *health,
+        })
+        .collect::<Vec<_>>();
+    routes.sort_unstable_by_key(|route| {
+        std::cmp::Reverse(route.health.updated_at)
+    });
+    routes.truncate(MAX_ENTRIES);
+    let mut transports = TRANSPORT_REPUTATION
+        .lock()
+        .iter()
+        .map(|(key, health)| PersistedTransport {
+            authority: key.authority.clone(),
+            proxy: key.proxy,
+            transport: key.transport,
+            health: *health,
+        })
+        .collect::<Vec<_>>();
+    transports.sort_unstable_by_key(|transport| {
+        std::cmp::Reverse(transport.health.updated_at)
+    });
+    transports.truncate(MAX_ENTRIES);
+    let bytes = serde_json::to_vec(&PersistedStore {
+        version: SCHEMA_VERSION,
+        routes,
+        transports,
+    })
+    .map_err(std::io::Error::other)?;
+    if let Some(parent) = path.parent() {
+        tokio::fs::create_dir_all(parent).await?;
+    }
+    let temporary = temporary_path(&path);
+    tokio::fs::write(&temporary, bytes).await?;
+    if let Err(error) = tokio::fs::rename(&temporary, &path).await {
+        if error.kind() != std::io::ErrorKind::AlreadyExists {
+            let _ = tokio::fs::remove_file(&temporary).await;
+            return Err(error);
+        }
+        tokio::fs::remove_file(&path).await?;
+        tokio::fs::rename(&temporary, &path).await?;
+    }
+    Ok(())
+}
+
+fn reputation_path() -> Option<PathBuf> {
+    crate::State::get_if_initialized()
+        .map(|state| state.directories.settings_dir.join(FILE_NAME))
+}
+
+fn temporary_path(path: &Path) -> PathBuf {
+    let mut name = path.file_name().unwrap_or_default().to_os_string();
+    name.push(".tmp");
+    path.with_file_name(name)
+}
+
+fn now_secs() -> u64 {
+    SystemTime::now()
+        .duration_since(UNIX_EPOCH)
+        .map(|duration| duration.as_secs())
+        .unwrap_or(0)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn transport_reputation_is_independent() {
+        record_transport_success(
+            "transport-reputation.example:443",
+            ProxyPolicy::Direct,
+            NativeTransport::H2Single,
+            1024.0,
+        );
+        assert!(
+            get_transport(
+                "transport-reputation.example:443",
+                ProxyPolicy::Direct,
+                NativeTransport::H2Single,
+            )
+            .is_some()
+        );
+        assert!(
+            get_transport(
+                "transport-reputation.example:443",
+                ProxyPolicy::Direct,
+                NativeTransport::Http1MultiRange,
+            )
+            .is_none()
+        );
+        assert!(
+            get_transport(
+                "transport-reputation.example:443",
+                ProxyPolicy::Direct,
+                NativeTransport::H2MultiRange,
+            )
+            .is_none()
+        );
+        record_transport_success(
+            "transport-reputation.example:443",
+            ProxyPolicy::Direct,
+            NativeTransport::H2MultiRange,
+            4096.0,
+        );
+        assert_eq!(
+            get_transport(
+                "transport-reputation.example:443",
+                ProxyPolicy::Direct,
+                NativeTransport::H2Single,
+            )
+            .unwrap()
+            .throughput_bps,
+            Some(1024.0),
+        );
+    }
+}

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

@@ -0,0 +1,161 @@
+//! Slow-transfer and optimal-stop policy for the native download engine.
+
+use std::time::{Duration, Instant};
+
+const WINDOW: Duration = Duration::from_secs(3);
+const MIN_FLOW: Duration = Duration::from_secs(10);
+const REQUIRED_SLOW_WINDOWS: u8 = 2;
+const MIN_REMAINING_BYTES: u64 = 1024 * 1024;
+const ABSOLUTE_SPEED_FLOOR: u64 = 16 * 1024;
+const COLD_SPEED_FLOOR: u64 = 256 * 1024;
+const RECONNECT_OVERHEAD: Duration = Duration::from_millis(600);
+const MIN_SAVINGS: Duration = Duration::from_secs(2);
+#[cfg(not(test))]
+const IDLE_TIMEOUT: Duration = Duration::from_secs(20);
+#[cfg(test)]
+const IDLE_TIMEOUT: Duration = Duration::from_millis(500);
+
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+pub(crate) enum SlowDecision {
+    Continue,
+    Probe { bytes_per_second: u64 },
+    Idle { elapsed: Duration },
+    Commit,
+}
+
+pub(crate) struct NativeSlowPolicy {
+    started_at: Instant,
+    window_started_at: Instant,
+    window_start_bytes: u64,
+    slow_windows: u8,
+    committed: bool,
+    expected_speed: Option<u64>,
+}
+
+impl NativeSlowPolicy {
+    pub(crate) fn new(
+        starting_bytes: u64,
+        expected_speed: Option<u64>,
+    ) -> Self {
+        let now = Instant::now();
+        Self {
+            started_at: now,
+            window_started_at: now,
+            window_start_bytes: starting_bytes,
+            slow_windows: 0,
+            committed: false,
+            expected_speed,
+        }
+    }
+
+    pub(crate) fn observe(
+        &mut self,
+        downloaded: u64,
+        remaining: u64,
+    ) -> SlowDecision {
+        if self.committed {
+            return SlowDecision::Continue;
+        }
+        let elapsed = self.window_started_at.elapsed();
+        if downloaded == self.window_start_bytes && elapsed >= IDLE_TIMEOUT {
+            return SlowDecision::Idle { elapsed };
+        }
+        if elapsed < WINDOW {
+            return SlowDecision::Continue;
+        }
+        let speed = downloaded
+            .saturating_sub(self.window_start_bytes)
+            .checked_div(elapsed.as_secs().max(1))
+            .unwrap_or(0);
+        self.window_started_at = Instant::now();
+        self.window_start_bytes = downloaded;
+
+        if remaining < MIN_REMAINING_BYTES
+            || estimated_duration(remaining, speed)
+                <= RECONNECT_OVERHEAD + MIN_SAVINGS
+        {
+            self.committed = true;
+            return SlowDecision::Commit;
+        }
+        let expected_floor = self
+            .expected_speed
+            .map(|expected| expected.saturating_mul(40) / 100)
+            .unwrap_or(COLD_SPEED_FLOOR)
+            .max(ABSOLUTE_SPEED_FLOOR);
+        if speed < expected_floor {
+            self.slow_windows = self.slow_windows.saturating_add(1);
+        } else {
+            self.slow_windows = 0;
+        }
+        if self.started_at.elapsed() >= MIN_FLOW
+            && self.slow_windows >= REQUIRED_SLOW_WINDOWS
+        {
+            self.slow_windows = 0;
+            SlowDecision::Probe {
+                bytes_per_second: speed,
+            }
+        } else {
+            SlowDecision::Continue
+        }
+    }
+
+    pub(crate) fn commit(&mut self) {
+        self.committed = true;
+    }
+}
+
+pub(crate) fn should_switch(
+    current_speed: u64,
+    candidate_speed: u64,
+    remaining_bytes: u64,
+    restart_bytes: u64,
+) -> bool {
+    if current_speed == 0 || candidate_speed == 0 {
+        return false;
+    }
+    if candidate_speed.saturating_mul(100) < current_speed.saturating_mul(125) {
+        return false;
+    }
+    let stay = estimated_duration(remaining_bytes, current_speed);
+    let switch =
+        RECONNECT_OVERHEAD + estimated_duration(restart_bytes, candidate_speed);
+    let required_savings = MIN_SAVINGS.max(stay.mul_f64(0.15));
+    switch.saturating_add(required_savings) < stay
+}
+
+fn estimated_duration(bytes: u64, speed: u64) -> Duration {
+    if speed == 0 {
+        return Duration::MAX;
+    }
+    Duration::from_secs_f64(bytes as f64 / speed as f64)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn switch_requires_repayment_of_restarted_bytes() {
+        assert!(!should_switch(1024, 2048, 1024, 1024 * 1024));
+    }
+
+    #[test]
+    fn faster_candidate_wins_early_in_a_large_download() {
+        assert!(should_switch(
+            256 * 1024,
+            2 * 1024 * 1024,
+            64 * 1024 * 1024,
+            64 * 1024 * 1024,
+        ));
+    }
+
+    #[test]
+    fn idle_body_is_a_terminal_recovery_decision() {
+        let mut policy = NativeSlowPolicy::new(0, None);
+        std::thread::sleep(IDLE_TIMEOUT + Duration::from_millis(10));
+        assert!(matches!(
+            policy.observe(0, 64 * 1024 * 1024),
+            SlowDecision::Idle { .. }
+        ));
+    }
+}

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

@@ -0,0 +1,311 @@
+//! Positional output for native HTTP/1.1 and HTTP/2 range downloads.
+
+use crate::util::io::{self, IOError};
+use std::path::{Path, PathBuf};
+use std::sync::Arc;
+use tokio::fs::{File, OpenOptions};
+use tokio::io::{AsyncSeekExt, AsyncWriteExt, BufWriter, SeekFrom};
+
+const RANGE_WRITER_BUFFER_CAPACITY: usize = 256 * 1024;
+
+#[cfg(test)]
+struct RangeWriteTestProbe {
+    delay: std::time::Duration,
+    in_flight: std::sync::atomic::AtomicUsize,
+    max_in_flight: std::sync::atomic::AtomicUsize,
+}
+
+#[cfg(test)]
+static RANGE_WRITE_TEST_PROBE: std::sync::Mutex<
+    Option<Arc<RangeWriteTestProbe>>,
+> = std::sync::Mutex::new(None);
+
+/// A preallocated `.part` file. Writers open their own handle so a slow range
+/// never serializes unrelated ranges behind a file mutex.
+pub(crate) struct RangeOutput {
+    path: PathBuf,
+    size: u64,
+}
+
+/// An independent sequential writer for one half-open byte range.
+pub(crate) struct RangeWriter {
+    file: BufWriter<File>,
+    offset: u64,
+    end_exclusive: u64,
+    path: PathBuf,
+}
+
+impl RangeOutput {
+    pub(crate) async fn create(
+        path: &Path,
+        size: u64,
+    ) -> Result<Arc<Self>, IOError> {
+        io::retry_windows_sharing_violation(
+            path,
+            "creating ranged download output",
+            || async {
+                let file = OpenOptions::new()
+                    .create(true)
+                    .truncate(true)
+                    .read(true)
+                    .write(true)
+                    .open(path)
+                    .await?;
+                file.set_len(size).await?;
+                Ok(())
+            },
+        )
+        .await
+        .map_err(|error| io::io_error_with_lock_info(error, path))?;
+        Ok(Arc::new(Self {
+            path: path.to_path_buf(),
+            size,
+        }))
+    }
+
+    pub(crate) async fn open_range(
+        &self,
+        start: u64,
+        end_exclusive: u64,
+    ) -> Result<RangeWriter, IOError> {
+        if start >= end_exclusive || end_exclusive > self.size {
+            return Err(IOError::with_path(
+                std::io::Error::new(
+                    std::io::ErrorKind::InvalidInput,
+                    "invalid range bounds",
+                ),
+                &self.path,
+            ));
+        }
+        let path = self.path.clone();
+        let mut file = io::retry_windows_sharing_violation(
+            &path,
+            "opening ranged download output",
+            || async {
+                OpenOptions::new().read(true).write(true).open(&path).await
+            },
+        )
+        .await
+        .map_err(|error| io::io_error_with_lock_info(error, &path))?;
+        file.seek(SeekFrom::Start(start))
+            .await
+            .map_err(|error| io::io_error_with_lock_info(error, &path))?;
+        Ok(RangeWriter {
+            file: BufWriter::with_capacity(RANGE_WRITER_BUFFER_CAPACITY, file),
+            offset: start,
+            end_exclusive,
+            path,
+        })
+    }
+
+    /// Legacy adapter; not used by the concurrent range hot path.
+    #[allow(dead_code)]
+    pub(crate) async fn write_at(
+        &self,
+        offset: u64,
+        bytes: &[u8],
+        path: &Path,
+    ) -> Result<(), IOError> {
+        let end = offset.checked_add(bytes.len() as u64).ok_or_else(|| {
+            IOError::with_path(
+                std::io::Error::new(
+                    std::io::ErrorKind::InvalidInput,
+                    "offset overflow",
+                ),
+                path,
+            )
+        })?;
+        let mut writer = self.open_range(offset, end).await?;
+        writer.write_next(bytes).await?;
+        writer.flush().await
+    }
+
+    /// Legacy adapter for callers that previously flushed the shared handle.
+    #[allow(dead_code)]
+    pub(crate) async fn flush(&self, path: &Path) -> Result<(), IOError> {
+        let mut file = io::retry_windows_sharing_violation(
+            path,
+            "flushing ranged download output",
+            || async {
+                OpenOptions::new().read(true).write(true).open(path).await
+            },
+        )
+        .await
+        .map_err(|error| io::io_error_with_lock_info(error, path))?;
+        file.flush()
+            .await
+            .map_err(|error| io::io_error_with_lock_info(error, path))
+    }
+}
+
+impl RangeWriter {
+    #[cfg(test)]
+    pub(crate) fn offset(&self) -> u64 {
+        self.offset
+    }
+
+    pub(crate) fn remaining(&self) -> u64 {
+        self.end_exclusive.saturating_sub(self.offset)
+    }
+
+    pub(crate) async fn write_next(
+        &mut self,
+        bytes: &[u8],
+    ) -> Result<(), IOError> {
+        if bytes.len() as u64 > self.remaining() {
+            return Err(IOError::with_path(
+                std::io::Error::new(
+                    std::io::ErrorKind::InvalidInput,
+                    "write exceeds range",
+                ),
+                &self.path,
+            ));
+        }
+        #[cfg(test)]
+        let probe = RANGE_WRITE_TEST_PROBE.lock().unwrap().clone();
+        #[cfg(test)]
+        if let Some(probe) = &probe {
+            let in_flight = probe
+                .in_flight
+                .fetch_add(1, std::sync::atomic::Ordering::AcqRel)
+                + 1;
+            probe
+                .max_in_flight
+                .fetch_max(in_flight, std::sync::atomic::Ordering::AcqRel);
+            tokio::time::sleep(probe.delay).await;
+        }
+
+        let result =
+            self.file.write_all(bytes).await.map_err(|error| {
+                io::io_error_with_lock_info(error, &self.path)
+            });
+        #[cfg(test)]
+        if let Some(probe) = probe {
+            probe
+                .in_flight
+                .fetch_sub(1, std::sync::atomic::Ordering::AcqRel);
+        }
+        result?;
+        self.offset += bytes.len() as u64;
+        Ok(())
+    }
+
+    pub(crate) async fn flush(&mut self) -> Result<(), IOError> {
+        self.file
+            .flush()
+            .await
+            .map_err(|error| io::io_error_with_lock_info(error, &self.path))
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use futures::future::try_join_all;
+    use std::sync::atomic::Ordering;
+
+    #[tokio::test]
+    async fn writes_non_overlapping_ranges_with_independent_writers() {
+        let directory = tempfile::tempdir().unwrap();
+        let path = directory.path().join("ranged.part");
+        let output = RangeOutput::create(&path, 12).await.unwrap();
+        let mut a = output.open_range(0, 4).await.unwrap();
+        let mut b = output.open_range(4, 8).await.unwrap();
+        let mut c = output.open_range(8, 12).await.unwrap();
+        a.write_next(b"abcd").await.unwrap();
+        b.write_next(b"efgh").await.unwrap();
+        c.write_next(b"ijkl").await.unwrap();
+        a.flush().await.unwrap();
+        b.flush().await.unwrap();
+        c.flush().await.unwrap();
+        assert_eq!(tokio::fs::read(path).await.unwrap(), b"abcdefghijkl");
+    }
+
+    #[tokio::test]
+    async fn rejects_writes_beyond_range() {
+        let directory = tempfile::tempdir().unwrap();
+        let path = directory.path().join("ranged.part");
+        let output = RangeOutput::create(&path, 8).await.unwrap();
+        let mut writer = output.open_range(0, 4).await.unwrap();
+        assert!(writer.write_next(b"12345").await.is_err());
+        assert_eq!(writer.offset(), 0);
+    }
+
+    #[tokio::test]
+    async fn rejects_empty_and_out_of_file_ranges() {
+        let directory = tempfile::tempdir().unwrap();
+        let path = directory.path().join("ranged.part");
+        let output = RangeOutput::create(&path, 8).await.unwrap();
+        assert!(output.open_range(2, 2).await.is_err());
+        assert!(output.open_range(7, 9).await.is_err());
+    }
+
+    #[tokio::test]
+    async fn buffers_small_exact_and_large_writes() {
+        let directory = tempfile::tempdir().unwrap();
+        let path = directory.path().join("ranged.part");
+        let small = vec![1_u8; 1024];
+        let exact = vec![2_u8; RANGE_WRITER_BUFFER_CAPACITY];
+        let large = vec![3_u8; RANGE_WRITER_BUFFER_CAPACITY + 1];
+        let total = small.len() + exact.len() + large.len();
+        let output = RangeOutput::create(&path, total as u64).await.unwrap();
+        let mut writer = output.open_range(0, total as u64).await.unwrap();
+
+        writer.write_next(&small).await.unwrap();
+        writer.write_next(&exact).await.unwrap();
+        writer.write_next(&large).await.unwrap();
+        writer.flush().await.unwrap();
+
+        let mut expected = small;
+        expected.extend(exact);
+        expected.extend(large);
+        assert_eq!(tokio::fs::read(path).await.unwrap(), expected);
+    }
+
+    #[tokio::test]
+    async fn concurrent_writers_preserve_all_ranges_and_overlap() {
+        let probe = Arc::new(RangeWriteTestProbe {
+            delay: std::time::Duration::from_millis(2),
+            in_flight: std::sync::atomic::AtomicUsize::new(0),
+            max_in_flight: std::sync::atomic::AtomicUsize::new(0),
+        });
+        *RANGE_WRITE_TEST_PROBE.lock().unwrap() = Some(Arc::clone(&probe));
+
+        for range_count in [2, 4, 8] {
+            let directory = tempfile::tempdir().unwrap();
+            let path = directory.path().join("ranged.part");
+            let range_length = 32 * 1024;
+            let expected = (0..range_length * range_count)
+                .map(|index| (index % 251) as u8)
+                .collect::<Vec<_>>();
+            let output = RangeOutput::create(&path, expected.len() as u64)
+                .await
+                .unwrap();
+            let writes = (0..range_count).map(|index| {
+                let output = Arc::clone(&output);
+                let start = index * range_length;
+                let bytes = expected[start..start + range_length].to_vec();
+                async move {
+                    let mut writer = output
+                        .open_range(start as u64, (start + range_length) as u64)
+                        .await?;
+                    let mut offset = 0;
+                    while offset < bytes.len() {
+                        let length = ((index * 521 + offset) % 4096 + 1)
+                            .min(bytes.len() - offset);
+                        writer
+                            .write_next(&bytes[offset..offset + length])
+                            .await?;
+                        offset += length;
+                    }
+                    writer.flush().await
+                }
+            });
+            try_join_all(writes).await.unwrap();
+            assert_eq!(tokio::fs::read(path).await.unwrap(), expected);
+        }
+        *RANGE_WRITE_TEST_PROBE.lock().unwrap() = None;
+
+        assert!(probe.max_in_flight.load(Ordering::Acquire) > 1);
+    }
+}

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

@@ -0,0 +1,12 @@
+#![allow(dead_code)]
+//! Shared download helpers used by both the legacy and XMCL-compatible
+//! engines.
+//!
+//! Route resolution, integrity verification, `.part` handling, atomic
+//! rename, and Windows file-lock retry logic will move here as the
+//! legacy engine is split out of `fetch.rs`.
+
+pub const SEGMENTED_DOWNLOAD_THRESHOLD: u64 = 512 * 1024;
+pub const XMCL_RANGE_CONCURRENCY: usize = 4;
+pub const XMCL_BMCL_CONCURRENCY: usize = 16;
+pub const XMCL_OTHER_CONCURRENCY: usize = 16;

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

@@ -0,0 +1,244 @@
+#![allow(dead_code)]
+//! Slow-download detection.
+//!
+//! The detector is intentionally local and cheap. It only fires while a
+//! download is stalled or wasting work, and the resulting events are
+//! written to the local compact log.
+
+use serde::{Deserialize, Serialize};
+use std::time::Instant;
+
+#[derive(Debug, Clone, Copy, Eq, PartialEq, Serialize, Deserialize)]
+#[serde(rename_all = "snake_case")]
+pub enum SlowRule {
+    R1NoProgress,
+    R2BelowExpectation,
+    R3SegmentWaste,
+    R4FrequentSwitches,
+}
+
+#[derive(
+    Debug, Clone, Copy, Default, Eq, PartialEq, Serialize, Deserialize,
+)]
+#[serde(rename_all = "snake_case")]
+pub enum SlowPreset {
+    #[default]
+    Sensitive,
+    Standard,
+    Relaxed,
+}
+
+impl SlowPreset {
+    pub const fn no_progress_secs(self) -> u64 {
+        match self {
+            Self::Sensitive => 3,
+            Self::Standard => 5,
+            Self::Relaxed => 10,
+        }
+    }
+
+    pub const fn min_speed_bps(self) -> u64 {
+        match self {
+            Self::Sensitive => 64 * 1024,
+            Self::Standard => 32 * 1024,
+            Self::Relaxed => 16 * 1024,
+        }
+    }
+
+    pub const fn min_remaining_bytes(self) -> u64 {
+        match self {
+            Self::Sensitive => 256 * 1024,
+            Self::Standard => 1024 * 1024,
+            Self::Relaxed => 4 * 1024 * 1024,
+        }
+    }
+
+    pub const fn max_segment_waste_ratio(self) -> f64 {
+        match self {
+            Self::Sensitive => 0.15,
+            Self::Standard => 0.25,
+            Self::Relaxed => 0.5,
+        }
+    }
+
+    pub const fn max_switches_per_file(self) -> u32 {
+        match self {
+            Self::Sensitive => 2,
+            Self::Standard => 4,
+            Self::Relaxed => 8,
+        }
+    }
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct SlowEvent {
+    pub rule: SlowRule,
+    pub engine: String,
+    pub source: String,
+    pub detail: String,
+}
+
+#[derive(Debug, Clone)]
+pub struct SlowDetector {
+    preset: SlowPreset,
+    started_at: Option<Instant>,
+    last_progress_at: Option<Instant>,
+    last_bytes: u64,
+    total: Option<u64>,
+    segment_waste: u64,
+    route_switches: u32,
+}
+
+impl Default for SlowDetector {
+    fn default() -> Self {
+        Self::new(SlowPreset::Standard)
+    }
+}
+
+impl SlowDetector {
+    pub fn new(preset: SlowPreset) -> Self {
+        Self {
+            preset,
+            started_at: None,
+            last_progress_at: None,
+            last_bytes: 0,
+            total: None,
+            segment_waste: 0,
+            route_switches: 0,
+        }
+    }
+
+    pub fn start(&mut self, total: Option<u64>) {
+        let now = Instant::now();
+        self.started_at = Some(now);
+        self.last_progress_at = Some(now);
+        self.last_bytes = 0;
+        self.total = total;
+        self.segment_waste = 0;
+        self.route_switches = 0;
+    }
+
+    pub fn progress(&mut self, bytes: u64, source: &str) -> Option<SlowEvent> {
+        let now = Instant::now();
+        let Some(started) = self.started_at else {
+            return None;
+        };
+        let elapsed = now.duration_since(started).as_secs();
+        let delta = bytes.saturating_sub(self.last_bytes);
+
+        if delta > 0 {
+            self.last_bytes = bytes;
+            self.last_progress_at = Some(now);
+            return None;
+        }
+
+        let no_progress_secs = now
+            .duration_since(self.last_progress_at.unwrap_or(started))
+            .as_secs();
+        if no_progress_secs >= self.preset.no_progress_secs() {
+            let remaining = self.total.map(|total| total.saturating_sub(bytes));
+            if remaining.map_or(true, |value| {
+                value >= self.preset.min_remaining_bytes()
+            }) {
+                return Some(SlowEvent {
+                    rule: SlowRule::R1NoProgress,
+                    engine: "xmcl".to_string(),
+                    source: source.to_string(),
+                    detail: format!(
+                        "no_progress={no_progress_secs}s elapsed={elapsed}s"
+                    ),
+                });
+            }
+        }
+
+        None
+    }
+
+    pub fn record_segment_waste(
+        &mut self,
+        wasted_bytes: u64,
+        source: &str,
+    ) -> Option<SlowEvent> {
+        self.segment_waste = self.segment_waste.saturating_add(wasted_bytes);
+        let total = self.total.unwrap_or(0);
+        if total == 0 {
+            return None;
+        }
+        let ratio = self.segment_waste as f64 / total as f64;
+        if ratio >= self.preset.max_segment_waste_ratio() {
+            Some(SlowEvent {
+                rule: SlowRule::R3SegmentWaste,
+                engine: "xmcl".to_string(),
+                source: source.to_string(),
+                detail: format!(
+                    "wasted={} total={total} ratio={ratio:.2}",
+                    self.segment_waste
+                ),
+            })
+        } else {
+            None
+        }
+    }
+
+    pub fn record_route_switch(&mut self, source: &str) -> Option<SlowEvent> {
+        self.route_switches += 1;
+        if self.route_switches > self.preset.max_switches_per_file() {
+            Some(SlowEvent {
+                rule: SlowRule::R4FrequentSwitches,
+                engine: "xmcl".to_string(),
+                source: source.to_string(),
+                detail: format!("switches={}", self.route_switches),
+            })
+        } else {
+            None
+        }
+    }
+
+    pub fn finish(&mut self) {
+        self.started_at = None;
+        self.last_progress_at = None;
+        self.last_bytes = 0;
+        self.total = None;
+        self.segment_waste = 0;
+        self.route_switches = 0;
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::time::Duration;
+
+    #[test]
+    fn no_progress_fires_after_preset_window() {
+        let mut detector = SlowDetector::new(SlowPreset::Sensitive);
+        detector.start(Some(4 * 1024 * 1024));
+        detector.progress(0, "official");
+        std::thread::sleep(Duration::from_millis(3100));
+        let event = detector.progress(0, "official");
+        assert!(matches!(
+            event,
+            Some(SlowEvent {
+                rule: SlowRule::R1NoProgress,
+                ..
+            })
+        ));
+    }
+
+    #[test]
+    fn route_switch_fires_after_preset_limit() {
+        let mut detector = SlowDetector::new(SlowPreset::Sensitive);
+        detector.start(Some(1024));
+        let mut fired = None;
+        for _ in 0..3 {
+            fired = detector.record_route_switch("bmclapi");
+        }
+        assert!(matches!(
+            fired,
+            Some(SlowEvent {
+                rule: SlowRule::R4FrequentSwitches,
+                ..
+            })
+        ));
+    }
+}