uv_distribution/

distribution_database.rs

use std::cmp::Reverse;
use std::future::Future;
use std::io;
use std::path::Path;
use std::pin::Pin;
use std::sync::Arc;
use std::task::{Context, Poll};

use futures::{FutureExt, TryStreamExt};
use http_content_range::{ContentRange, ContentRangeBytes, ContentRangeUnbound};
use rayon::in_place_scope;
use rayon::prelude::*;
use rustc_hash::FxHashMap;
use tokio::io::{AsyncRead, AsyncSeekExt, AsyncWriteExt, ReadBuf};
use tokio::sync::Semaphore;
use tokio_util::compat::FuturesAsyncReadCompatExt;
use tracing::{Instrument, debug, info_span, instrument, warn};
use url::Url;

use uv_cache::{ArchiveFileId, ArchiveId, Cache, CacheBucket, CacheEntry, WheelCache};
use uv_cache_info::{CacheInfo, Timestamp};
use uv_client::{
    CacheControl, CachedClientError, Connectivity, DataWithCachePolicy, RegistryClient,
    RequestBuilder, RetryState,
};
use uv_configuration::initialize_rayon_once;
use uv_distribution_filename::WheelFilename;
use uv_distribution_types::{
    ArchiveHashPolicy, BuildInfo, BuildableSource, BuiltDist, Dist, DistRef, HashCollection,
    HashValidation, Hashed, IndexUrl, InstalledDist, MetadataHashPolicy, Name, SourceDist,
    SourceUrl, parse_url_hashes,
};
use uv_extract::dirhash::{DirectoryDigest, HashedFile};
use uv_extract::hash::Hasher;
use uv_fs::{LockedFile, write_atomic};
use uv_git::{GIT_LFS, GitError};
use uv_platform_tags::Tags;
use uv_preview::PreviewFeature;
use uv_pypi_types::{HashDigest, HashDigests, PyProjectToml};
use uv_python::PythonVariant;
use uv_redacted::DisplaySafeUrl;
use uv_types::{BuildContext, BuildStack};

use crate::archive::Archive;
use crate::error::PythonVersion;
use crate::extracted_wheel::{ExtractedWheel, HashedWheel, WheelExtractor};
use crate::hash::http_hash_algorithms;
use crate::metadata::{ArchiveMetadata, Metadata};
use crate::source::SourceDistributionBuilder;
use crate::{Error, LocalWheel, Reporter, RequiresDist};

/// A cached high-level interface to convert distributions (a requirement resolved to a location)
/// to a wheel or wheel metadata.
///
/// For wheel metadata, this happens by either fetching the metadata from the remote wheel or by
/// building the source distribution. For wheel files, either the wheel is downloaded or a source
/// distribution is downloaded, built and the new wheel gets returned.
///
/// All kinds of wheel sources (index, URL, path) and source distribution source (index, URL, path,
/// Git) are supported.
///
/// This struct also has the task of acquiring locks around source dist builds in general and git
/// operation especially, as well as respecting concurrency limits.
pub struct DistributionDatabase<'a, Context: BuildContext> {
    build_context: &'a Context,
    builder: SourceDistributionBuilder<'a, Context>,
    client: ManagedClient<'a>,
    reporter: Option<Arc<dyn Reporter>>,
    content_addressed_cache: bool,
}

impl<'a, Context: BuildContext> DistributionDatabase<'a, Context> {
    pub fn new(
        client: &'a RegistryClient,
        build_context: &'a Context,
        downloads_semaphore: Arc<Semaphore>,
    ) -> Self {
        // When ZIP validation is disabled, the extracted tree can contain files that aren't
        // represented in the central directory and therefore aren't included in its digest.
        // Avoid using an incomplete digest as a content-addressed archive ID.
        let content_addressed_cache = uv_preview::is_enabled(PreviewFeature::ContentAddressedCache)
            && !uv_extract::insecure_no_validate();
        Self {
            build_context,
            builder: SourceDistributionBuilder::new(build_context),
            client: ManagedClient::new(client, downloads_semaphore),
            reporter: None,
            content_addressed_cache,
        }
    }

    /// Set the build stack to use for the [`DistributionDatabase`].
    #[must_use]
    pub fn with_build_stack(self, build_stack: &'a BuildStack) -> Self {
        Self {
            builder: self.builder.with_build_stack(build_stack),
            ..self
        }
    }

    /// Set the [`Reporter`] to use for the [`DistributionDatabase`].
    #[must_use]
    pub fn with_reporter(self, reporter: Arc<dyn Reporter>) -> Self {
        Self {
            builder: self.builder.with_reporter(reporter.clone()),
            reporter: Some(reporter),
            ..self
        }
    }

    /// Handle a specific `reqwest` error, and convert it to [`io::Error`].
    fn handle_response_errors(&self, err: reqwest::Error) -> io::Error {
        if err.is_timeout() {
            // Assumption: The connect timeout with the 10s default is not the culprit.
            io::Error::new(
                io::ErrorKind::TimedOut,
                format!(
                    "Failed to download distribution due to network timeout. Try increasing UV_HTTP_TIMEOUT (current value: {}s).",
                    self.client.unmanaged.read_timeout().as_secs()
                ),
            )
        } else {
            io::Error::other(err)
        }
    }

    /// Acquire an advisory lock for a wheel cache entry.
    ///
    /// A remote wheel's content hash is not always available until after the download, so
    /// concurrent cache fills coordinate on the wheel cache entry instead. The entry is already
    /// scoped to the distribution's source and wheel filename.
    ///
    /// Callers hold the returned lock across cache lookup, download or extraction, and publication.
    /// A process that waited for another cache fill therefore rechecks and reuses the completed entry.
    async fn lock_wheel(
        wheel_entry: &CacheEntry,
        filename: &WheelFilename,
    ) -> Result<LockedFile, Error> {
        // For backwards compatibility, we use the full wheel stem on Windows. Local wheel
        // extraction and older uv versions use the same key, so changing it would prevent them
        // from coordinating through a shared cache.
        #[cfg(windows)]
        let lock_key = filename.stem();
        // On other platforms, we use the bounded cache key to avoid filesystem filename limits.
        #[cfg(not(windows))]
        let lock_key = filename.cache_key();

        let lock_entry = wheel_entry.with_file(format!("{lock_key}.lock"));
        lock_entry.lock().await.map_err(Error::CacheLock)
    }

    /// Either fetch the wheel or fetch and build the source distribution
    ///
    /// Returns a wheel that's compliant with the given platform tags.
    ///
    /// While hashes will be generated in some cases, hash-checking is only enforced for source
    /// distributions, and should be enforced by the caller for wheels.
    #[instrument(skip_all, fields(%dist))]
    pub async fn get_or_build_wheel(
        &self,
        dist: &Dist,
        tags: &Tags,
        hashes: ArchiveHashPolicy<'_>,
    ) -> Result<LocalWheel, Error> {
        match dist {
            Dist::Built(built) => self.get_wheel(built, hashes).await,
            Dist::Source(source) => self.build_wheel(source, tags, hashes).await,
        }
    }

    /// Either fetch the only wheel metadata (directly from the index or with range requests) or
    /// fetch and build the source distribution.
    ///
    /// While hashes will be generated in some cases, hash-checking is only enforced for source
    /// distributions, and should be enforced by the caller for wheels.
    #[instrument(skip_all, fields(%dist))]
    pub async fn get_installed_metadata(
        &self,
        dist: &InstalledDist,
    ) -> Result<ArchiveMetadata, Error> {
        // If the metadata was provided by the user directly, prefer it.
        if let Some(metadata) = self
            .build_context
            .dependency_metadata()
            .get(dist.name(), Some(dist.version()))
        {
            return Ok(ArchiveMetadata::from_metadata23(metadata));
        }

        let metadata = dist
            .read_metadata()
            .map_err(|err| Error::ReadInstalled(Box::new(dist.clone()), err))?;

        Ok(ArchiveMetadata::from_metadata23(metadata.clone()))
    }

