fix: Use the right channels when upgrading global packages

src/cli/global/common.rs

@@ -153,13 +153,13 @@ pub(super) fn channel_name_from_prefix(
 /// # Returns
 ///
 /// The package records (with dependencies records) for the given package MatchSpec
-pub fn load_package_records(
+pub fn load_package_records<'a>(
     package_matchspec: MatchSpec,
-    sparse_repodata: &IndexMap<(Channel, Platform), SparseRepoData>,
+    sparse_repodata: impl IntoIterator<Item = &'a SparseRepoData>,
 ) -> miette::Result<Vec<RepoDataRecord>> {
     let package_name = package_name(&package_matchspec)?;
     let available_packages =
-        SparseRepoData::load_records_recursive(sparse_repodata.values(), vec![package_name], None)
+        SparseRepoData::load_records_recursive(sparse_repodata, vec![package_name], None)
             .into_diagnostic()?;
     let virtual_packages = rattler_virtual_packages::VirtualPackage::current()
         .into_diagnostic()?

src/cli/global/install.rs

@@ -258,7 +258,7 @@ pub async fn execute(args: Args) -> miette::Result<()> {
     // Install the package(s)
     let mut executables = vec![];
     for (package_name, package_matchspec) in args.specs()? {
-        let records = load_package_records(package_matchspec, &sparse_repodata)?;
+        let records = load_package_records(package_matchspec, sparse_repodata.values())?;
 
         let (prefix_package, scripts, _) = globally_install_package(
             &package_name,

src/cli/global/upgrade.rs

@@ -1,12 +1,14 @@
 use std::collections::HashMap;
+use std::sync::Arc;
 use std::time::Duration;
 
 use clap::Parser;
 use indexmap::IndexMap;
 use indicatif::ProgressBar;
 use itertools::Itertools;
-use miette::IntoDiagnostic;
+use miette::{IntoDiagnostic, Report};
 use rattler_conda_types::{Channel, MatchSpec, PackageName, Platform};
+use tokio::task::JoinSet;
 
 use crate::config::Config;
 use crate::progress::{global_multi_progress, long_running_progress_style};
@@ -59,56 +61,77 @@ pub(super) async fn upgrade_packages(
     cli_channels: &[String],
     platform: &Platform,
 ) -> miette::Result<()> {
-    // Get channels and versions of globally installed packages
-    let mut installed_versions = HashMap::with_capacity(specs.len());
-    let mut channels = config.compute_channels(cli_channels).into_diagnostic()?;
-
-    for package_name in specs.keys() {
-        let prefix_record = find_installed_package(package_name).await?;
-        let last_installed_channel = Channel::from_str(
-            prefix_record.repodata_record.channel.clone(),
-            config.channel_config(),
-        )
-        .into_diagnostic()?;
-
-        channels.push(last_installed_channel);
-
-        let installed_version = prefix_record
-            .repodata_record
-            .package_record
-            .version
-            .into_version();
-        installed_versions.insert(package_name.clone(), installed_version);
+    let channel_cli = config.compute_channels(cli_channels).into_diagnostic()?;
+
+    // Get channels and version of globally installed packages in parallel
+    let mut channels = HashMap::with_capacity(specs.len());
+    let mut versions = HashMap::with_capacity(specs.len());
+    let mut set: JoinSet<Result<_, Report>> = JoinSet::new();
+    for package_name in specs.keys().cloned() {
+        let channel_config = config.channel_config().clone();
+        set.spawn(async move {
+            let p = find_installed_package(&package_name).await?;
+            let channel =
+                Channel::from_str(p.repodata_record.channel, &channel_config).into_diagnostic()?;
+            let version = p.repodata_record.package_record.version.into_version();
+            Ok((package_name, channel, version))
+        });
+    }
+    while let Some(data) = set.join_next().await {
+        let (package_name, channel, version) = data.into_diagnostic()??;
+        channels.insert(package_name.clone(), channel);
+        versions.insert(package_name, version);
     }
-    channels = channels.into_iter().unique().collect();
 
-    // Fetch sparse repodata
-    let (authenticated_client, sparse_repodata) =
-        get_client_and_sparse_repodata(&channels, *platform, &config).await?;
+    // Fetch sparse repodata across all channels
+    let all_channels = channels.values().chain(channel_cli.iter()).unique();
+    let (client, repodata) =
+        get_client_and_sparse_repodata(all_channels, *platform, &config).await?;
+
+    // Resolve environments in parallel
+    let mut set: JoinSet<Result<_, Report>> = JoinSet::new();
+    let repodata = Arc::new(repodata);
+    let channels = Arc::new(channels);
+    for (package_name, package_matchspec) in specs {
+        let repodata = Arc::clone(&repodata);
+        let channels = Arc::clone(&channels);
+        let channel_cli = channel_cli.clone();
+        set.spawn_blocking(move || {
+            // Filter repodata based on channels specific to the package (and from the CLI)
+            let specific_repodata = repodata.iter().filter_map(|((c, _), v)| {
+                if channel_cli.contains(c) || channels.get(&package_name).unwrap() == c {
+                    Some(v)
+                } else {
+                    None
+                }
+            });
+            let records = load_package_records(package_matchspec.clone(), specific_repodata)?;
+            Ok((package_name, package_matchspec, records))
+        });
+    }
 
     // Upgrade each package when relevant
     let mut upgraded = false;
-    for (package_name, package_matchspec) in specs {
-        let matchspec_has_version = package_matchspec.version.is_some();
-        let records = load_package_records(package_matchspec, &sparse_repodata)?;
-        let package_record = records
+    while let Some(data) = set.join_next().await {
+        let (package_name, package_matchspec, records) = data.into_diagnostic()??;
+        let toinstall_version = records
             .iter()
             .find(|r| r.package_record.name == package_name)
+            .map(|p| p.package_record.version.version().to_owned())
             .ok_or_else(|| {
                 miette::miette!(
                     "Package {} not found in the specified channels",
                     package_name.as_normalized()
                 )
             })?;
-        let toinstall_version = package_record.package_record.version.version().to_owned();
-        let installed_version = installed_versions
+        let installed_version = versions
             .get(&package_name)
             .expect("should have the installed version")
             .to_owned();
 
         // Perform upgrade if a specific version was requested
         // OR if a more recent version is available
-        if matchspec_has_version || toinstall_version > installed_version {
+        if package_matchspec.version.is_some() || toinstall_version > installed_version {
             let message = format!(
                 "{} v{} -> v{}",
                 package_name.as_normalized(),
@@ -124,13 +147,7 @@ pub(super) async fn upgrade_packages(
                 console::style("Updating").green(),
                 message
             ));
-            globally_install_package(
-                &package_name,
-                records,
-                authenticated_client.clone(),
-                platform,
-            )
-            .await?;
+            globally_install_package(&package_name, records, client.clone(), platform).await?;
             pb.finish_with_message(format!("{} {}", console::style("Updated").green(), message));
             upgraded = true;
         }