refactor(network): use async for registry network operations

Cargo.toml

@@ -23,6 +23,7 @@ anstyle = "1.0.13"
 anstyle-hyperlink = "1.0.1"
 anstyle-progress = "0.1.0"
 anyhow = "1.0.102"
+async-trait = "0.1.89"
 base64 = "0.22.1"
 blake3 = "1.8.3"
 build-rs = { version = "0.3.4", path = "crates/build-rs" }
@@ -49,6 +50,8 @@ curl = "0.4.49"
 curl-sys = "=0.4.83"
 filetime = "0.2.27"
 flate2 = { version = "1.1.9", default-features = false, features = ["zlib-rs"] }
+futures = { version = "0.3", default-features = false, features = ["std", "executor"]}
+futures-timer = "3.0.3"
 git2 = "0.20.4"
 git2-curl = "0.21.0"
 # When updating this, also see if `gix-transport` further down needs updating or some auth-related tests will fail.
@@ -60,6 +63,7 @@ heck = "0.5.0"
 hex = "0.4.3"
 hmac = "0.12.1"
 home = "0.5.12"
+http = "1.4.0"
 http-auth = { version = "0.1.10", default-features = false }
 ignore = "0.4.25"
 im-rc = "15.1.0"
@@ -164,6 +168,7 @@ anstyle.workspace = true
 anstyle-hyperlink = { workspace = true, features = ["file"] }
 anstyle-progress.workspace = true
 anyhow.workspace = true
+async-trait.workspace = true
 base64.workspace = true
 blake3.workspace = true
 cargo-credential.workspace = true
@@ -178,6 +183,8 @@ curl = { workspace = true, features = ["http2"] }
 curl-sys.workspace = true
 filetime.workspace = true
 flate2.workspace = true
+futures.workspace = true
+futures-timer.workspace = true
 git2.workspace = true
 git2-curl.workspace = true
 gix.workspace = true
@@ -186,6 +193,7 @@ heck.workspace = true
 hex.workspace = true
 hmac.workspace = true
 home.workspace = true
+http.workspace = true
 http-auth.workspace = true
 ignore.workspace = true
 im-rc.workspace = true

crates/resolver-tests/src/lib.rs

@@ -7,10 +7,10 @@
 pub mod helpers;
 pub mod sat;
 
+use std::cell::RefCell;
 use std::cmp::{max, min};
 use std::collections::{BTreeMap, HashSet};
 use std::fmt;
-use std::task::Poll;
 use std::time::Instant;
 
 use cargo::core::Resolve;