    /// Retrieve distribution metadata and any hashes requested for resolution.
    ///
    /// Source archive hashes are validated before executing build backends; wheel archive hash
    /// validation is deferred to installation.
    #[instrument(skip_all, fields(%dist))]
    pub async fn get_or_build_wheel_metadata(
        &self,
        dist: &Dist,
        hashes: MetadataHashPolicy<'_>,
    ) -> Result<ArchiveMetadata, Error> {
        match dist {
            Dist::Built(built) => self.get_wheel_metadata(built, hashes).await,
            Dist::Source(source) => {
                self.build_wheel_metadata(&BuildableSource::Dist(source), hashes)
                    .await
            }
        }
    }

    /// Fetch a wheel from the cache or download it from the index.
    ///
    /// While hashes will be generated in all cases, hash-checking is _not_ enforced and should
    /// instead be enforced by the caller.
    async fn get_wheel(
        &self,
        dist: &BuiltDist,
        hashes: ArchiveHashPolicy<'_>,
    ) -> Result<LocalWheel, Error> {
        match dist {
            BuiltDist::Registry(wheels) => {
                let wheel = wheels.best_wheel();
                let url = wheel.file.url.to_url()?;
                let size = wheel.file.size;

                // Create a cache entry for the wheel.
                let wheel_entry = self.build_context.cache().entry(
                    CacheBucket::Wheels,
                    WheelCache::Index(&wheel.index).wheel_dir(wheel.name().as_ref()),
                    wheel.filename.cache_key(),
                );

                // If the URL is a file URL, load the wheel directly.
                if url.scheme() == "file" {
                    let path = url
                        .to_file_path()
                        .map_err(|()| Error::NonFileUrl(url.clone()))?;
                    return self
                        .load_wheel(&path, &wheel.filename, wheel_entry, dist, hashes)
                        .await;
                }

                // Download and unzip.
                match self
                    .stream_wheel(
                        url.clone(),
                        dist.index(),
                        &wheel.filename,
                        size,
                        &wheel_entry,
                        dist,
                        hashes,
                    )
                    .await
                {
                    Ok(archive) => Ok(LocalWheel {
                        dist: Dist::Built(dist.clone()),
                        archive: self
                            .build_context
                            .cache()
                            .archive(&archive.id)
                            .into_boxed_path(),
                        hashes: archive.hashes,
                        filename: wheel.filename.clone(),
                        cache: CacheInfo::default(),
                        build: None,
                    }),
                    Err(Error::Extract(name, err)) => {
                        if err.is_http_streaming_unsupported() {
                            warn!(
                                "Streaming unsupported for {dist}; downloading wheel to disk ({err})"
                            );
                        } else if err.is_http_streaming_failed() {
                            warn!("Streaming failed for {dist}; downloading wheel to disk ({err})");
                        } else {
                            return Err(Error::Extract(name, err));
                        }

                        // If the request failed because streaming was unsupported or failed,
                        // download the wheel directly.
                        let archive = self
                            .download_wheel(
                                url,
                                dist.index(),
                                &wheel.filename,
                                size,
                                &wheel_entry,
                                dist,
                                hashes,
                            )
                            .await?;

                        Ok(LocalWheel {
                            dist: Dist::Built(dist.clone()),
                            archive: self
                                .build_context
                                .cache()
                                .archive(&archive.id)
                                .into_boxed_path(),
                            hashes: archive.hashes,
                            filename: wheel.filename.clone(),
                            cache: CacheInfo::default(),
                            build: None,
                        })
                    }
                    Err(err) => Err(err),
                }
            }

            BuiltDist::DirectUrl(wheel) => {
                // Create a cache entry for the wheel.
                let wheel_entry = self.build_context.cache().entry(
                    CacheBucket::Wheels,
                    WheelCache::Url(&wheel.url).wheel_dir(wheel.name().as_ref()),
                    wheel.filename.cache_key(),
                );

                // Download and unzip.
                match self
                    .stream_wheel(
                        wheel.url.raw().clone(),
                        None,
                        &wheel.filename,
                        wheel.size,
                        &wheel_entry,
                        dist,
                        hashes,
                    )
                    .await
                {
                    Ok(archive) => Ok(LocalWheel {
                        dist: Dist::Built(dist.clone()),
                        archive: self
                            .build_context
                            .cache()
                            .archive(&archive.id)
                            .into_boxed_path(),
                        hashes: archive.hashes,
                        filename: wheel.filename.clone(),
                        cache: CacheInfo::default(),
                        build: None,
                    }),
                    Err(Error::Extract(name, err)) => {
                        if err.is_http_streaming_unsupported() {
                            warn!(
                                "Streaming unsupported for {dist}; downloading wheel to disk ({err})"
                            );
                        } else if err.is_http_streaming_failed() {
                            warn!("Streaming failed for {dist}; downloading wheel to disk ({err})");
                        } else {
                            return Err(Error::Extract(name, err));
                        }

                        // If the request failed because streaming was unsupported or failed,
                        // download the wheel directly.
                        let archive = self
                            .download_wheel(
                                wheel.url.raw().clone(),
                                None,
                                &wheel.filename,
                                wheel.size,
                                &wheel_entry,
                                dist,
                                hashes,
                            )
                            .await?;
                        Ok(LocalWheel {
                            dist: Dist::Built(dist.clone()),
                            archive: self
                                .build_context
                                .cache()
                                .archive(&archive.id)
                                .into_boxed_path(),
                            hashes: archive.hashes,
                            filename: wheel.filename.clone(),
                            cache: CacheInfo::default(),
                            build: None,
                        })
                    }
                    Err(err) => Err(err),
                }
            }

            BuiltDist::GitPath(wheel) => {
                // Fetch the Git repository.
                let fetch = self
                    .build_context
                    .git()
                    .fetch(
                        &wheel.git,
                        self.client.unmanaged.git_http_settings(wheel.git.url()),
                        self.build_context.cache().bucket(CacheBucket::Git),
                        self.reporter.clone().map(<dyn Reporter>::into_git_reporter),
                    )
                    .await?;

                if wheel.git.lfs().enabled() && !fetch.lfs_ready() {
                    if GIT_LFS.is_err() {
                        return Err(Error::MissingWheelGitLfsArtifacts(
                            wheel.url.to_url(),
                            GitError::GitLfsNotFound,
                        ));
                    }
                    return Err(Error::MissingWheelGitLfsArtifacts(
                        wheel.url.to_url(),
                        GitError::GitLfsNotConfigured,
                    ));
                }

                let git_sha = fetch.git().precise().expect("Exact commit after checkout");
                let cache_entry = self.build_context.cache().entry(
                    CacheBucket::Wheels,
                    WheelCache::Git(&wheel.url, git_sha.as_short_str()).root(),
                    wheel.filename.stem(),
                );

                let install_path = fetch.path().join(&wheel.install_path);

                self.load_wheel(&install_path, &wheel.filename, cache_entry, dist, hashes)
                    .await
            }

            BuiltDist::Path(wheel) => {
                let cache_entry = self.build_context.cache().entry(
                    CacheBucket::Wheels,
                    WheelCache::Url(&wheel.url).wheel_dir(wheel.name().as_ref()),
                    wheel.filename.cache_key(),
                );

                self.load_wheel(
                    &wheel.install_path,
                    &wheel.filename,
                    cache_entry,
                    dist,
                    hashes,
                )
                .await
            }
        }
    }

