Use async for registry network operations

Converts Poll into async. Some error messages are impacted
due to slightly different code paths being taken.

This is a major change to the sparse registry backend
in http_remote.rs.

Author: arlosi

crates/resolver-tests/src/lib.rs

@@ -11,7 +11,6 @@ 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;
@@ -135,12 +134,12 @@ pub fn resolve_with_global_context_raw(
         used: RefCell<HashSet<PackageId>>,
     }
     impl<'a> Registry for MyRegistry<'a> {
-        fn query(
+        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),
@@ -153,7 +152,7 @@ pub fn resolve_with_global_context_raw(
                     f(IndexSummary::Candidate(summary.clone()));
                 }
             }
-            Poll::Ready(Ok(()))
+            Ok(())
         }
 
         fn describe_source(&self, _src: SourceId) -> String {
@@ -163,10 +162,6 @@ pub fn resolve_with_global_context_raw(
         fn is_replaced(&self, _src: SourceId) -> bool {
             false
         }
-
-        fn block_until_ready(&self) -> CargoResult<()> {
-            Ok(())
-        }
     }
     impl<'a> Drop for MyRegistry<'a> {
         fn drop(&mut self) {

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;
@@ -444,15 +443,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,7 +308,7 @@ 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();
@@ -325,28 +325,18 @@ 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(&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() {
-            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 = crate::util::block_on_stream(pending).flatten();
 
     let mut updates = String::new();
     for (pkg_id, summaries) in summaries {