lux_lib/operations/install/

mod.rs

use std::{collections::HashMap, io, sync::Arc};

use crate::{
    build::{Build, BuildBehaviour, BuildError, RemotePackageSourceSpec, SrcRockSource},
    config::Config,
    lockfile::{
        FlushLockfileError, LocalPackage, LocalPackageId, LockConstraint, Lockfile, OptState,
        PinnedState, ReadWrite,
    },
    lua_installation::{LuaInstallation, LuaInstallationError},
    lua_rockspec::BuildBackendSpec,
    lua_version::LuaVersionUnset,
    luarocks::{
        install_binary_rock::{BinaryRockInstall, InstallBinaryRockError},
        luarocks_installation::{LuaRocksError, LuaRocksInstallError, LuaRocksInstallation},
    },
    operations::resolve::{Resolve, ResolveDependenciesError},
    package::{PackageName, PackageNameList},
    remote_package_db::{RemotePackageDB, RemotePackageDBError, RemotePackageDbIntegrityError},
    rockspec::Rockspec,
    tree::{self, InstallTree, Tree, TreeError},
    workspace::{Workspace, WorkspaceTreeError},
};

pub use crate::operations::install::spec::PackageInstallSpec;

use super::{DownloadedRockspec, RemoteRockDownload};
use bon::Builder;
use bytes::Bytes;
use futures::StreamExt;
use itertools::Itertools;
use miette::Diagnostic;
use thiserror::Error;

use tracing::{info_span, span, Instrument};
pub mod spec;

/// A rocks package installer, providing fine-grained control
/// over how packages should be installed.
/// Can install multiple packages in parallel.
#[derive(Builder)]
#[builder(start_fn = new, finish_fn(name = _build, vis = ""))]
pub struct Install<'a, T>
where
    T: InstallTree + Clone + Send + Sync,
{
    #[builder(start_fn)]
    config: &'a Config,
    #[builder(field)]
    packages: Vec<PackageInstallSpec>,
    #[builder(setters(name = "_tree", vis = ""))]
    tree: T,
    package_db: Option<RemotePackageDB>,
}

impl<'a, State> InstallBuilder<'a, Tree, State>
where
    State: install_builder::State,
{
    pub fn workspace(
        self,
        workspace: &'a Workspace,
    ) -> Result<InstallBuilder<'a, Tree, install_builder::SetTree<State>>, WorkspaceTreeError>
    where
        State::Tree: install_builder::IsUnset,
    {
        let config = self.config;
        Ok(self._tree(workspace.tree(config)?))
    }
}

impl<'a, T, State> InstallBuilder<'a, T, State>
where
    State: install_builder::State,
    T: InstallTree + Clone + Send + Sync,
{
    pub fn tree(self, tree: T) -> InstallBuilder<'a, T, install_builder::SetTree<State>>
    where
        State::Tree: install_builder::IsUnset,
    {
        self._tree(tree)
    }

    pub fn packages(self, packages: Vec<PackageInstallSpec>) -> Self {
        Self { packages, ..self }
    }

    pub fn package(self, package: PackageInstallSpec) -> Self {
        Self {
            packages: self
                .packages
                .into_iter()
                .chain(std::iter::once(package))
                .collect(),
            ..self
        }
    }
}