    /// Convert a source distribution into a wheel, fetching it from the cache or building it if
    /// necessary.
    ///
    /// The returned wheel is guaranteed to come from a distribution with a matching hash, and
    /// no build processes will be executed for distributions with mismatched hashes.
    async fn build_wheel(
        &self,
        dist: &SourceDist,
        tags: &Tags,
        hashes: ArchiveHashPolicy<'_>,
    ) -> Result<LocalWheel, Error> {
        let built_wheel = self
            .builder
            .download_and_build(&BuildableSource::Dist(dist), tags, hashes, &self.client)
            .boxed_local()
            .await?;

        // Check that the wheel is compatible with its install target.
        //
        // When building a build dependency for a cross-install, the build dependency needs
        // to install and run on the host instead of the target. In this case the `tags` are already
        // for the host instead of the target, so this check passes.
        if !built_wheel.filename.is_compatible(tags) {
            return if tags.is_cross() {
                Err(Error::BuiltWheelIncompatibleTargetPlatform {
                    filename: built_wheel.filename,
                    python_platform: tags.python_platform().clone(),
                    python_version: PythonVersion {
                        version: tags.python_version(),
                        variant: if tags.is_freethreaded() {
                            PythonVariant::Freethreaded
                        } else {
                            PythonVariant::Default
                        },
                    },
                })
            } else {
                Err(Error::BuiltWheelIncompatibleHostPlatform {
                    filename: built_wheel.filename,
                    python_platform: tags.python_platform().clone(),
                    python_version: PythonVersion {
                        version: tags.python_version(),
                        variant: if tags.is_freethreaded() {
                            PythonVariant::Freethreaded
                        } else {
                            PythonVariant::Default
                        },
                    },
                })
            };
        }

        // Acquire the advisory lock.
        let wheel_entry = CacheEntry::from_path(built_wheel.target.as_ref());
        let _lock = Self::lock_wheel(&wheel_entry, &built_wheel.filename).await?;

        // If the wheel was unzipped previously, respect it. Source distributions are
        // cached under a unique revision ID, so unzipped directories are never stale.
        match self.build_context.cache().resolve_link(&built_wheel.target) {
            Ok(archive) => {
                return Ok(LocalWheel {
                    dist: Dist::Source(dist.clone()),
                    archive: archive.into_boxed_path(),
                    filename: built_wheel.filename,
                    hashes: built_wheel.hashes,
                    cache: built_wheel.cache_info,
                    build: Some(built_wheel.build_info),
                });
            }
            Err(err) if err.kind() == io::ErrorKind::NotFound => {}
            Err(err) => return Err(Error::CacheRead(err)),
        }

        // Otherwise, unzip the wheel.
        let id = self
            .unzip_wheel(
                &built_wheel.path,
                &built_wheel.target,
                DistRef::Source(dist),
            )
            .await?;

        Ok(LocalWheel {
            dist: Dist::Source(dist.clone()),
            archive: self.build_context.cache().archive(&id).into_boxed_path(),
            hashes: built_wheel.hashes,
            filename: built_wheel.filename,
            cache: built_wheel.cache_info,
            build: Some(built_wheel.build_info),
        })
    }

    /// Fetch wheel metadata, along with any hashes requested for resolution.
    ///
    /// Wheel archive hash validation is deferred to installation. Metadata sidecar hashes are
    /// checked by the client when downloading sidecars with index-provided hashes.
    async fn get_wheel_metadata(
        &self,
        dist: &BuiltDist,
        hashes: MetadataHashPolicy<'_>,
    ) -> Result<ArchiveMetadata, Error> {
        let hash_policy = match hashes.validation {
            HashValidation::None => {
                let compute_hashes = match hashes.collection {
                    HashCollection::None => false,
                    HashCollection::Url | HashCollection::All => match dist {
                        BuiltDist::Registry(dist) => {
                            // Generate missing hashes for indexes and `--find-links` without hashes.
                            // This hashes only the selected wheel, not every distribution for the version.
                            hashes.collection == HashCollection::All
                                && dist.best_wheel().file.hashes.is_empty()
                        }
                        BuiltDist::DirectUrl(dist) => parse_url_hashes(&dist.url).is_none(),
                        BuiltDist::Path(_) | BuiltDist::GitPath(_) => true,
                    },
                };
                if compute_hashes {
                    ArchiveHashPolicy::Generate
                } else {
                    ArchiveHashPolicy::None
                }
            }
            HashValidation::Any(_) | HashValidation::All(_) => hashes.validation.into(),
        };

        // Fetch the entire wheel only when we need to compute a hash for resolution.
        // TODO(charlie): Request the hashes via a separate method, to reduce the coupling in this API.
        if hash_policy == ArchiveHashPolicy::Generate {
            let wheel = self.get_wheel(dist, hash_policy).await?;
            // If the metadata was provided by the user directly, prefer it.
            let metadata = if let Some(metadata) = self
                .build_context
                .dependency_metadata()
                .get(dist.name(), Some(dist.version()))
            {
                metadata
            } else {
                wheel.metadata()?
            };
            let hashes = wheel.hashes;
            return Ok(ArchiveMetadata {
                metadata: Metadata::from_metadata23(metadata),
                hashes,
            });
        }

        // If the metadata was provided by the user directly, prefer it.
        if let Some(metadata) = self
            .build_context
            .dependency_metadata()
            .get(dist.name(), Some(dist.version()))
        {
            return Ok(ArchiveMetadata::from_metadata23(metadata));
        }

        let result = self
            .client
            .managed(|client| {
                client
                    .wheel_metadata(
                        dist,
                        self.build_context.git(),
                        self.build_context.capabilities(),
                        self.reporter.clone().map(<dyn Reporter>::into_git_reporter),
                    )
                    .boxed_local()
            })
            .await;

        match result {
            Ok(metadata) => {
                // Validate that the metadata is consistent with the distribution.
                Ok(ArchiveMetadata::from_metadata23(metadata))
            }
            Err(err) if err.is_http_streaming_unsupported() => {
                warn!(
                    "Streaming unsupported when fetching metadata for {dist}; downloading wheel directly ({err})"
                );

                // If the request failed due to an error that could be resolved by
                // downloading the wheel directly, try that.
                let wheel = self.get_wheel(dist, hash_policy).await?;
                let metadata = wheel.metadata()?;
                let hashes = wheel.hashes;
                Ok(ArchiveMetadata {
                    metadata: Metadata::from_metadata23(metadata),
                    hashes,
                })
            }
            Err(err) => Err(err.into()),
        }
    }

