mod.rs

//! Given a set of requirements, find a set of compatible packages.

use std::borrow::Cow;
use std::fmt::{Display, Formatter};
use std::ops::Deref;
use std::sync::Arc;

use anyhow::Result;
use dashmap::{DashMap, DashSet};
use futures::{FutureExt, StreamExt};
use itertools::Itertools;
use pubgrub::error::PubGrubError;
use pubgrub::range::Range;
use pubgrub::solver::{Incompatibility, State};
use rustc_hash::{FxHashMap, FxHashSet};
use tokio_stream::wrappers::ReceiverStream;
use tracing::{debug, info_span, instrument, trace, Instrument};
use url::Url;

use distribution_types::{
    BuiltDist, Dist, DistributionMetadata, IncompatibleDist, IncompatibleSource, IncompatibleWheel,
    InstalledDist, Name, RemoteSource, ResolvedDist, ResolvedDistRef, SourceDist, VersionOrUrl,
};
pub(crate) use locals::Locals;
use pep440_rs::{Version, MIN_VERSION};
use pep508_rs::{MarkerEnvironment, MarkerTree, Requirement};
use platform_tags::Tags;
use pypi_types::Metadata23;
pub(crate) use urls::Urls;
use uv_client::{FlatIndex, RegistryClient};
use uv_distribution::DistributionDatabase;
use uv_interpreter::Interpreter;
use uv_normalize::{ExtraName, PackageName};
use uv_types::{BuildContext, Constraints, InstalledPackagesProvider, Overrides};

use crate::candidate_selector::{CandidateDist, CandidateSelector};

use crate::dependency_provider::UvDependencyProvider;
use crate::editables::Editables;
use crate::error::ResolveError;
use crate::manifest::Manifest;
use crate::pins::FilePins;
use crate::preferences::Preferences;
use crate::pubgrub::{
    PubGrubDependencies, PubGrubDistribution, PubGrubPackage, PubGrubPriorities, PubGrubPython,
    PubGrubSpecifier,
};
use crate::python_requirement::PythonRequirement;
use crate::resolution::ResolutionGraph;
pub use crate::resolver::index::InMemoryIndex;
pub use crate::resolver::provider::{
    DefaultResolverProvider, PackageVersionsResult, ResolverProvider, VersionsResponse,
    WheelMetadataResult,
};
use crate::resolver::reporter::Facade;
pub use crate::resolver::reporter::{BuildId, Reporter};
use crate::yanks::AllowedYanks;
use crate::{DependencyMode, Exclusions, Options, VersionMap};

mod index;
mod locals;
mod provider;
mod reporter;
mod urls;

/// The package version is unavailable and cannot be used
/// Unlike [`PackageUnavailable`] this applies to a single version of the package
#[derive(Debug, Clone)]
pub(crate) enum UnavailableVersion {
    /// Version is incompatible because it has no usable distributions
    IncompatibleDist(IncompatibleDist),
}

/// The package is unavailable and cannot be used
#[derive(Debug, Clone)]
pub(crate) enum UnavailablePackage {
    /// Index lookups were disabled (i.e., `--no-index`) and the package was not found in a flat index (i.e. from `--find-links`)
    NoIndex,
    /// Network requests were disabled (i.e., `--offline`), and the package was not found in the cache.
    Offline,
    /// The package was not found in the registry
    NotFound,
}

enum ResolverVersion {
    /// A usable version
    Available(Version),
    /// A version that is not usable for some reason
    Unavailable(Version, UnavailableVersion),
}

pub struct Resolver<
    'a,
    Provider: ResolverProvider,
    InstalledPackages: InstalledPackagesProvider + Send + Sync,
> {
    project: Option<PackageName>,
    requirements: Vec<Requirement>,
    constraints: Constraints,
    overrides: Overrides,
    preferences: Preferences,
    exclusions: Exclusions,
    editables: Editables,
    urls: Urls,
    locals: Locals,
    dependency_mode: DependencyMode,
    markers: &'a MarkerEnvironment,
    python_requirement: PythonRequirement,
    selector: CandidateSelector,
    index: &'a InMemoryIndex,
    installed_packages: &'a InstalledPackages,
    /// Incompatibilities for packages that are entirely unavailable
    unavailable_packages: DashMap<PackageName, UnavailablePackage>,
    /// The set of all registry-based packages visited during resolution.
    visited: DashSet<PackageName>,
    reporter: Option<Arc<dyn Reporter>>,
    provider: Provider,
}

impl<
        'a,
        Context: BuildContext + Send + Sync,
        InstalledPackages: InstalledPackagesProvider + Send + Sync,
    > Resolver<'a, DefaultResolverProvider<'a, Context>, InstalledPackages>
{
    /// Initialize a new resolver using the default backend doing real requests.
    ///
    /// Reads the flat index entries.
    #[allow(clippy::too_many_arguments)]
    pub fn new(
        manifest: Manifest,
        options: Options,
        markers: &'a MarkerEnvironment,
        interpreter: &'a Interpreter,
        tags: &'a Tags,
        client: &'a RegistryClient,
        flat_index: &'a FlatIndex,
        index: &'a InMemoryIndex,
        build_context: &'a Context,
        installed_packages: &'a InstalledPackages,
    ) -> Result<Self, ResolveError> {
        let provider = DefaultResolverProvider::new(
            client,
            DistributionDatabase::new(client, build_context),
            flat_index,
            tags,
            PythonRequirement::new(interpreter, markers),
            AllowedYanks::from_manifest(&manifest, markers),
            options.exclude_newer,
            build_context.no_binary(),
            build_context.no_build(),
        );
        Self::new_custom_io(
            manifest,
            options,
            markers,
            PythonRequirement::new(interpreter, markers),
            index,
            provider,
            installed_packages,
        )
    }
}