impl<State, T> InstallBuilder<'_, T, State>
where
    State: install_builder::State + install_builder::IsComplete,
    T: InstallTree + Clone + Send + Sync + 'static,
{
    /// Install the packages.
    pub async fn install(self) -> Result<Vec<LocalPackage>, InstallError> {
        let install_built = self._build();
        if install_built.packages.is_empty() {
            return Ok(Vec::default());
        }
        let count = install_built.packages.len();
        let span = if count > 1 {
            info_span!("Installing", count,)
        } else {
            let install_spec = &install_built.packages[0];
            info_span!("Installing", package = install_spec.package.to_string(),)
        };
        let _enter = span.enter();
        let package_db = match install_built.package_db {
            Some(db) => db,
            None => RemotePackageDB::from_config(install_built.config).await?,
        };

        let duplicate_entrypoints = install_built
            .packages
            .iter()
            .filter(|pkg| pkg.entry_type == tree::EntryType::Entrypoint)
            .map(|pkg| pkg.package.name())
            .duplicates()
            .cloned()
            .collect_vec();

        if !duplicate_entrypoints.is_empty() {
            return Err(InstallError::DuplicateEntrypoints(PackageNameList::new(
                duplicate_entrypoints,
            )));
        }

        install_impl(
            install_built.packages,
            Arc::new(package_db),
            install_built.config,
            &install_built.tree,
        )
        .await
    }
}