    /// Build the wheel metadata for a source distribution, or fetch it from the cache if possible.
    ///
    /// Requested hashes are validated before executing a build backend. User-provided metadata
    /// can avoid downloading or validating the source archive.
    pub async fn build_wheel_metadata(
        &self,
        source: &BuildableSource<'_>,
        hashes: MetadataHashPolicy<'_>,
    ) -> Result<ArchiveMetadata, Error> {
        // If the metadata was provided by the user directly, prefer it.
        if let Some(dist) = source.as_dist() {
            if let Some(metadata) = self
                .build_context
                .dependency_metadata()
                .get(dist.name(), dist.version())
            {
                // If we skipped the build, we should still resolve any Git dependencies to precise
                // commits.
                self.builder.resolve_revision(source, &self.client).await?;

                return Ok(ArchiveMetadata::from_metadata23(metadata));
            }
        }

        let url_hashes = if let BuildableSource::Dist(SourceDist::DirectUrl(dist)) = source {
            parse_url_hashes(&dist.url)
        } else if let BuildableSource::Url(SourceUrl::Direct(url)) = source {
            parse_url_hashes(url.url)
        } else {
            None
        };

        let build_hash_policy = match hashes.validation {
            HashValidation::None => match hashes.collection {
                HashCollection::None => ArchiveHashPolicy::None,
                // If resolving metadata requires a build, validate any URL hash before executing
                // the backend, even when the caller only requested hash collection.
                HashCollection::Url | HashCollection::All => match url_hashes.as_ref() {
                    Some(digests) => ArchiveHashPolicy::All(digests.as_slice()),
                    None => ArchiveHashPolicy::Generate,
                },
            },
            HashValidation::Any(_) | HashValidation::All(_) => hashes.validation.into(),
        };
        let metadata = self
            .builder
            .download_and_build_metadata(source, build_hash_policy, &self.client)
            .boxed_local()
            .await?;

        Ok(metadata)
    }

    /// Return the [`RequiresDist`] from a `pyproject.toml`, if it can be statically extracted.
    pub async fn requires_dist(
        &self,
        path: &Path,
        pyproject_toml: &PyProjectToml,
    ) -> Result<Option<RequiresDist>, Error> {
        self.builder
            .source_tree_requires_dist(
                path,
                pyproject_toml,
                self.client.unmanaged.credentials_cache(),
            )
            .await
    }

    /// Stream a wheel from a URL, unzipping it into the cache as it's downloaded.
    ///
    /// Note that `progress_size_hint` is a size hint for the progress reporter,
    /// and is not guaranteed to be the true size of the wheel if/when fetched from the
    /// origin.
    async fn stream_wheel(
        &self,
        url: DisplaySafeUrl,
        index: Option<&IndexUrl>,
        filename: &WheelFilename,
        progress_size_hint: Option<u64>,
        wheel_entry: &CacheEntry,
        dist: &BuiltDist,
        hashes: ArchiveHashPolicy<'_>,
    ) -> Result<Archive, Error> {
        // Sometimes we can promote the size hint to a trusted effective size.
        let expected_size = match dist {
            BuiltDist::Registry(dist) if dist.best_wheel().size_is_authoritative => {
                progress_size_hint
            }
            BuiltDist::DirectUrl(_) => progress_size_hint,
            _ => None,
        };

        // Acquire an advisory lock, to guard against concurrent writes.
        let _lock = Self::lock_wheel(wheel_entry, filename).await?;

        // Create an entry for the HTTP cache.
        let http_entry = wheel_entry.with_file(format!("{}.http", filename.cache_key()));

        let download = |response: reqwest::Response, _: &mut RetryState| {
            async {
                let progress_size_hint = progress_size_hint.or_else(|| content_length(&response));

                let progress = self.reporter.as_ref().map(|reporter| {
                    (
                        reporter,
                        reporter.on_download_start(dist.name(), progress_size_hint),
                    )
                });

                let reader = response
                    .bytes_stream()
                    .map_err(|err| self.handle_response_errors(err))
                    .into_async_read();

                // Create a hasher for each hash algorithm.
                let algorithms = http_hash_algorithms(hashes);
                let mut hashers = algorithms.into_iter().map(Hasher::from).collect::<Vec<_>>();
                let mut hasher = uv_extract::hash::HashReader::new(reader.compat(), &mut hashers);

                // Download and unzip the wheel to a temporary directory.
                let extractor = WheelExtractor::new(
                    self.build_context.cache().root(),
                    self.content_addressed_cache,
                )
                .map_err(Error::CacheWrite)?;

                let mut extracted = match progress {
                    Some((reporter, progress)) => {
                        let mut reader = ProgressReader::new(&mut hasher, progress, &**reporter);
                        extractor
                            .extract_streaming(&mut reader)
                            .await
                            .map_err(|err| Error::Extract(filename.to_string(), err))?
                    }
                    None => extractor
                        .extract_streaming(&mut hasher)
                        .await
                        .map_err(|err| Error::Extract(filename.to_string(), err))?,
                };
                // Exhaust the reader to compute the hashes.
                hasher.finish().await.map_err(Error::HashExhaustion)?;
                let actual_size = hasher.bytes_read();
                if let Some(expected) = expected_size
                    && actual_size != expected
                {
                    return Err(Error::MismatchedSize {
                        distribution: dist.to_string(),
                        expected,
                        actual: actual_size,
                    });
                }

                // Before we make the wheel accessible by persisting it, ensure that the RECORD is
                // valid.
                extracted.validate_and_heal_record(dist)?;

                // Persist the temporary directory to the directory store.
                let id = self
                    .persist_extracted_wheel(extracted, wheel_entry.path())
                    .await?;

                if let Some((reporter, progress)) = progress {
                    reporter.on_download_complete(dist.name(), progress);
                }

                Ok(Archive::new(
                    id,
                    hashers.into_iter().map(HashDigest::from).collect(),
                    filename.clone(),
                    Some(actual_size),
                ))
            }
            .instrument(info_span!("wheel", wheel = %dist))
        };

        // Fetch the archive from the cache, or download it if necessary.
        let req = self.request(url.clone())?;

        // Determine the cache control policy for the URL.
        let cache_control = match self.client.unmanaged.connectivity() {
            Connectivity::Online
                if let Some(header) = index.and_then(|index| {
                    self.build_context
                        .locations()
                        .artifact_cache_control_for(index)
                }) =>
            {
                CacheControl::Override(header)
            }
            Connectivity::Online => CacheControl::from(
                self.build_context
                    .cache()
                    .freshness(&http_entry, Some(&filename.name), None)
                    .map_err(Error::CacheRead)?,
            ),
            Connectivity::Offline => CacheControl::AllowStale,
        };

        let archive = self
            .client
            .managed(|client| {
                client.cached_client().get_serde_with_retry(
                    req,
                    &http_entry,
                    cache_control.clone(),
                    download,
                )
            })
            .await
            .map_err(|err| match err {
                CachedClientError::Callback { err, .. } => err,
                CachedClientError::Client(err) => Error::Client(err),
            })?;

        if let (Some(expected), Some(actual)) = (expected_size, archive.size)
            && expected != actual
        {
            return Err(Error::MismatchedSize {
                distribution: dist.to_string(),
                expected,
                actual,
            });
        }

        // If the archive is missing the required hashes or size, or has since been removed, force a refresh.
        let archive = Some(archive)
            .filter(|archive| archive.has_digests(hashes))
            .filter(|archive| archive.exists(self.build_context.cache()))
            .filter(|archive| expected_size.is_none() || archive.size.is_some());

        let archive = if let Some(archive) = archive {
            archive
        } else {
            self.client
                .managed(async |client| {
                    client
                        .cached_client()
                        .skip_cache_with_retry(
                            self.request(url)?,
                            &http_entry,
                            cache_control,
                            download,
                        )
                        .await
                        .map_err(|err| match err {
                            CachedClientError::Callback { err, .. } => err,
                            CachedClientError::Client(err) => Error::Client(err),
                        })
                })
                .await?
        };

        Ok(archive)
    }