impl<
        'a,
        Provider: ResolverProvider,
        InstalledPackages: InstalledPackagesProvider + Send + Sync,
    > Resolver<'a, Provider, InstalledPackages>
{
    /// Initialize a new resolver using a user provided backend.
    pub fn new_custom_io(
        manifest: Manifest,
        options: Options,
        markers: &'a MarkerEnvironment,
        python_requirement: PythonRequirement,
        index: &'a InMemoryIndex,
        provider: Provider,
        installed_packages: &'a InstalledPackages,
    ) -> Result<Self, ResolveError> {
        Ok(Self {
            index,
            unavailable_packages: DashMap::default(),
            visited: DashSet::default(),
            selector: CandidateSelector::for_resolution(options, &manifest, markers),
            dependency_mode: options.dependency_mode,
            urls: Urls::from_manifest(&manifest, markers)?,
            locals: Locals::from_manifest(&manifest, markers),
            project: manifest.project,
            requirements: manifest.requirements,
            constraints: manifest.constraints,
            overrides: manifest.overrides,
            preferences: Preferences::from_iter(manifest.preferences, markers),
            exclusions: manifest.exclusions,
            editables: Editables::from_requirements(manifest.editables),
            markers,
            python_requirement,
            reporter: None,
            provider,
            installed_packages,
        })
    }

    /// Set the [`Reporter`] to use for this installer.
    #[must_use]
    pub fn with_reporter(self, reporter: impl Reporter + 'static) -> Self {
        let reporter = Arc::new(reporter);
        Self {
            reporter: Some(reporter.clone()),
            provider: self.provider.with_reporter(Facade { reporter }),
            ..self
        }
    }

    /// Resolve a set of requirements into a set of pinned versions.
    pub async fn resolve(self) -> Result<ResolutionGraph, ResolveError> {
        // A channel to fetch package metadata (e.g., given `flask`, fetch all versions) and version
        // metadata (e.g., given `flask==1.0.0`, fetch the metadata for that version).
        // Channel size is set to the same size as the task buffer for simplicity.
        let (request_sink, request_stream) = tokio::sync::mpsc::channel(50);

        // Run the fetcher.
        let requests_fut = self.fetch(request_stream).fuse();

        // Run the solver.
        let resolve_fut = self.solve(request_sink).fuse();

        // Wait for both to complete.
        match tokio::try_join!(requests_fut, resolve_fut) {
            Ok(((), resolution)) => {
                self.on_complete();
                Ok(resolution)
            }
            Err(err) => {
                // Add version information to improve unsat error messages.
                Err(if let ResolveError::NoSolution(err) = err {
                    ResolveError::NoSolution(
                        err.with_available_versions(
                            &self.python_requirement,
                            &self.visited,
                            &self.index.packages,
                        )
                        .with_selector(self.selector.clone())
                        .with_python_requirement(&self.python_requirement)
                        .with_index_locations(self.provider.index_locations())
                        .with_unavailable_packages(&self.unavailable_packages),
                    )
                } else {
                    err
                })
            }
        }
    }

    /// Run the `PubGrub` solver.
    #[instrument(skip_all)]
    async fn solve(
        &self,
        request_sink: tokio::sync::mpsc::Sender<Request>,
    ) -> Result<ResolutionGraph, ResolveError> {
        let root = PubGrubPackage::Root(self.project.clone());
        let state = ResolverState {
            pubgrub: State::init(root.clone(), MIN_VERSION.clone()),
            next: root,
            pins: FilePins::default(),
            priorities: PubGrubPriorities::default(),
            added_dependencies: FxHashMap::default(),
            markers: FxHashMap::default(),
        };
        let mut forked_states = vec![state];
        let mut resolutions = vec![];

        debug!(
            "Solving with target Python version {}",
            self.python_requirement.target()
        );

        'FORK: while let Some(mut state) = forked_states.pop() {
            loop {
                // Run unit propagation.
                state.pubgrub.unit_propagation(state.next.clone())?;

                // Pre-visit all candidate packages, to allow metadata to be fetched in parallel. If
                // the dependency mode is direct, we only need to visit the root package.
                if self.dependency_mode.is_transitive() {
                    Self::pre_visit(
                        state.pubgrub.partial_solution.prioritized_packages(),
                        &request_sink,
                    )
                    .await?;
                }

                // Choose a package version.
                let Some(highest_priority_pkg) = state
                    .pubgrub
                    .partial_solution
                    .pick_highest_priority_pkg(|package, _range| {
                        state.priorities.get(package).unwrap_or_default()
                    })
                else {
                    resolutions.push(state.into_resolution());
                    continue 'FORK;
                };
                state.next = highest_priority_pkg;

                let term_intersection = state
                    .pubgrub
                    .partial_solution
                    .term_intersection_for_package(&state.next)
                    .ok_or_else(|| {
                        PubGrubError::Failure(
                            "a package was chosen but we don't have a term.".into(),
                        )
                    })?;
                let decision = self
                    .choose_version(
                        &state.next,
                        term_intersection.unwrap_positive(),
                        &mut state.pins,
                        &request_sink,
                    )
                    .await?;

                // Pick the next compatible version.
                let version = match decision {
                    None => {
                        debug!("No compatible version found for: {next}", next = state.next);

                        let term_intersection = state
                            .pubgrub
                            .partial_solution
                            .term_intersection_for_package(&state.next)
                            .expect("a package was chosen but we don't have a term.");

                        let reason = {
                            if let PubGrubPackage::Package {
                                name: ref package_name,
                                ..
                            } = state.next
                            {
                                // Check if the decision was due to the package being unavailable
                                self.unavailable_packages.get(package_name).map(
                                    |entry| match *entry {
                                        UnavailablePackage::NoIndex => {
                                            "was not found in the provided package locations"
                                        }
                                        UnavailablePackage::Offline => "was not found in the cache",
                                        UnavailablePackage::NotFound => {
                                            "was not found in the package registry"
                                        }
                                    },
                                )
                            } else {
                                None
                            }
                        };

                        let inc = Incompatibility::no_versions(
                            state.next.clone(),
                            term_intersection.clone(),
                            reason.map(ToString::to_string),
                        );

                        state.pubgrub.add_incompatibility(inc);
                        continue;
                    }
                    Some(version) => version,
                };
                let version = match version {
                    ResolverVersion::Available(version) => version,
                    ResolverVersion::Unavailable(version, unavailable) => {
                        let reason = match unavailable {
                            // Incompatible requires-python versions are special in that we track
                            // them as incompatible dependencies instead of marking the package version
                            // as unavailable directly
                            UnavailableVersion::IncompatibleDist(
                                IncompatibleDist::Source(IncompatibleSource::RequiresPython(
                                    requires_python,
                                ))
                                | IncompatibleDist::Wheel(IncompatibleWheel::RequiresPython(
                                    requires_python,
                                )),
                            ) => {
                                let python_version = requires_python
                                    .iter()
                                    .map(PubGrubSpecifier::try_from)
                                    .fold_ok(Range::full(), |range, specifier| {
                                        range.intersection(&specifier.into())
                                    })?;

                                let package = &state.next;
                                for kind in [PubGrubPython::Installed, PubGrubPython::Target] {
                                    state.pubgrub.add_incompatibility(
                                        Incompatibility::from_dependency(
                                            package.clone(),
                                            Range::singleton(version.clone()),
                                            (PubGrubPackage::Python(kind), python_version.clone()),
                                        ),
                                    );
                                }
                                state
                                    .pubgrub
                                    .partial_solution
                                    .add_decision(state.next.clone(), version);
                                continue;
                            }
                            UnavailableVersion::IncompatibleDist(incompatibility) => {
                                incompatibility.to_string()
                            }
                        };
                        state
                            .pubgrub
                            .add_incompatibility(Incompatibility::unavailable(
                                state.next.clone(),
                                version.clone(),
                                reason,
                            ));
                        continue;
                    }
                };

                self.on_progress(&state.next, &version);

                if state
                    .added_dependencies
                    .entry(state.next.clone())
                    .or_default()
                    .insert(version.clone())
                {
                    // Retrieve that package dependencies.
                    let package = &state.next;
                    let forks = self
                        .get_dependencies_forking(
                            package,
                            &version,
                            &mut state.priorities,
                            &request_sink,
                        )
                        .await?;
                    for fork in forks {
                        let mut state = state.clone();
                        let dependencies = match fork {
                            Dependencies::Unavailable(reason) => {
                                let message = {
                                    if matches!(package, PubGrubPackage::Root(_)) {
                                        // Including front-matter for the root package is redundant
                                        reason.clone()
                                    } else {
                                        format!("its dependencies are unusable because {reason}")
                                    }
                                };
                                state
                                    .pubgrub
                                    .add_incompatibility(Incompatibility::unavailable(
                                        package.clone(),
                                        version.clone(),
                                        message,
                                    ));
                                forked_states.push(state);
                                continue;
                            }
                            Dependencies::Available(constraints)
                                if constraints
                                    .iter()
                                    .any(|(dependency, _, _)| dependency == package) =>
                            {
                                return Err(PubGrubError::SelfDependency {
                                    package: package.clone(),
                                    version: version.clone(),
                                }
                                .into());
                            }
                            Dependencies::Available(constraints) => constraints,
                        };

                        /*
                        let parent = &package;
                        // eprintln!("===== {:?} {:?} =====", parent.name(), version);
                        for &(ref package, ref versions, ref marker) in dependencies.iter() {
                            // use std::ops::Bound;
                            // let range = match versions.iter().next().unwrap() {
                            // (Bound::Included(s), Bound::Included(e)) => {
                            // format!("{s}..={e}")
                            // }
                            // (Bound::Included(s), Bound::Excluded(e)) => {
                            // format!("{s}..{e}")
                            // }
                            // (Bound::Included(s), Bound::Unbounded) => {
                            // format!("{s}..")
                            // }
                            // (Bound::Unbounded, Bound::Included(e)) => {
                            // format!("..={e}")
                            // }
                            // (Bound::Unbounded, Bound::Excluded(e)) => {
                            // format!("..{e}")
                            // }
                            // (Bound::Unbounded, Bound::Unbounded) => {
                            // format!("..")
                            // }
                            // _ => unreachable!(),
                            // };
                            if let Some(ref marker) = *marker {
                                // eprintln!("dep: {:?} {} ; {}", package.name(), range, marker);
                                // state
                                // .markers
                                // .entry(package.clone())
                                // .or_insert_with(|| MarkerTree::And(vec![]))
                                // .and(marker.clone());
                            } else {
                                // eprintln!("dep: {:?} {}", package.name(), range);
                            }
                        }
                        // eprintln!("==========");
                        */

                        // Add that package and version if the dependencies are not problematic.
                        let dep_incompats = state.pubgrub.add_incompatibility_from_dependencies(
                            package.clone(),
                            version.clone(),
                            dependencies
                                .into_iter()
                                .map(|(package, versions, _)| (package, versions)),
                        );

                        state.pubgrub.partial_solution.add_version(
                            package.clone(),
                            version.clone(),
                            dep_incompats,
                            &state.pubgrub.incompatibility_store,
                        );
                        forked_states.push(state);
                    }
                    continue 'FORK;
                } else {
                    // `dep_incompats` are already in `incompatibilities` so we know there are not satisfied
                    // terms and can add the decision directly.
                    state
                        .pubgrub
                        .partial_solution
                        .add_decision(state.next.clone(), version);
                }
            }
        }
        let mut combined = Resolution::default();
        for resolution in resolutions {
            combined.union(resolution);
        }
        ResolutionGraph::from_state(
            &self.index,
            &self.preferences,
            self.editables.clone(),
            combined,
        )
    }

    /// Visit a [`PubGrubPackage`] prior to selection. This should be called on a [`PubGrubPackage`]
    /// before it is selected, to allow metadata to be fetched in parallel.
    async fn visit_package(
        &self,
        package: &PubGrubPackage,
        priorities: &mut PubGrubPriorities,
        request_sink: &tokio::sync::mpsc::Sender<Request>,
    ) -> Result<(), ResolveError> {
        match package {
            PubGrubPackage::Root(_) => {}
            PubGrubPackage::Python(_) => {}
            PubGrubPackage::Package {
                name: package_name,
                url: None,
                ..
            } => {
                // Emit a request to fetch the metadata for this package.
                if self.index.packages.register(package_name.clone()) {
                    priorities.add(package_name.clone());
                    request_sink
                        .send(Request::Package(package_name.clone()))
                        .await?;
                }
            }
            PubGrubPackage::Package {
                name: package_name,
                url: Some(url),
                ..
            } => {
                // Emit a request to fetch the metadata for this distribution.
                let dist = Dist::from_url(package_name.clone(), url.clone())?;
                if self.index.distributions.register(dist.package_id()) {
                    priorities.add(dist.name().clone());
                    request_sink.send(Request::Dist(dist)).await?;
                }
            }
        }
        Ok(())
    }

    /// Visit the set of [`PubGrubPackage`] candidates prior to selection. This allows us to fetch
    /// metadata for all of the packages in parallel.
    async fn pre_visit<'data>(
        packages: impl Iterator<Item = (&'data PubGrubPackage, &'data Range<Version>)>,
        request_sink: &tokio::sync::mpsc::Sender<Request>,
    ) -> Result<(), ResolveError> {
        // Iterate over the potential packages, and fetch file metadata for any of them. These
        // represent our current best guesses for the versions that we _might_ select.
        for (package, range) in packages {
            let PubGrubPackage::Package {
                name: package_name,
                url: None,
                ..
            } = package
            else {
                continue;
            };
            request_sink
                .send(Request::Prefetch(package_name.clone(), range.clone()))
                .await?;
        }
        Ok(())
    }

    /// Given a set of candidate packages, choose the next package (and version) to add to the
    /// partial solution.
    ///
    /// Returns [None] when there are no versions in the given range.
    #[instrument(skip_all, fields(%package))]
    async fn choose_version(
        &self,
        package: &'a PubGrubPackage,
        range: &Range<Version>,
        pins: &mut FilePins,
        request_sink: &tokio::sync::mpsc::Sender<Request>,
    ) -> Result<Option<ResolverVersion>, ResolveError> {
        match package {
            PubGrubPackage::Root(_) => Ok(Some(ResolverVersion::Available(MIN_VERSION.clone()))),

            PubGrubPackage::Python(PubGrubPython::Installed) => {
                let version = self.python_requirement.installed();
                if range.contains(version) {
                    Ok(Some(ResolverVersion::Available(version.deref().clone())))
                } else {
                    Ok(None)
                }
            }

            PubGrubPackage::Python(PubGrubPython::Target) => {
                let version = self.python_requirement.target();
                if range.contains(version) {
                    Ok(Some(ResolverVersion::Available(version.deref().clone())))
                } else {
                    Ok(None)
                }
            }

            PubGrubPackage::Package {
                name: package_name,
                extra,
                marker,
                url: Some(url),
            } => {
                if let Some(extra) = extra {
                    debug!(
                        "Searching for a compatible version of {package_name}[{extra}] @ {url} ({range})",
                    );
                } else {
                    debug!(
                        "Searching for a compatible version of {package_name} @ {url} ({range})"
                    );
                }

                // If the dist is an editable, return the version from the editable metadata.
                if let Some((_local, metadata)) = self.editables.get(package_name) {
                    let version = &metadata.version;

                    // The version is incompatible with the requirement.
                    if !range.contains(version) {
                        return Ok(None);
                    }

                    // The version is incompatible due to its Python requirement.
                    if let Some(requires_python) = metadata.requires_python.as_ref() {
                        let target = self.python_requirement.target();
                        if !requires_python.contains(target) {
                            return Ok(Some(ResolverVersion::Unavailable(
                                version.clone(),
                                UnavailableVersion::IncompatibleDist(IncompatibleDist::Source(
                                    IncompatibleSource::RequiresPython(requires_python.clone()),
                                )),
                            )));
                        }
                    }

                    return Ok(Some(ResolverVersion::Available(version.clone())));
                }

                let dist = PubGrubDistribution::from_url(package_name, url);
                let metadata = self
                    .index
                    .distributions
                    .wait(&dist.package_id())
                    .await
                    .ok_or(ResolveError::Unregistered)?;
                let version = &metadata.version;

                // The version is incompatible with the requirement.
                if !range.contains(version) {
                    return Ok(None);
                }

                // The version is incompatible due to its Python requirement.
                if let Some(requires_python) = metadata.requires_python.as_ref() {
                    let target = self.python_requirement.target();
                    if !requires_python.contains(target) {
                        return Ok(Some(ResolverVersion::Unavailable(
                            version.clone(),
                            UnavailableVersion::IncompatibleDist(IncompatibleDist::Source(
                                IncompatibleSource::RequiresPython(requires_python.clone()),
                            )),
                        )));
                    }
                }

                Ok(Some(ResolverVersion::Available(version.clone())))
            }

            PubGrubPackage::Package {
                name: package_name,
                extra,
                marker,
                url: None,
            } => {
                // Wait for the metadata to be available.
                let versions_response = self
                    .index
                    .packages
                    .wait(package_name)
                    .instrument(info_span!("package_wait", %package_name))
                    .await
                    .ok_or(ResolveError::Unregistered)?;
                self.visited.insert(package_name.clone());

                let empty_version_map = VersionMap::default();
                let version_map = match *versions_response {
                    VersionsResponse::Found(ref version_map) => version_map,
                    VersionsResponse::NoIndex => {
                        self.unavailable_packages
                            .insert(package_name.clone(), UnavailablePackage::NoIndex);
                        &empty_version_map
                    }
                    VersionsResponse::Offline => {
                        self.unavailable_packages
                            .insert(package_name.clone(), UnavailablePackage::Offline);
                        &empty_version_map
                    }
                    VersionsResponse::NotFound => {
                        self.unavailable_packages
                            .insert(package_name.clone(), UnavailablePackage::NotFound);
                        &empty_version_map
                    }
                };

                if let Some(extra) = extra {
                    debug!(
                        "Searching for a compatible version of {package_name}[{extra}] ({range})",
                    );
                } else {
                    debug!("Searching for a compatible version of {package_name} ({range})");
                }

                // Find a version.
                let Some(candidate) = self.selector.select(
                    package_name,
                    range,
                    version_map,
                    &self.preferences,
                    self.installed_packages,
                    &self.exclusions,
                ) else {
                    // Short circuit: we couldn't find _any_ versions for a package.
                    return Ok(None);
                };

                let dist = match candidate.dist() {
                    CandidateDist::Compatible(dist) => dist,
                    CandidateDist::Incompatible(incompatibility) => {
                        // If the version is incompatible because no distributions are compatible, exit early.
                        return Ok(Some(ResolverVersion::Unavailable(
                            candidate.version().clone(),
                            UnavailableVersion::IncompatibleDist(incompatibility.clone()),
                        )));
                    }
                };

                let filename = match dist.for_installation() {
                    ResolvedDistRef::Installable(dist) => {
                        dist.filename().unwrap_or(Cow::Borrowed("unknown filename"))
                    }
                    ResolvedDistRef::Installed(_) => Cow::Borrowed("installed"),
                };
                if let Some(extra) = extra {
                    debug!(
                        "Selecting: {}[{}]=={} ({})",
                        candidate.name(),
                        extra,
                        candidate.version(),
                        filename,
                    );
                } else {
                    debug!(
                        "Selecting: {}=={} ({})",
                        candidate.name(),
                        candidate.version(),
                        filename,
                    );
                }

                // We want to return a package pinned to a specific version; but we _also_ want to
                // store the exact file that we selected to satisfy that version.
                pins.insert(&candidate, dist);

                let version = candidate.version().clone();

                // Emit a request to fetch the metadata for this version.

                if self.index.distributions.register(candidate.package_id()) {
                    let request = match dist.for_resolution() {
                        ResolvedDistRef::Installable(dist) => Request::Dist(dist.clone()),
                        ResolvedDistRef::Installed(dist) => Request::Installed(dist.clone()),
                    };
                    request_sink.send(request).await?;
                }
                Ok(Some(ResolverVersion::Available(version)))
            }
        }
    }

    /// Given a candidate package and version, return its dependencies.
    #[instrument(skip_all, fields(%package, %version))]
    async fn get_dependencies_forking(
        &self,
        package: &PubGrubPackage,
        version: &Version,
        priorities: &mut PubGrubPriorities,
        request_sink: &tokio::sync::mpsc::Sender<Request>,
    ) -> Result<Vec<Dependencies>, ResolveError> {
        type Dep = (PubGrubPackage, Range<Version>, Option<MarkerTree>);
        let deps: Vec<Dep> = match self
            .get_dependencies(package, version, priorities, request_sink)
            .await?
        {
            Dependencies::Available(deps) => deps,
            Dependencies::Unavailable(err) => return Ok(vec![Dependencies::Unavailable(err)]),
        };
        // let mut by_grouping: FxHashMap<(&PackageName, &Option<ExtraName>, bool), Vec<&Dep>> =
        // FxHashMap::default();
        // let mut by_grouping: FxHashMap<&PackageName, Vec<&Dep>> = FxHashMap::default();
        // This assumes that the version ranges are non-overlapping. When we do this for real,
        // I believe it is true that we do NOT want to fork when there are overlapping version
        // ranges.
        let mut by_grouping: FxHashMap<&PackageName, FxHashMap<&Range<Version>, Vec<&Dep>>> =
            FxHashMap::default();
        for dep in deps.iter() {
            let (ref pkg, ref range, _) = *dep;
            let name = match *pkg {
                PubGrubPackage::Root(_) | PubGrubPackage::Python(_) => unreachable!(),
                PubGrubPackage::Package { ref name, .. } => name,
            };
            by_grouping
                .entry(name)
                .or_insert(FxHashMap::default())
                .entry(range)
                .or_insert(vec![])
                .push(dep);
        }
        let mut forks: Vec<Vec<Dep>> = vec![vec![]];
        for (_, mut groups) in by_grouping {
            if groups.len() <= 1 {
                for deps in groups.into_values() {
                    for fork in forks.iter_mut() {
                        fork.extend(deps.iter().map(|dep| (*dep).clone()));
                    }
                }
            } else {
                let mut new_forks: Vec<Vec<Dep>> = vec![];
                for deps in groups.into_values() {
                    let mut new_forks_for_group = forks.clone();
                    for fork in new_forks_for_group.iter_mut() {
                        fork.extend(deps.iter().map(|dep| (*dep).clone()));
                    }
                    new_forks.extend(new_forks_for_group);
                }
                forks = new_forks;
            }
        }
        Ok(forks.into_iter().map(Dependencies::Available).collect())
    }

    /// Given a candidate package and version, return its dependencies.
    #[instrument(skip_all, fields(%package, %version))]
    async fn get_dependencies(
        &self,
        package: &PubGrubPackage,
        version: &Version,
        priorities: &mut PubGrubPriorities,
        request_sink: &tokio::sync::mpsc::Sender<Request>,
    ) -> Result<Dependencies, ResolveError> {
        match package {
            PubGrubPackage::Root(_) => {
                // Add the root requirements.
                let constraints = PubGrubDependencies::from_requirements(
                    &self.requirements,
                    &self.constraints,
                    &self.overrides,
                    None,
                    None,
                    None,
                    &self.urls,
                    &self.locals,
                    self.markers,
                );

                let mut constraints = match constraints {
                    Ok(constraints) => constraints,
                    Err(err) => {
                        return Ok(Dependencies::Unavailable(uncapitalize(err.to_string())));
                    }
                };

                for (package, version, _) in constraints.iter() {
                    debug!("Adding direct dependency: {package}{version}");

                    // Emit a request to fetch the metadata for this package.
                    self.visit_package(package, priorities, request_sink)
                        .await?;
                }

                // Add a dependency on each editable.
                for (editable, metadata) in self.editables.iter() {
                    constraints.push(
                        PubGrubPackage::from_package(metadata.name.clone(), None, None, &self.urls),
                        Range::singleton(metadata.version.clone()),
                        None,
                    );
                    for extra in &editable.extras {
                        constraints.push(
                            PubGrubPackage::from_package(
                                metadata.name.clone(),
                                Some(extra.clone()),
                                None,
                                &self.urls,
                            ),
                            Range::singleton(metadata.version.clone()),
                            None,
                        );
                    }
                }

                Ok(Dependencies::Available(constraints.into()))
            }

            PubGrubPackage::Python(_) => Ok(Dependencies::Available(Vec::default())),

            /*
            PubGrubPackage::Package {
                name: package_name,
                extra,
                marker: Some(marker),
                url,
            } => {
                let package = PubGrubPackage::Package {
                    name: package_name.clone(),
                    extra: None,
                    marker: None,
                    url: url.clone(),
                };
                let version = Range::singleton(version.clone());
                let constraint = (package, version, None);
                Ok(Dependencies::Available(vec![constraint]))
            }
            */
            PubGrubPackage::Package {
                name: package_name,
                extra,
                marker,
                url,
            } => {
                // If we're excluding transitive dependencies, short-circuit.
                if self.dependency_mode.is_direct() {
                    // If an extra is provided, wait for the metadata to be available, since it's
                    // still required for reporting diagnostics.
                    if extra.is_some() {
                        // Determine the distribution to lookup.
                        let dist = match url {
                            Some(url) => PubGrubDistribution::from_url(package_name, url),
                            None => PubGrubDistribution::from_registry(package_name, version),
                        };
                        let package_id = dist.package_id();

                        // Wait for the metadata to be available.
                        self.index
                            .distributions
                            .wait(&package_id)
                            .instrument(info_span!("distributions_wait", %package_id))
                            .await
                            .ok_or(ResolveError::Unregistered)?;
                    }

                    return Ok(Dependencies::Available(Vec::default()));
                }

                // Determine if the distribution is editable.
                if let Some((_local, metadata)) = self.editables.get(package_name) {
                    let mut constraints = PubGrubDependencies::from_requirements(
                        &metadata.requires_dist,
                        &self.constraints,
                        &self.overrides,
                        Some(package_name),
                        extra.as_ref(),
                        marker.clone(),
                        &self.urls,
                        &self.locals,
                        self.markers,
                    )?;

                    for (dep_package, dep_version, _) in constraints.iter() {
                        debug!("Adding transitive dependency for {package}{version}: {dep_package}{dep_version}");

                        // Emit a request to fetch the metadata for this package.
                        self.visit_package(dep_package, priorities, request_sink)
                            .await?;
                    }

                    // If a package has an extra, insert a constraint on the base package.
                    if extra.is_some() || marker.is_some() {
                        constraints.push(
                            PubGrubPackage::Package {
                                name: package_name.clone(),
                                extra: None,
                                marker: None,
                                url: url.clone(),
                            },
                            Range::singleton(version.clone()),
                            None,
                        );
                    }

                    return Ok(Dependencies::Available(constraints.into()));
                }

                // Determine the distribution to lookup.
                let dist = match url {
                    Some(url) => PubGrubDistribution::from_url(package_name, url),
                    None => PubGrubDistribution::from_registry(package_name, version),
                };
                let package_id = dist.package_id();

                // If the package does not exist in the registry or locally, we cannot fetch its dependencies
                if self.unavailable_packages.get(package_name).is_some()
                    && self
                        .installed_packages
                        .get_packages(package_name)
                        .is_empty()
                {
                    debug_assert!(
                        false,
                        "Dependencies were requested for a package that is not available"
                    );
                    return Ok(Dependencies::Unavailable(
                        "The package is unavailable".to_string(),
                    ));
                }

                // Wait for the metadata to be available.
                let metadata = self
                    .index
                    .distributions
                    .wait(&package_id)
                    .instrument(info_span!("distributions_wait", %package_id))
                    .await
                    .ok_or(ResolveError::Unregistered)?;

                let mut constraints = PubGrubDependencies::from_requirements(
                    &metadata.requires_dist,
                    &self.constraints,
                    &self.overrides,
                    Some(package_name),
                    extra.as_ref(),
                    marker.clone(),
                    &self.urls,
                    &self.locals,
                    self.markers,
                )?;

                for (package, version, _) in constraints.iter() {
                    debug!("Adding transitive dependency: {package}{version}");

                    // Emit a request to fetch the metadata for this package.
                    self.visit_package(package, priorities, request_sink)
                        .await?;
                }

                // If a package has an extra, insert a constraint on the base package.
                if extra.is_some() || marker.is_some() {
                    constraints.push(
                        PubGrubPackage::Package {
                            name: package_name.clone(),
                            extra: None,
                            marker: None,
                            url: url.clone(),
                        },
                        Range::singleton(version.clone()),
                        None,
                    );
                }

                Ok(Dependencies::Available(constraints.into()))
            }
        }
    }

    /// Fetch the metadata for a stream of packages and versions.
    async fn fetch(
        &self,
        request_stream: tokio::sync::mpsc::Receiver<Request>,
    ) -> Result<(), ResolveError> {
        let mut response_stream = ReceiverStream::new(request_stream)
            .map(|request| self.process_request(request).boxed())
            .buffer_unordered(50);

        while let Some(response) = response_stream.next().await {
            match response? {
                Some(Response::Package(package_name, version_map)) => {
                    trace!("Received package metadata for: {package_name}");
                    self.index.packages.done(package_name, version_map);
                }
                Some(Response::Installed { dist, metadata }) => {
                    trace!("Received installed distribution metadata for: {dist}");
                    self.index.distributions.done(dist.package_id(), metadata);
                }
                Some(Response::Dist {
                    dist: Dist::Built(dist),
                    metadata,
                    precise: _,
                }) => {
                    trace!("Received built distribution metadata for: {dist}");
                    self.index.distributions.done(dist.package_id(), metadata);
                }
                Some(Response::Dist {
                    dist: Dist::Source(distribution),
                    metadata,
                    precise,
                }) => {
                    trace!("Received source distribution metadata for: {distribution}");
                    self.index
                        .distributions
                        .done(distribution.package_id(), metadata);
                    if let Some(precise) = precise {
                        match distribution {
                            SourceDist::DirectUrl(sdist) => {
                                self.index.redirects.insert(sdist.url.to_url(), precise);
                            }
                            SourceDist::Git(sdist) => {
                                self.index.redirects.insert(sdist.url.to_url(), precise);
                            }
                            SourceDist::Path(sdist) => {
                                self.index.redirects.insert(sdist.url.to_url(), precise);
                            }
                            SourceDist::Registry(_) => {}
                        }
                    }
                }
                None => {}
            }
        }

        Ok::<(), ResolveError>(())
    }

    #[instrument(skip_all, fields(%request))]
    async fn process_request(&self, request: Request) -> Result<Option<Response>, ResolveError> {
        match request {
            // Fetch package metadata from the registry.
            Request::Package(package_name) => {
                let package_versions = self
                    .provider
                    .get_package_versions(&package_name)
                    .boxed()
                    .await
                    .map_err(ResolveError::Client)?;

                Ok(Some(Response::Package(package_name, package_versions)))
            }

            // Fetch distribution metadata from the distribution database.
            Request::Dist(dist) => {
                let (metadata, precise) = self
                    .provider
                    .get_or_build_wheel_metadata(&dist)
                    .boxed()
                    .await
                    .map_err(|err| match dist.clone() {
                        Dist::Built(BuiltDist::Path(built_dist)) => {
                            ResolveError::Read(Box::new(built_dist), err)
                        }
                        Dist::Source(SourceDist::Path(source_dist)) => {
                            ResolveError::Build(Box::new(source_dist), err)
                        }
                        Dist::Built(built_dist) => ResolveError::Fetch(Box::new(built_dist), err),
                        Dist::Source(source_dist) => {
                            ResolveError::FetchAndBuild(Box::new(source_dist), err)
                        }
                    })?;
                Ok(Some(Response::Dist {
                    dist,
                    metadata,
                    precise,
                }))
            }

            Request::Installed(dist) => {
                let metadata = dist
                    .metadata()
                    .map_err(|err| ResolveError::ReadInstalled(Box::new(dist.clone()), err))?;
                Ok(Some(Response::Installed { dist, metadata }))
            }

            // Pre-fetch the package and distribution metadata.
            Request::Prefetch(package_name, range) => {
                // Wait for the package metadata to become available.
                let versions_response = self
                    .index
                    .packages
                    .wait(&package_name)
                    .await
                    .ok_or(ResolveError::Unregistered)?;

                let version_map = match *versions_response {
                    VersionsResponse::Found(ref version_map) => version_map,
                    // Short-circuit if we did not find any versions for the package
                    VersionsResponse::NoIndex => {
                        self.unavailable_packages
                            .insert(package_name.clone(), UnavailablePackage::NoIndex);

                        return Ok(None);
                    }
                    VersionsResponse::Offline => {
                        self.unavailable_packages
                            .insert(package_name.clone(), UnavailablePackage::Offline);

                        return Ok(None);
                    }
                    VersionsResponse::NotFound => {
                        self.unavailable_packages
                            .insert(package_name.clone(), UnavailablePackage::NotFound);

                        return Ok(None);
                    }
                };

                // Try to find a compatible version. If there aren't any compatible versions,
                // short-circuit.
                let Some(candidate) = self.selector.select(
                    &package_name,
                    &range,
                    version_map,
                    &self.preferences,
                    self.installed_packages,
                    &self.exclusions,
                ) else {
                    return Ok(None);
                };

                // If there is not a compatible distribution, short-circuit.
                let Some(dist) = candidate.compatible() else {
                    return Ok(None);
                };

                // Emit a request to fetch the metadata for this version.
                if self.index.distributions.register(candidate.package_id()) {
                    let dist = dist.for_resolution().to_owned();

                    let response = match dist {
                        ResolvedDist::Installable(dist) => {
                            let (metadata, precise) = self
                                .provider
                                .get_or_build_wheel_metadata(&dist)
                                .boxed()
                                .await
                                .map_err(|err| match dist.clone() {
                                    Dist::Built(BuiltDist::Path(built_dist)) => {
                                        ResolveError::Read(Box::new(built_dist), err)
                                    }
                                    Dist::Source(SourceDist::Path(source_dist)) => {
                                        ResolveError::Build(Box::new(source_dist), err)
                                    }
                                    Dist::Built(built_dist) => {
                                        ResolveError::Fetch(Box::new(built_dist), err)
                                    }
                                    Dist::Source(source_dist) => {
                                        ResolveError::FetchAndBuild(Box::new(source_dist), err)
                                    }
                                })?;
                            Response::Dist {
                                dist,
                                metadata,
                                precise,
                            }
                        }
                        ResolvedDist::Installed(dist) => {
                            let metadata = dist.metadata().map_err(|err| {
                                ResolveError::ReadInstalled(Box::new(dist.clone()), err)
                            })?;
                            Response::Installed { dist, metadata }
                        }
                    };

                    Ok(Some(response))
                } else {
                    Ok(None)
                }
            }
        }
    }

    fn on_progress(&self, package: &PubGrubPackage, version: &Version) {
        if let Some(reporter) = self.reporter.as_ref() {
            match package {
                PubGrubPackage::Root(_) => {}
                PubGrubPackage::Python(_) => {}
                PubGrubPackage::Package {
                    name: package_name,
                    url: Some(url),
                    ..
                } => {
                    reporter.on_progress(package_name, &VersionOrUrl::Url(url));
                }
                PubGrubPackage::Package {
                    name: package_name,
                    url: None,
                    ..
                } => {
                    reporter.on_progress(package_name, &VersionOrUrl::Version(version));
                }
            }
        }
    }

    fn on_complete(&self) {
        if let Some(reporter) = self.reporter.as_ref() {
            reporter.on_complete();
        }
    }
}