#[derive(Error, Debug, Diagnostic)]
pub enum InstallError {
    #[error("unable to resolve dependencies:\n{0}")]
    #[diagnostic(forward(0))]
    ResolveDependencies(#[from] ResolveDependenciesError),
    #[error(transparent)]
    #[diagnostic(transparent)]
    LuaVersionUnset(#[from] LuaVersionUnset),
    #[error(transparent)]
    #[diagnostic(transparent)]
    LuaInstallation(#[from] LuaInstallationError),
    #[error(transparent)]
    #[diagnostic(transparent)]
    FlushLockfile(#[from] FlushLockfileError),
    #[error(transparent)]
    #[diagnostic(transparent)]
    Tree(#[from] TreeError),
    #[error(transparent)]
    #[diagnostic(transparent)]
    WorkspaceTree(#[from] WorkspaceTreeError),
    #[error("error instantiating LuaRocks compatibility layer:\n{0}")]
    #[diagnostic(forward(0))]
    LuaRocks(#[from] LuaRocksError),
    #[error("error installing LuaRocks compatibility layer:\n{0}")]
    #[diagnostic(forward(0))]
    LuaRocksInstall(#[from] LuaRocksInstallError),
    #[error("failed to build {0}: {1}")]
    Build(PackageName, BuildError),
    #[error("failed to install build depencency {0}:\n{1}")]
    BuildDependency(PackageName, BuildError),
    #[error("error initialising remote package DB:\n{0}")]
    #[diagnostic(forward(0))]
    RemotePackageDB(#[from] RemotePackageDBError),
    #[error("failed to install pre-built rock {0}:\n{1}")]
    InstallBinaryRock(PackageName, InstallBinaryRockError),
    #[error("integrity error for package '{package}'")]
    Integrity {
        package: PackageName,
        #[diagnostic_source]
        err: RemotePackageDbIntegrityError,
    },
    #[error("cannot install duplicate entrypoints:\n{0}")]
    DuplicateEntrypoints(PackageNameList),
}

// TODO(vhyrro): This function has too many arguments. Refactor it.
#[allow(clippy::too_many_arguments)]
async fn install_impl<T>(
    packages: Vec<PackageInstallSpec>,
    package_db: Arc<RemotePackageDB>,
    config: &Config,
    tree: &T,
) -> Result<Vec<LocalPackage>, InstallError>
where
    T: InstallTree + Clone + Send + Sync + 'static,
{
    let (dep_tx, mut dep_rx) = tokio::sync::mpsc::unbounded_channel();
    let (build_dep_tx, mut build_dep_rx) = tokio::sync::mpsc::unbounded_channel();

    let lockfile = tree.lockfile()?;
    let build_lockfile = tree.build_tree(config)?.lockfile()?;

    Resolve::new()
        .dependencies_tx(dep_tx)
        .build_dependencies_tx(build_dep_tx)
        .packages(packages)
        .package_db(package_db.clone())
        .lockfile(Arc::new(lockfile.clone()))
        .build_lockfile(Arc::new(build_lockfile.clone()))
        .config(config)
        .get_all_dependencies()
        .await?;

    let lua = Arc::new(LuaInstallation::new_from_config(config).await?);

    // We have to install transitive build dependencies sequentially
    while let Some(build_dep_spec) = build_dep_rx.recv().await {
        let rockspec = build_dep_spec.downloaded_rock.rockspec();
        let package = rockspec.package().clone();
        let span = info_span!(
            "Installing build dependency",
            package = package.to_string(),
            version = rockspec.version().to_string()
        );
        async {
            let build_tree = tree.build_tree(config)?;
            // We have to write to the build tree's lockfile after each build,
            // so that each transitive build dependency is available for the
            // next build dependencies that may depend on it.
            let mut build_lockfile = build_tree.lockfile()?.write_guard();
            let pkg = Build::new()
                .rockspec(rockspec)
                .lua(&lua)
                .tree(&build_tree)
                .entry_type(tree::EntryType::Entrypoint)
                .config(config)
                .constraint(build_dep_spec.spec.constraint())
                .behaviour(build_dep_spec.build_behaviour)
                .build()
                .await
                .map_err(|err| InstallError::BuildDependency(package, err))?;
            build_lockfile.add_entrypoint(&pkg);
            Ok::<_, InstallError>(())
        }
        .instrument(span)
        .await?;
    }

    let mut all_packages = HashMap::with_capacity(dep_rx.len());
    while let Some(dep) = dep_rx.recv().await {
        all_packages.insert(dep.spec.id(), dep);
    }

    let installed_packages =
        futures::stream::iter(all_packages.clone().into_values().map(|install_spec| {
            let downloaded_rock = install_spec.downloaded_rock;
            let config = config.clone();
            let tree = tree.clone();
            let lua = lua.clone();

            tokio::spawn({
                async move {
                    let pkg = match downloaded_rock {
                        RemoteRockDownload::RockspecOnly { rockspec_download } => {
                            install_rockspec(
                                rockspec_download,
                                None,
                                install_spec.spec.constraint(),
                                install_spec.build_behaviour,
                                install_spec.pin,
                                install_spec.opt,
                                install_spec.entry_type,
                                &lua,
                                &tree,
                                &config,
                            )
                            .await?
                        }
                        RemoteRockDownload::BinaryRock {
                            rockspec_download,
                            packed_rock,
                        } => {
                            install_binary_rock(
                                rockspec_download,
                                packed_rock,
                                install_spec.spec.constraint(),
                                install_spec.build_behaviour,
                                install_spec.pin,
                                install_spec.opt,
                                install_spec.entry_type,
                                &config,
                                &tree,
                            )
                            .await?
                        }
                        RemoteRockDownload::SrcRock {
                            rockspec_download,
                            src_rock,
                            source_url,
                        } => {
                            let src_rock_source = SrcRockSource {
                                bytes: src_rock,
                                source_url,
                            };
                            install_rockspec(
                                rockspec_download,
                                Some(src_rock_source),
                                install_spec.spec.constraint(),
                                install_spec.build_behaviour,
                                install_spec.pin,
                                install_spec.opt,
                                install_spec.entry_type,
                                &lua,
                                &tree,
                                &config,
                            )
                            .await?
                        }
                    };

                    Ok::<_, InstallError>((pkg.id(), (pkg, install_spec.entry_type)))
                }
                .instrument(tracing::trace_span!("install_worker"))
            })
        }))
        .buffered(config.max_jobs())
        .collect::<Vec<_>>()
        .instrument(tracing::trace_span!("install_collector"))
        .await
        .into_iter()
        .flatten()
        .try_collect::<_, HashMap<LocalPackageId, (LocalPackage, tree::EntryType)>, _>()?;

    let write_dependency = |lockfile: &mut Lockfile<ReadWrite>,
                            id: &LocalPackageId,
                            pkg: &LocalPackage,
                            entry_type: tree::EntryType|
     -> io::Result<()> {
        if entry_type == tree::EntryType::Entrypoint {
            lockfile.add_entrypoint(pkg);
        }

        for dependency_id in all_packages
            .get(id)
            .map(|pkg| pkg.spec.dependencies())
            .unwrap_or_default()
            .into_iter()
        {
            lockfile.add_dependency(
                pkg,
                installed_packages
                    .get(dependency_id)
                    .map(|(pkg, _)| pkg)
                    .ok_or(io::Error::other(
                        r#"
error writing dependencies to the lockfile.
A required dependency was not installed correctly.
This is likely because an install thread panicked and was interrupted unexpectedly.

[THIS IS A BUG!]
"#,
                    ))?,
            );
        }
        Ok(())
    };

    lockfile.map_then_flush(|lockfile| {
        for (id, (pkg, is_entrypoint)) in installed_packages.iter() {
            write_dependency(lockfile, id, pkg, *is_entrypoint)?;
        }
        Ok::<_, io::Error>(())
    })?;

    Ok(installed_packages
        .into_values()
        .map(|(pkg, _)| pkg)
        .collect_vec())
}

#[allow(clippy::too_many_arguments)]
async fn install_rockspec<T>(
    rockspec_download: DownloadedRockspec,
    src_rock_source: Option<SrcRockSource>,
    constraint: LockConstraint,
    behaviour: BuildBehaviour,
    pin: PinnedState,
    opt: OptState,
    entry_type: tree::EntryType,
    lua: &LuaInstallation,
    tree: &T,
    config: &Config,
) -> Result<LocalPackage, InstallError>
where
    T: InstallTree + Sync,
{
    let package = rockspec_download.rockspec.package().clone();
    let rockspec = rockspec_download.rockspec;
    let span = info_span!(
        "Installing",
        package = package.to_string(),
        version = rockspec.version().to_string(),
    );
    let _enter = span.enter();
    let source = rockspec_download.source;

    if let Some(BuildBackendSpec::LuaRock(_)) = &rockspec.build().current_platform().build_backend {
        let luarocks_tree = tree.build_tree(config)?;
        let luarocks = LuaRocksInstallation::new(config, luarocks_tree)?;
        luarocks.ensure_installed(lua).await?;
    }

    let source_spec = match src_rock_source {
        Some(src_rock_source) => RemotePackageSourceSpec::SrcRock(src_rock_source),
        None => RemotePackageSourceSpec::RockSpec(rockspec_download.source_url),
    };

    let pkg = Build::new()
        .rockspec(&rockspec)
        .lua(lua)
        .tree(tree)
        .entry_type(entry_type)
        .config(config)
        .pin(pin)
        .opt(opt)
        .constraint(constraint)
        .behaviour(behaviour)
        .source(source)
        .source_spec(source_spec)
        .build()
        .await
        .map_err(|err| InstallError::Build(package, err))?;
    Ok(pkg)
}

#[allow(clippy::too_many_arguments)]
async fn install_binary_rock(
    rockspec_download: DownloadedRockspec,
    packed_rock: Bytes,
    constraint: LockConstraint,
    behaviour: BuildBehaviour,
    pin: PinnedState,
    opt: OptState,
    entry_type: tree::EntryType,
    config: &Config,
    tree: &impl InstallTree,
) -> Result<LocalPackage, InstallError> {
    let rockspec = rockspec_download.rockspec;
    let package = rockspec.package().clone();
    let span = span!(
        tracing::Level::INFO,
        "Installing (pre-built)",
        package = package.to_string(),
        version = rockspec.version().to_string(),
    );
    let _enter = span.enter();
    let pkg = BinaryRockInstall::new(
        &rockspec,
        rockspec_download.source,
        packed_rock,
        entry_type,
        config,
        tree,
    )
    .pin(pin)
    .opt(opt)
    .constraint(constraint)
    .behaviour(behaviour)
    .install()
    .await
    .map_err(|err| InstallError::InstallBinaryRock(package, err))?;
    Ok(pkg)
}