    /// Download a wheel from a URL, then unzip it into the cache.
    ///
    /// Note that, like [`Self::stream_wheel`], `progress_size_hint` is
    /// a size hint for the progress reporter, and is not guaranteed to be the size
    /// of the wheel if/when fetched from the origin.
    async fn download_wheel(
        &self,
        url: DisplaySafeUrl,
        index: Option<&IndexUrl>,
        filename: &WheelFilename,
        progress_size_hint: Option<u64>,
        wheel_entry: &CacheEntry,
        dist: &BuiltDist,
        hashes: ArchiveHashPolicy<'_>,
    ) -> Result<Archive, Error> {
        // Sometimes we can promote the size hint to a trusted effective size.
        let expected_size = match dist {
            BuiltDist::Registry(dist) if dist.best_wheel().size_is_authoritative => {
                progress_size_hint
            }
            BuiltDist::DirectUrl(_) => progress_size_hint,
            _ => None,
        };

        // Acquire an advisory lock, to guard against concurrent writes.
        let _lock = Self::lock_wheel(wheel_entry, filename).await?;

        // Create an entry for the HTTP cache.
        let http_entry = wheel_entry.with_file(format!("{}.http", filename.cache_key()));

        let download_url = url.clone();

        let download = async |response, retry_state: &mut RetryState| {
            self.download_wheel_response(
                response,
                &download_url,
                retry_state,
                filename,
                progress_size_hint,
                expected_size,
                wheel_entry,
                dist,
                hashes,
            )
            .instrument(info_span!("wheel", wheel = %dist))
            .await
        };

        // Fetch the archive from the cache, or download it if necessary.
        let req = self.request(url.clone())?;

        // Determine the cache control policy for the URL.
        let cache_control = match self.client.unmanaged.connectivity() {
            Connectivity::Online
                if let Some(header) = index.and_then(|index| {
                    self.build_context
                        .locations()
                        .artifact_cache_control_for(index)
                }) =>
            {
                CacheControl::Override(header)
            }
            Connectivity::Online => CacheControl::from(
                self.build_context
                    .cache()
                    .freshness(&http_entry, Some(&filename.name), None)
                    .map_err(Error::CacheRead)?,
            ),
            Connectivity::Offline => CacheControl::AllowStale,
        };

        let archive = self
            .client
            .managed(|client| {
                client.cached_client().get_serde_with_retry(
                    req,
                    &http_entry,
                    cache_control.clone(),
                    download,
                )
            })
            .await
            .map_err(|err| match err {
                CachedClientError::Callback { err, .. } => err,
                CachedClientError::Client(err) => Error::Client(err),
            })?;

        if let (Some(expected), Some(actual)) = (expected_size, archive.size)
            && expected != actual
        {
            return Err(Error::MismatchedSize {
                distribution: dist.to_string(),
                expected,
                actual,
            });
        }

        // If the archive is missing the required hashes or size, or has since been removed, force a refresh.
        let archive = Some(archive)
            .filter(|archive| archive.has_digests(hashes))
            .filter(|archive| archive.exists(self.build_context.cache()))
            .filter(|archive| expected_size.is_none() || archive.size.is_some());

        let archive = if let Some(archive) = archive {
            archive
        } else {
            self.client
                .managed(async |client| {
                    client
                        .cached_client()
                        .skip_cache_with_retry(
                            self.request(url)?,
                            &http_entry,
                            cache_control,
                            download,
                        )
                        .await
                        .map_err(|err| match err {
                            CachedClientError::Callback { err, .. } => err,
                            CachedClientError::Client(err) => Error::Client(err),
                        })
                })
                .await?
        };

        Ok(archive)
    }