// struct Fork<
// 'a,
// Provider: ResolverProvider,
// InstalledPackages: InstalledPackagesProvider + Send + Sync,
// > {
// resolver: Resolver<'a, Provider, InstalledPackages>,
// state: ResolverState,
// }

// struct Fork {
// state: ResolverState,
// deps: Dependencies,
// }

#[derive(Clone)]
pub(crate) struct ResolverState {
    pub(crate) pubgrub: State<UvDependencyProvider>,
    pub(crate) next: PubGrubPackage,
    pub(crate) pins: FilePins,
    pub(crate) priorities: PubGrubPriorities,
    pub(crate) added_dependencies: FxHashMap<PubGrubPackage, FxHashSet<Version>>,
    pub(crate) markers: FxHashMap<PubGrubPackage, MarkerTree>,
}

impl ResolverState {
    fn into_resolution(self) -> Resolution {
        let packages = self.pubgrub.partial_solution.extract_solution();
        let mut dependencies = FxHashMap::default();
        for (package, version) in &packages {
            for id in &self.pubgrub.incompatibilities[package] {
                let pubgrub::solver::Kind::FromDependencyOf(
                    ref self_package,
                    ref self_range,
                    ref dep_package,
                    ref dep_range,
                ) = self.pubgrub.incompatibility_store[*id].kind
                else {
                    continue;
                };
                if package != self_package {
                    continue;
                }
                let PubGrubPackage::Package {
                    name: ref self_name,
                    ..
                } = self_package
                else {
                    continue;
                };
                let PubGrubPackage::Package {
                    name: ref dep_name, ..
                } = dep_package
                else {
                    continue;
                };
                if self_name == dep_name {
                    continue;
                }
                if self_range.contains(version) {
                    let names = ResolutionDependencyNames {
                        from: self_name.clone(),
                        to: dep_name.clone(),
                    };
                    let versions = ResolutionDependencyVersions {
                        from: self_range.clone(),
                        to: dep_range.clone(),
                    };
                    dependencies
                        .entry(names)
                        .or_insert_with(|| vec![])
                        .push(versions);
                }
            }
        }
        let packages = packages
            .into_iter()
            .map(|(package, version)| {
                let marker = self.markers.get(&package).cloned();
                let map = FxHashMap::from_iter([(version, marker)]);
                (package, map)
            })
            .collect();
        Resolution {
            packages,
            dependencies,
            pins: self.pins,
        }
    }
}