@@ -131,15 +131,15 @@ pub fn resolve_with_global_context_raw(
 ) -> CargoResult<Resolve> {
     struct MyRegistry<'a> {
         list: &'a [Summary],
-        used: HashSet<PackageId>,
+        used: RefCell<HashSet<PackageId>>,
     }
     impl<'a> Registry for MyRegistry<'a> {
-        fn query(
-            &mut self,
+        async fn query(
+            &self,
             dep: &Dependency,
             kind: QueryKind,
             f: &mut dyn FnMut(IndexSummary),
-        ) -> Poll<CargoResult<()>> {
+        ) -> CargoResult<()> {
             for summary in self.list.iter() {
                 let matched = match kind {
                     QueryKind::Exact => dep.matches(summary),
@@ -148,11 +148,11 @@ pub fn resolve_with_global_context_raw(
                     QueryKind::Normalized => true,
                 };
                 if matched {
-                    self.used.insert(summary.package_id());
+                    self.used.borrow_mut().insert(summary.package_id());
                     f(IndexSummary::Candidate(summary.clone()));
                 }
             }
-            Poll::Ready(Ok(()))
+            Ok(())
         }
 
         fn describe_source(&self, _src: SourceId) -> String {
@@ -162,22 +162,18 @@ pub fn resolve_with_global_context_raw(
         fn is_replaced(&self, _src: SourceId) -> bool {
             false
         }
-
-        fn block_until_ready(&mut self) -> CargoResult<()> {
-            Ok(())
-        }
     }
     impl<'a> Drop for MyRegistry<'a> {
         fn drop(&mut self) {
-            if std::thread::panicking() && self.list.len() != self.used.len() {
+            if std::thread::panicking() && self.list.len() != self.used.get_mut().len() {
                 // we found a case that causes a panic and did not use all of the input.
                 // lets print the part of the input that was used for minimization.
                 eprintln!(
                     "Part used before drop: {:?}",
                     PrettyPrintRegistry(
                         self.list
                             .iter()
-                            .filter(|s| { self.used.contains(&s.package_id()) })
+                            .filter(|s| { self.used.get_mut().contains(&s.package_id()) })
                             .cloned()
                             .collect()
                     )
@@ -187,7 +183,7 @@ pub fn resolve_with_global_context_raw(
     }
     let mut registry = MyRegistry {
         list: registry,
-        used: HashSet::new(),
+        used: RefCell::new(HashSet::new()),
     };
 
     let root_summary =

crates/xtask-bump-check/Cargo.toml

@@ -9,6 +9,7 @@ anyhow.workspace = true
 cargo.workspace = true
 cargo-util.workspace = true
 clap.workspace = true
+futures.workspace = true
 git2.workspace = true
 semver.workspace = true
 tracing-subscriber.workspace = true

crates/xtask-bump-check/src/xtask.rs

@@ -15,7 +15,6 @@
 use std::collections::HashMap;
 use std::fmt::Write;
 use std::fs;
-use std::task;
 
 use cargo::CargoResult;
 use cargo::core::Package;
@@ -433,15 +432,9 @@ fn check_crates_io<'a>(
         let current = member.version();
         let version_req = format!(">={current}");
         let query = Dependency::parse(*name, Some(&version_req), source_id)?;
-        let possibilities = loop {
-            // Exact to avoid returning all for path/git
-            match registry.query_vec(&query, QueryKind::Exact) {
-                task::Poll::Ready(res) => {
-                    break res?;
-                }
-                task::Poll::Pending => registry.block_until_ready()?,
-            }
-        };
+        // Exact to avoid returning all for path/git
+        let possibilities =
+            futures::executor::block_on(registry.query_vec(&query, QueryKind::Exact))?;
         if possibilities.is_empty() {
             tracing::trace!("dep `{name}` has no version greater than or equal to `{current}`");
         } else {

src/cargo/core/compiler/future_incompat.rs

@@ -40,11 +40,11 @@ use crate::sources::source::QueryKind;
 use crate::util::CargoResult;
 use crate::util::cache_lock::CacheLockMode;
 use anyhow::{Context, bail, format_err};
+use futures::stream::FuturesUnordered;
 use serde::{Deserialize, Serialize};
 use std::collections::{BTreeMap, BTreeSet, HashMap, HashSet};
 use std::fmt::Write as _;
 use std::io::{Read, Write};
-use std::task::Poll;
 
 pub const REPORT_PREAMBLE: &str = "\
 The following warnings were discovered during the build. These warnings are an
@@ -308,15 +308,15 @@ fn get_updates(ws: &Workspace<'_>, package_ids: &BTreeSet<PackageId>) -> Option<
         .ok()?;
     // Create a set of updated registry sources.
     let map = SourceConfigMap::new(ws.gctx()).ok()?;
-    let mut package_ids: BTreeSet<_> = package_ids
+    let package_ids: BTreeSet<_> = package_ids
         .iter()
         .filter(|pkg_id| pkg_id.source_id().is_registry())
         .collect();
     let source_ids: HashSet<_> = package_ids
         .iter()
         .map(|pkg_id| pkg_id.source_id())
         .collect();
-    let mut sources: HashMap<_, _> = source_ids
+    let sources: HashMap<_, _> = source_ids
         .into_iter()
         .filter_map(|sid| {
             let source = map.load(sid, &HashSet::new()).ok()?;
@@ -325,28 +325,20 @@ fn get_updates(ws: &Workspace<'_>, package_ids: &BTreeSet<PackageId>) -> Option<
         .collect();
 
     // Query the sources for new versions, mapping `package_ids` into `summaries`.
-    let mut summaries = Vec::new();
-    while !package_ids.is_empty() {
-        package_ids.retain(|&pkg_id| {
-            let Some(source) = sources.get_mut(&pkg_id.source_id()) else {
-                return false;
-            };
-            let Ok(dep) = Dependency::parse(pkg_id.name(), None, pkg_id.source_id()) else {
-                return false;
-            };
-            match source.query_vec(&dep, QueryKind::Exact) {
-                Poll::Ready(Ok(sum)) => {
-                    summaries.push((pkg_id, sum));
-                    false
-                }
-                Poll::Ready(Err(_)) => false,
-                Poll::Pending => true,
-            }
-        });
-        for (_, source) in sources.iter_mut() {
-            source.block_until_ready().ok()?;
+    let pending = FuturesUnordered::new();
+    for pkg_id in package_ids {
+        if let Some(source) = sources.get(&pkg_id.source_id())
+            && let Ok(dep) = Dependency::parse(pkg_id.name(), None, pkg_id.source_id())
+        {
+            pending.push(async move {
+                let sum = source.query_vec(&dep, QueryKind::Exact).await.ok()?;
+                Some((pkg_id, sum))
+            });
         }
     }
+    let summaries = futures::executor::block_on_stream(pending)
+        .flatten()
+        .collect::<Vec<_>>();
 
     let mut updates = String::new();
     for (pkg_id, summaries) in summaries {

src/cargo/core/package.rs

@@ -1,5 +1,5 @@
 use std::cell::OnceCell;
-use std::cell::{Cell, Ref, RefCell, RefMut};
+use std::cell::{Cell, Ref, RefCell};
 use std::cmp::Ordering;
 use std::collections::{BTreeMap, BTreeSet, HashMap, HashSet};
 use std::fmt;
@@ -663,10 +663,6 @@ impl<'gctx> PackageSet<'gctx> {
         self.sources.borrow()
     }
 
-    pub fn sources_mut(&self) -> RefMut<'_, SourceMap<'gctx>> {
-        self.sources.borrow_mut()
-    }
-
     /// Merge the given set into self.
     pub fn add_set(&mut self, set: PackageSet<'gctx>) {
         assert!(!self.downloading.get());
@@ -707,9 +703,9 @@ impl<'a, 'gctx> Downloads<'a, 'gctx> {
         // Ask the original source for this `PackageId` for the corresponding
         // package. That may immediately come back and tell us that the package
         // is ready, or it could tell us that it needs to be downloaded.
-        let mut sources = self.set.sources.borrow_mut();
+        let sources = self.set.sources.borrow_mut();
         let source = sources
-            .get_mut(id.source_id())
+            .get(id.source_id())
             .ok_or_else(|| internal(format!("couldn't find source for `{}`", id)))?;
         let pkg = source
             .download(id)
@@ -924,9 +920,9 @@ impl<'a, 'gctx> Downloads<'a, 'gctx> {
 
         // Inform the original source that the download is finished which
         // should allow us to actually get the package and fill it in now.
-        let mut sources = self.set.sources.borrow_mut();
+        let sources = self.set.sources.borrow_mut();
         let source = sources
-            .get_mut(dl.id.source_id())
+            .get(dl.id.source_id())
             .ok_or_else(|| internal(format!("couldn't find source for `{}`", dl.id)))?;
         let start = Instant::now();
         let pkg = source.finish_download(dl.id, data)?;