    /// Download and extract a wheel from an HTTP response.
    ///
    /// This helper is called by [`Self::download_wheel`] and handles partial content/resumptions
    /// of interrupted downloads if the origin supports it. Hard failures are propagated
    /// back to the caller and thus to our general retry machinery.
    ///
    /// The caller is responsible for obtaining a lock on the wheel cache.
    async fn download_wheel_response(
        &self,
        mut response: reqwest::Response,
        url: &DisplaySafeUrl,
        retry_state: &mut RetryState,
        filename: &WheelFilename,
        progress_size_hint: Option<u64>,
        expected_size: Option<u64>,
        wheel_entry: &CacheEntry,
        dist: &BuiltDist,
        hashes: ArchiveHashPolicy<'_>,
    ) -> Result<Archive, Error> {
        let progress_size_hint = progress_size_hint.or_else(|| content_length(&response));
        let mut download_size = content_length(&response).or(expected_size);

        let progress = self.reporter.as_ref().map(|reporter| {
            (
                reporter,
                reporter.on_download_start(dist.name(), progress_size_hint),
            )
        });

        let algorithms = http_hash_algorithms(hashes);

        // Download the wheel to a temporary file.
        let temp_file =
            tempfile::tempfile_in(self.build_context.cache().root()).map_err(Error::CacheWrite)?;
        let mut writer = tokio::io::BufWriter::new(fs_err::tokio::File::from_std(
            // It's an unnamed file on Linux so that's the best approximation.
            fs_err::File::from_parts(temp_file, self.build_context.cache().root()),
        ));

        // States for the download loop below.
        let mut hashers = algorithms.into_iter().map(Hasher::from).collect::<Vec<_>>();
        // The total number of bytes retrieved, accumulated over individual requests.
        // Note that this is *not* the same as the number of bytes actually written
        // to the temporary file, since a copy from the response to the file can fail.
        let mut bytes_retrieved = 0;
        // The most recent range request's starting offset.
        let mut resumed_at = None;
        // The `Content-Range` that the download was last resumed at.
        let mut resumed_range: Option<ContentRangeBytes> = None;

        // This loop is where we handle resumption of interrupted downloads, as well as
        // range requests (if the origin supports them).
        //
        // Errors returned from this loop reach the outer retry classifier, which may restart
        // the full download.
        loop {
            // Reject conflicting full-response lengths before reading the body. A range response's
            // Content-Length describes only that range, so it cannot be compared to the wheel size.
            if response.status() == reqwest::StatusCode::OK
                && let (Some(expected), Some(actual)) = (expected_size, content_length(&response))
                && expected != actual
            {
                return Err(Error::MismatchedContentLength {
                    distribution: dist.to_string(),
                    expected,
                    actual,
                });
            }

            // Check whether the response indicates range request support. A `206 Partial Content`
            // implies range support while an `Accept-Ranges: bytes` header explicitly advertises it.
            let supports_range_requests = response.status() == reqwest::StatusCode::PARTIAL_CONTENT
                || response
                    .headers()
                    .get(reqwest::header::ACCEPT_RANGES)
                    .is_some_and(|value| value == "bytes");

            // A server can advertise range requests but ignore one. In that case, the
            // response is a complete download and must replace the partial bytes.
            let replaces_partial_download =
                resumed_at.is_some() && response.status() != reqwest::StatusCode::PARTIAL_CONTENT;
            if replaces_partial_download {
                writer
                    .get_mut()
                    .set_len(0)
                    .await
                    .map_err(Error::CacheWrite)?;
                writer
                    .seek(io::SeekFrom::Start(0))
                    .await
                    .map_err(Error::CacheWrite)?;
                hashers = http_hash_algorithms(hashes)
                    .into_iter()
                    .map(Hasher::from)
                    .collect();
                bytes_retrieved = 0;
            }

            let reader = response
                .bytes_stream()
                .map_err(|err| self.handle_response_errors(err))
                .into_async_read();
            let mut hasher = uv_extract::hash::HashReader::new(reader.compat(), &mut hashers);

            // Drain the response. This could be a partial response or a full one.
            // Note that the partial response here can take several forms: it can be an interrupted
            // request *or* it can be a `206 Partial Content`.
            let copy_result = match progress {
                Some((reporter, progress)) => {
                    // Wrap the reader in a progress reporter. This will report 100%
                    // progress once the download is complete, before the wheel is unzipped.
                    let mut reader = ProgressReader::new(&mut hasher, progress, &**reporter);

                    tokio::io::copy(&mut reader, &mut writer)
                        .await
                        .map_err(Error::CacheWrite)
                }
                None => tokio::io::copy(&mut hasher, &mut writer)
                    .await
                    .map_err(Error::CacheWrite),
            };

            bytes_retrieved += hasher.bytes_read();

            let interrupted = copy_result.is_err();
            let err = match copy_result {
                // Draining the response succeeded. However, the response could be a `206 Partial Content`,
                // so we can't assume that we're done.
                Ok(_) => {
                    // No `resumed_range` means this was a normal full response, so there's
                    // no resumption to do. We can leave the loop.
                    let Some(range) = resumed_range else {
                        break;
                    };

                    // We should be in sync with the origin. Failure here means that the
                    // origin actually sent us an under- or over-length response. Treat this
                    // as a non-retryable error, since the HTTP body was received successfully.
                    //
                    // Byte ranges are inclusive, not exclusive.
                    if bytes_retrieved != range.last_byte + 1 {
                        return Err(Error::MismatchedRangeSize {
                            distribution: dist.to_string(),
                            expected: range.last_byte - range.first_byte + 1,
                            actual: hasher.bytes_read(),
                        });
                    }

                    // We've successfully performed the download over one or more
                    // range requests. We can leave the loop.
                    if bytes_retrieved == range.complete_length {
                        break;
                    }

                    // A successful range response may cover only part of the requested
                    // bytes. Keep requesting the remainder before extracting the wheel.
                    //
                    // Observe that we don't return this error; we bind it for handling
                    // below.
                    Error::CacheWrite(io::Error::new(
                        io::ErrorKind::UnexpectedEof,
                        "Range response did not complete the download",
                    ))
                }
                // Draining the response failed, potentially because the connection was interrupted.
                // Like above, we don't return this error until we know how we want to treat it.
                Err(err) => err,
            };

            // At this point we have *some* error state: either a synthetic error (from a
            // partial response that did not complete the download) or a real error (e.g.
            // from an interrupted download). We can only resume if certain conditions
            // below are met.

            // If the response replaces an already in-flight partial download or
            // indicates a lack of range-request support, then we can't resume.
            // Return the error back to the retry stack.
            if replaces_partial_download || !supports_range_requests {
                return Err(err);
            }

            // Some sanity checks: we should have made *some* progress as part of
            // partial response, plus our writer's offset should agree with the number
            // of bytes retrieved, *and* we should be making forward progress (i.e. not
            // resuming behind where we already are).
            writer.flush().await.map_err(Error::CacheWrite)?;
            let offset = writer
                .get_mut()
                .stream_position()
                .await
                .map_err(Error::CacheWrite)?;
            if offset == 0
                || offset != bytes_retrieved
                || resumed_at.is_some_and(|previous| offset <= previous)
            {
                return Err(err);
            }

            // Recovering from a failed body consumes the same budget as a full restart.
            // A successfully completed range needs no retry, even if more bytes remain.
            if interrupted {
                let Some(backoff) = retry_state.should_retry(&err, 0) else {
                    return Err(err);
                };
                retry_state.sleep_backoff(backoff).await;
            }

            // Finally our resumption, which is a range request.
            debug!("Resuming download of {url} at byte {offset}");
            let resumed_response = retry_state
                .send(self.request_with_offset(url.clone(), offset))
                .await?;

            // A chunked response can fail after all wheel bytes arrive, leaving no satisfiable
            // range. Return the original error so the outer retry policy can restart in full.
            if resumed_response.status() == reqwest::StatusCode::RANGE_NOT_SATISFIABLE {
                debug!("Range not satisfiable while resuming {url}; abandoning resumed download");
                return Err(err);
            }
            resumed_response.error_for_status_ref()?;

            resumed_range = if resumed_response.status() == reqwest::StatusCode::PARTIAL_CONTENT {
                // The origin honored our range request, so we update `resumed_range`
                // for the next iteration.
                let Some(range) = content_range(&resumed_response, offset, download_size) else {
                    warn!(
                        "Invalid range request response from server that declares HTTP range \
                         request support, abandoning resumed download: {url}"
                    );
                    return Err(err);
                };
                download_size = Some(range.complete_length);
                Some(range)
            } else {
                // The origin is allowed to ignore our range request and send a full response instead.
                // That means we have no `resumed_range` to honor on the next iteration.
                None
            };

            response = resumed_response;
            resumed_at = Some(offset);
        }

        // We've left the resumption loop.
        // Sanity check: we should have written as many bytes as we expected.
        if let Some(expected) = expected_size
            && bytes_retrieved != expected
        {
            return Err(Error::MismatchedSize {
                distribution: dist.to_string(),
                expected,
                actual: bytes_retrieved,
            });
        }

        // Unzip the wheel to a temporary directory.
        let extractor = WheelExtractor::new(
            self.build_context.cache().root(),
            self.content_addressed_cache,
        )
        .map_err(Error::CacheWrite)?;
        let mut file = writer.into_inner();
        file.seek(io::SeekFrom::Start(0))
            .await
            .map_err(Error::CacheWrite)?;

        let file = file.into_std().await;
        let mut extracted = tokio::task::spawn_blocking(move || extractor.extract_seekable(file))
            .await?
            .map_err(|err| Error::Extract(filename.to_string(), err))?;
        let hashes = hashers.into_iter().map(HashDigest::from).collect();

        // Before we make the wheel accessible by persisting it, ensure that the RECORD is
        // valid.
        extracted.validate_and_heal_record(dist)?;

        // Persist the temporary directory to the directory store.
        let id = self
            .persist_extracted_wheel(extracted, wheel_entry.path())
            .await?;

        if let Some((reporter, progress)) = progress {
            reporter.on_download_complete(dist.name(), progress);
        }

        Ok(Archive::new(
            id,
            hashes,
            filename.clone(),
            Some(bytes_retrieved),
        ))
    }