#[derive(Debug, Default)]
pub(crate) struct Resolution {
    pub(crate) packages: FxHashMap<PubGrubPackage, FxHashMap<Version, Option<MarkerTree>>>,
    pub(crate) dependencies:
        FxHashMap<ResolutionDependencyNames, Vec<ResolutionDependencyVersions>>,
    pub(crate) pins: FilePins,
}

#[derive(Clone, Debug, Eq, Hash, PartialEq)]
pub(crate) struct ResolutionDependencyNames {
    pub(crate) from: PackageName,
    pub(crate) to: PackageName,
}

#[derive(Clone, Debug, Eq, Hash, PartialEq)]
pub(crate) struct ResolutionDependencyVersions {
    pub(crate) from: Range<Version>,
    pub(crate) to: Range<Version>,
}

impl Resolution {
    fn union(&mut self, other: Resolution) {
        for (package, versions) in other.packages {
            let entry = self.packages.entry(package);
            let mut map: &mut FxHashMap<Version, Option<MarkerTree>> = entry.or_default();
            for (version, marker) in versions {
                if !map.contains_key(&version) || map[&version].is_none() {
                    map.insert(version, marker);
                } else {
                    if let Some(marker) = marker {
                        map.get_mut(&version).unwrap().as_mut().unwrap().or(marker);
                    }
                }
            }
            // self.packages.entry(package).or_default().extend(versions);
        }
        for (names, ranges) in other.dependencies {
            self.dependencies.entry(names).or_default().extend(ranges);
        }
        self.pins.union(other.pins);
    }
}