    /// Load a wheel from a local path.
    async fn load_wheel(
        &self,
        path: &Path,
        filename: &WheelFilename,
        wheel_entry: CacheEntry,
        dist: &BuiltDist,
        hashes: ArchiveHashPolicy<'_>,
    ) -> Result<LocalWheel, Error> {
        // Acquire an advisory lock, to guard against concurrent writes.
        let _lock = Self::lock_wheel(&wheel_entry, filename).await?;

        // Determine the last-modified time of the wheel.
        let modified = Timestamp::from_path(path).map_err(Error::CacheRead)?;

        // Attempt to read the archive pointer from the cache.
        let pointer_entry = wheel_entry.with_file(format!("{}.rev", filename.cache_key()));
        let pointer = PathArchivePointer::read_from(&pointer_entry)?;

        // Extract the archive from the pointer.
        let archive = pointer
            .filter(|pointer| pointer.is_up_to_date(modified))
            .map(PathArchivePointer::into_archive)
            .filter(|archive| archive.has_digests(hashes));

        // If the file is already unzipped, and the cache is up-to-date, return it.
        if let Some(archive) = archive {
            Ok(LocalWheel {
                dist: Dist::Built(dist.clone()),
                archive: self
                    .build_context
                    .cache()
                    .archive(&archive.id)
                    .into_boxed_path(),
                hashes: archive.hashes,
                filename: filename.clone(),
                cache: CacheInfo::from_timestamp(modified),
                build: None,
            })
        } else if hashes.is_none() {
            // Otherwise, unzip the wheel.
            let archive = Archive::new(
                self.unzip_wheel(path, wheel_entry.path(), DistRef::Built(dist))
                    .await?,
                HashDigests::empty(),
                filename.clone(),
                None,
            );

            // Write the archive pointer to the cache.
            let pointer = PathArchivePointer {
                timestamp: modified,
                archive: archive.clone(),
            };
            pointer.write_to(&pointer_entry).await?;

            Ok(LocalWheel {
                dist: Dist::Built(dist.clone()),
                archive: self
                    .build_context
                    .cache()
                    .archive(&archive.id)
                    .into_boxed_path(),
                hashes: archive.hashes,
                filename: filename.clone(),
                cache: CacheInfo::from_timestamp(modified),
                build: None,
            })
        } else {
            // If necessary, compute the hashes of the wheel.
            let file = fs_err::tokio::File::open(path)
                .await
                .map_err(Error::CacheRead)?;
            let extractor = WheelExtractor::new(
                self.build_context.cache().root(),
                self.content_addressed_cache,
            )
            .map_err(Error::CacheWrite)?;

            // Create a hasher for each hash algorithm.
            let algorithms = hashes.algorithms();
            let mut hashers = algorithms.into_iter().map(Hasher::from).collect::<Vec<_>>();
            let mut hasher = uv_extract::hash::HashReader::new(file, &mut hashers);

            // Unzip the wheel to a temporary directory.
            let mut extracted = extractor
                .extract_streaming(&mut hasher)
                .await
                .map_err(|err| Error::Extract(filename.to_string(), err))?;

            // Exhaust the reader to compute the hash.
            hasher.finish().await.map_err(Error::HashExhaustion)?;

            let hashes = hashers.into_iter().map(HashDigest::from).collect();

            // Before we make the wheel accessible by persisting it, ensure that the RECORD is
            // valid.
            extracted.validate_and_heal_record(dist)?;

            // Persist the temporary directory to the directory store.
            let id = self
                .persist_extracted_wheel(extracted, wheel_entry.path())
                .await?;

            // Create an archive.
            let archive = Archive::new(id, hashes, filename.clone(), None);

            // Write the archive pointer to the cache.
            let pointer = PathArchivePointer {
                timestamp: modified,
                archive: archive.clone(),
            };
            pointer.write_to(&pointer_entry).await?;

            Ok(LocalWheel {
                dist: Dist::Built(dist.clone()),
                archive: self
                    .build_context
                    .cache()
                    .archive(&archive.id)
                    .into_boxed_path(),
                hashes: archive.hashes,
                filename: filename.clone(),
                cache: CacheInfo::from_timestamp(modified),
                build: None,
            })
        }
    }

    /// Unzip a wheel into the cache, returning the path to the unzipped directory.
    async fn unzip_wheel(
        &self,
        path: &Path,
        target: &Path,
        dist: DistRef<'_>,
    ) -> Result<ArchiveId, Error> {
        let content_addressed_cache = self.content_addressed_cache;

        let mut extracted = tokio::task::spawn_blocking({
            let path = path.to_owned();
            let root = self.build_context.cache().root().to_path_buf();
            move || -> Result<_, Error> {
                // Unzip the wheel into a temporary directory.
                let extractor = WheelExtractor::new(&root, content_addressed_cache)
                    .map_err(Error::CacheWrite)?;
                let reader = fs_err::File::open(&path).map_err(Error::CacheWrite)?;
                extractor
                    .extract_seekable(reader)
                    .map_err(|err| Error::Extract(path.to_string_lossy().into_owned(), err))
            }
        })
        .await??;

        // Before we make the wheel accessible by persisting it, ensure that the RECORD is valid.
        extracted.validate_and_heal_record(dist)?;

        // Persist the temporary directory to the directory store.
        let id = self.persist_extracted_wheel(extracted, target).await?;

        Ok(id)
    }

    /// Persist an extracted wheel into the archive store.
    ///
    /// A hash tree makes identical extracted trees converge on one archive entry. Without one,
    /// persistence retains the existing behavior of assigning a unique archive ID.
    async fn persist_extracted_wheel(
        &self,
        extracted: ExtractedWheel,
        target: &Path,
    ) -> Result<ArchiveId, Error> {
        let (temp_dir, hashed_wheel) = extracted.into_parts();
        let cache = self.build_context.cache();
        let (temp_dir, id) = if let Some(HashedWheel { files, tree }) = hashed_wheel {
            let digest = DirectoryDigest::from(tree.hash());
            let id = ArchiveId::from_digest(digest.into());
            let cache = cache.clone();
            let temp_dir = tokio::task::spawn_blocking(move || {
                persist_archive_files(&cache, temp_dir.path(), &files)
                    .map_err(Error::CacheWrite)?;
                Ok::<_, Error>(temp_dir)
            })
            .await??;
            (temp_dir, id)
        } else {
            (temp_dir, ArchiveId::default())
        };

        cache
            .persist_with_id(temp_dir, target, id)
            .await
            .map_err(Error::CacheWrite)
    }

    /// Returns a GET [`reqwest::Request`] for the given URL.
    fn request(&self, url: DisplaySafeUrl) -> Result<reqwest::Request, reqwest::Error> {
        self.client
            .unmanaged
            .uncached_client(&url)
            .get(Url::from(url))
            .header(
                // `reqwest` defaults to accepting compressed responses.
                // Specify identity encoding to get consistent .whl downloading
                // behavior from servers. ref: https://github.com/pypa/pip/pull/1688
                "accept-encoding",
                reqwest::header::HeaderValue::from_static("identity"),
            )
            .build()
    }

    /// Build a GET request with a `Range: bytes=<offset>-` header.
    ///
    /// Used to resume an interrupted download from `offset` bytes into the file.
    fn request_with_offset(&self, url: DisplaySafeUrl, offset: u64) -> RequestBuilder<'_> {
        self.client
            .unmanaged
            .uncached_client(&url)
            .get(Url::from(url))
            .header(
                "accept-encoding",
                reqwest::header::HeaderValue::from_static("identity"),
            )
            .header(reqwest::header::RANGE, format!("bytes={offset}-"))
    }

    /// Return the [`ManagedClient`] used by this resolver.
    pub fn client(&self) -> &ManagedClient<'a> {
        &self.client
    }
}

/// Share extracted files other than `RECORD` while keeping the unpublished archive complete.
fn persist_archive_files(cache: &Cache, archive: &Path, files: &[HashedFile]) -> io::Result<()> {
    initialize_rayon_once();
    let targets = files
        .par_iter()
        // Keep RECORD private, since it may have been healed after hashing.
        .filter(|file| !file.path().ends_with("RECORD"))
        .map(|file| {
            let id = ArchiveFileId::from_digest(&file.object_digest_hex());
            (archive.join(file.path()), cache.archive_file(&id))
        })
        .collect::<Vec<_>>();

    // Group files by shard so its directory is created once and its files are linked by the
    // same worker, avoiding contention between workers on each shard directory.
    let mut shards: FxHashMap<&Path, Vec<_>> = FxHashMap::default();
    for (source, target) in &targets {
        let Some(parent) = target.parent() else {
            return Err(io::Error::new(
                io::ErrorKind::InvalidInput,
                "archive file path must have a parent directory",
            ));
        };
        shards.entry(parent).or_default().push((source, target));
    }

    let mut shards = shards
        .into_iter()
        .map(|(parent, files)| (parent, files, Ok(())))
        .collect::<Vec<_>>();
    // Start larger shards first so their work can overlap the remaining directory creation.
    shards.sort_unstable_by_key(|(_, files, _)| Reverse(files.len()));

    // Creating shards concurrently contends on their shared parent. Keep creation on this
    // thread, while workers link files in the shards that are already available.
    in_place_scope(|scope| -> io::Result<()> {
        for (parent, files, result) in &mut shards {
            fs_err::create_dir_all(parent)?;
            scope.spawn(move |_| {
                *result = files
                    .iter()
                    .try_for_each(|(source, target)| persist_archive_file(source, target));
            });
        }
        Ok(())
    })?;

    shards.into_iter().try_for_each(|(_, _, result)| result)
}

/// Publish a shared object and retain a hardlink in the archive, with a copy fallback.
fn persist_archive_file(src: &Path, dst: &Path) -> io::Result<()> {
    // The shard already exists, and most objects are new, so try linking before checking for an
    // existing object. This avoids an extra filesystem lookup for every new object.
    match fs_err::hard_link(src, dst) {
        Ok(()) => return Ok(()),
        Err(_) if dst.try_exists()? => {}
        Err(_) => return uv_fs::copy_atomic_sync(src, dst),
    }

    // This archive is still private, so it is safe to replace its extracted copy before publication.
    if let Err(err) = fs_err::remove_file(src)
        && err.kind() != io::ErrorKind::NotFound
    {
        return Err(err);
    }

    fs_err::hard_link(dst, src).or_else(|_| uv_fs::copy_atomic_sync(dst, src))
}

/// A wrapper around `RegistryClient` that manages a concurrency limit.
pub struct ManagedClient<'a> {
    pub unmanaged: &'a RegistryClient,
    control: Arc<Semaphore>,
}

impl<'a> ManagedClient<'a> {
    /// Create a new `ManagedClient` using the given client and concurrency semaphore.
    fn new(client: &'a RegistryClient, control: Arc<Semaphore>) -> Self {
        ManagedClient {
            unmanaged: client,
            control,
        }
    }

    /// Perform a request using the client, respecting the concurrency limit.
    ///
    /// If the concurrency limit has been reached, this method will wait until a pending
    /// operation completes before executing the closure.
    pub async fn managed<F, T>(&self, f: impl FnOnce(&'a RegistryClient) -> F) -> T
    where
        F: Future<Output = T>,
    {
        let _permit = self.control.acquire().await.unwrap();
        f(self.unmanaged).await
    }

    /// Perform a request using a client that internally manages the concurrency limit.
    ///
    /// The callback is passed the client and a semaphore. It must acquire the semaphore before
    /// any request through the client and drop it after.
    ///
    /// This method serves as an escape hatch for functions that may want to send multiple requests
    /// in parallel.
    pub async fn manual<F, T>(&'a self, f: impl FnOnce(&'a RegistryClient, &'a Semaphore) -> F) -> T
    where
        F: Future<Output = T>,
    {
        f(self.unmanaged, &self.control).await
    }
}

/// Returns the value of the `Content-Length` header from the [`reqwest::Response`], if present.
fn content_length(response: &reqwest::Response) -> Option<u64> {
    response
        .headers()
        .get(reqwest::header::CONTENT_LENGTH)
        .and_then(|val| val.to_str().ok())
        .and_then(|val| val.parse::<u64>().ok())
}

/// Return the bounds of a range response starting at `offset` with a known complete length.
fn content_range(
    response: &reqwest::Response,
    offset: u64,
    download_size: Option<u64>,
) -> Option<ContentRangeBytes> {
    let range = response
        .headers()
        .get(reqwest::header::CONTENT_RANGE)
        .and_then(|value| value.to_str().ok())
        .and_then(ContentRange::parse)?;

    let range = match range {
        ContentRange::Bytes(range) => range,
        ContentRange::UnboundBytes(ContentRangeUnbound {
            first_byte,
            last_byte,
        }) => ContentRangeBytes {
            first_byte,
            last_byte,
            complete_length: download_size?,
        },
        ContentRange::Unsatisfied(_) => return None,
    };

    if range.first_byte != offset
        || range.last_byte >= range.complete_length
        || download_size.is_some_and(|size| size != range.complete_length)
    {
        return None;
    }

    Some(range)
}

/// An asynchronous reader that reports progress as bytes are read.
struct ProgressReader<'a, R> {
    reader: R,
    index: usize,
    reporter: &'a dyn Reporter,
}

impl<'a, R> ProgressReader<'a, R> {
    /// Create a new [`ProgressReader`] that wraps another reader.
    fn new(reader: R, index: usize, reporter: &'a dyn Reporter) -> Self {
        Self {
            reader,
            index,
            reporter,
        }
    }
}

impl<R> AsyncRead for ProgressReader<'_, R>
where
    R: AsyncRead + Unpin,
{
    fn poll_read(
        mut self: Pin<&mut Self>,
        cx: &mut Context<'_>,
        buf: &mut ReadBuf<'_>,
    ) -> Poll<io::Result<()>> {
        Pin::new(&mut self.as_mut().reader)
            .poll_read(cx, buf)
            .map_ok(|()| {
                self.reporter
                    .on_download_progress(self.index, buf.filled().len() as u64);
            })
    }
}

/// A pointer to an archive in the cache, fetched from an HTTP archive.
///
/// Encoded with `MsgPack`, and represented on disk by a `.http` file.
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub struct HttpArchivePointer {
    archive: Archive,
}

impl HttpArchivePointer {
    /// Read an [`HttpArchivePointer`] from the cache.
    pub fn read_from(path: impl AsRef<Path>) -> Result<Option<Self>, Error> {
        match fs_err::File::open(path.as_ref()) {
            Ok(file) => {
                let data = DataWithCachePolicy::from_reader(file)?.data;
                let archive = rmp_serde::from_slice::<Archive>(&data)?;
                Ok(Some(Self { archive }))
            }
            Err(err) if err.kind() == io::ErrorKind::NotFound => Ok(None),
            Err(err) => Err(Error::CacheRead(err)),
        }
    }

    /// Return the [`Archive`] from the pointer.
    pub fn into_archive(self) -> Archive {
        self.archive
    }

    /// Return the [`CacheInfo`] from the pointer.
    pub fn to_cache_info(&self) -> CacheInfo {
        CacheInfo::default()
    }

    /// Return the [`BuildInfo`] from the pointer.
    pub fn to_build_info(&self) -> Option<BuildInfo> {
        None
    }
}

/// A pointer to an archive in the cache, fetched from a local path.
///
/// Encoded with `MsgPack`, and represented on disk by a `.rev` file.
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub struct PathArchivePointer {
    timestamp: Timestamp,
    archive: Archive,
}

impl PathArchivePointer {
    /// Read an [`PathArchivePointer`] from the cache.
    pub fn read_from(path: impl AsRef<Path>) -> Result<Option<Self>, Error> {
        match fs_err::read(path) {
            Ok(cached) => Ok(Some(rmp_serde::from_slice::<Self>(&cached)?)),
            Err(err) if err.kind() == io::ErrorKind::NotFound => Ok(None),
            Err(err) => Err(Error::CacheRead(err)),
        }
    }

    /// Write an [`PathArchivePointer`] to the cache.
    async fn write_to(&self, entry: &CacheEntry) -> Result<(), Error> {
        write_atomic(entry.path(), rmp_serde::to_vec(&self)?)
            .await
            .map_err(Error::CacheWrite)
    }

    /// Returns `true` if the archive is up-to-date with the given modified timestamp.
    pub fn is_up_to_date(&self, modified: Timestamp) -> bool {
        self.timestamp == modified
    }

    /// Return the [`Archive`] from the pointer.
    pub fn into_archive(self) -> Archive {
        self.archive
    }

    /// Return the [`CacheInfo`] from the pointer.
    pub fn to_cache_info(&self) -> CacheInfo {
        CacheInfo::from_timestamp(self.timestamp)
    }

    /// Return the [`BuildInfo`] from the pointer.
    pub fn to_build_info(&self) -> Option<BuildInfo> {
        None
    }
}