/// Fetch the metadata for an item
#[derive(Debug)]
#[allow(clippy::large_enum_variant)]
pub(crate) enum Request {
    /// A request to fetch the metadata for a package.
    Package(PackageName),
    /// A request to fetch the metadata for a built or source distribution.
    Dist(Dist),
    /// A request to fetch the metadata from an already-installed distribution.
    Installed(InstalledDist),
    /// A request to pre-fetch the metadata for a package and the best-guess distribution.
    Prefetch(PackageName, Range<Version>),
}

impl Display for Request {
    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::Package(package_name) => {
                write!(f, "Versions {package_name}")
            }
            Self::Dist(dist) => {
                write!(f, "Metadata {dist}")
            }
            Self::Installed(dist) => {
                write!(f, "Installed metadata {dist}")
            }
            Self::Prefetch(package_name, range) => {
                write!(f, "Prefetch {package_name} {range}")
            }
        }
    }
}

#[derive(Debug)]
#[allow(clippy::large_enum_variant)]
enum Response {
    /// The returned metadata for a package hosted on a registry.
    Package(PackageName, VersionsResponse),
    /// The returned metadata for a distribution.
    Dist {
        dist: Dist,
        metadata: Metadata23,
        precise: Option<Url>,
    },
    /// The returned metadata for an already-installed distribution.
    Installed {
        dist: InstalledDist,
        metadata: Metadata23,
    },
}

/// An enum used by [`DependencyProvider`] that holds information about package dependencies.
/// For each [Package] there is a set of versions allowed as a dependency.
#[derive(Clone)]
enum Dependencies {
    /// Package dependencies are not available.
    Unavailable(String),
    /// Container for all available package versions.
    Available(Vec<(PubGrubPackage, Range<Version>, Option<MarkerTree>)>),
}

fn uncapitalize<T: AsRef<str>>(string: T) -> String {
    let mut chars = string.as_ref().chars();
    match chars.next() {
        None => String::new(),
        Some(first) => first.to_lowercase().chain(chars).collect(),
    }
}