async registry_api & repository-stats updater

Cargo.toml

@@ -35,7 +35,7 @@ crates-index = { version = "2.2.0", default-features = false, features = ["git",
 rayon = "1.6.1"
 num_cpus = "1.15.0"
 crates-index-diff = { version = "21.0.0", features = [ "max-performance" ]}
-reqwest = { version = "0.11", features = ["blocking", "json"] } # TODO: Remove blocking when async is ready
+reqwest = { version = "0.11", features = ["json", "gzip"] } 
 semver = { version = "1.0.4", features = ["serde"] }
 slug = "0.1.1"
 r2d2 = "0.8"
@@ -124,6 +124,7 @@ kuchikiki = "0.8"
 rand = "0.8"
 mockito = "1.0.2"
 test-case = "3.0.0"
+reqwest = { version = "0.11", features = ["blocking", "json"] } 
 aws-smithy-types = "1.0.1"
 aws-smithy-runtime = {version = "1.0.1", features = ["client", "test-util"]}
 aws-smithy-http = "0.60.0"

src/bin/cratesfyi.rs

@@ -545,21 +545,20 @@ impl DatabaseSubcommand {
             }
 
             Self::UpdateRepositoryFields => {
-                ctx.repository_stats_updater()?.update_all_crates()?;
+                ctx.runtime()?
+                    .block_on(ctx.repository_stats_updater()?.update_all_crates())?;
             }
 
             Self::BackfillRepositoryStats => {
-                ctx.repository_stats_updater()?.backfill_repositories()?;
+                ctx.runtime()?
+                    .block_on(ctx.repository_stats_updater()?.backfill_repositories())?;
             }
 
-            Self::UpdateCrateRegistryFields { name } => {
-                let registry_data = ctx.registry_api()?.get_crate_data(&name)?;
-
-                ctx.runtime()?.block_on(async move {
-                    let mut conn = ctx.pool()?.get_async().await?;
-                    db::update_crate_data_in_database(&mut conn, &name, &registry_data).await
-                })?
-            }
+            Self::UpdateCrateRegistryFields { name } => ctx.runtime()?.block_on(async move {
+                let mut conn = ctx.pool()?.get_async().await?;
+                let registry_data = ctx.registry_api()?.get_crate_data(&name).await?;
+                db::update_crate_data_in_database(&mut conn, &name, &registry_data).await
+            })?,
 
             Self::AddDirectory { directory } => {
                 ctx.runtime()?

src/docbuilder/rustwide_builder.rs

@@ -536,8 +536,8 @@ impl RustwideBuilder {
 
                     let release_data = if !is_local {
                         match self
-                            .registry_api
-                            .get_release_data(name, version)
+                            .runtime
+                            .block_on(self.registry_api.get_release_data(name, version))
                             .with_context(|| {
                                 format!("could not fetch releases-data for {name}-{version}")
                             }) {
@@ -591,7 +591,10 @@ impl RustwideBuilder {
 
                     // Some crates.io crate data is mutable, so we proactively update it during a release
                     if !is_local {
-                        match self.registry_api.get_crate_data(name) {
+                        match self
+                            .runtime
+                            .block_on(self.registry_api.get_crate_data(name))
+                        {
                             Ok(crate_data) => self.runtime.block_on(
                                 update_crate_data_in_database(&mut async_conn, name, &crate_data),
                             )?,
@@ -887,8 +890,8 @@ impl RustwideBuilder {
     }
 
     fn get_repo(&self, metadata: &MetadataPackage) -> Result<Option<i32>> {
-        self.repository_stats_updater
-            .load_repository(metadata)
+        self.runtime
+            .block_on(self.repository_stats_updater.load_repository(metadata))
             .map_err(Into::into)
     }
 }

src/registry_api.rs

@@ -1,4 +1,4 @@
-use crate::{error::Result, utils::retry};
+use crate::{error::Result, utils::retry_async};
 use anyhow::{anyhow, Context};
 use chrono::{DateTime, Utc};
 use reqwest::header::{HeaderValue, ACCEPT, USER_AGENT};
@@ -16,7 +16,7 @@ const APP_USER_AGENT: &str = concat!(
 pub struct RegistryApi {
     api_base: Url,
     max_retries: u32,
-    client: reqwest::blocking::Client,
+    client: reqwest::Client,
 }
 
 #[derive(Debug)]
@@ -56,7 +56,7 @@ impl RegistryApi {
         .into_iter()
         .collect();
 
-        let client = reqwest::blocking::Client::builder()
+        let client = reqwest::Client::builder()
             .default_headers(headers)
             .build()?;
 
@@ -67,17 +67,19 @@ impl RegistryApi {
         })
     }
 
-    pub fn get_crate_data(&self, name: &str) -> Result<CrateData> {
+    pub async fn get_crate_data(&self, name: &str) -> Result<CrateData> {
         let owners = self
             .get_owners(name)
+            .await
             .context(format!("Failed to get owners for {name}"))?;
 
         Ok(CrateData { owners })
     }
 
-    pub(crate) fn get_release_data(&self, name: &str, version: &str) -> Result<ReleaseData> {
+    pub(crate) async fn get_release_data(&self, name: &str, version: &str) -> Result<ReleaseData> {
         let (release_time, yanked, downloads) = self
             .get_release_time_yanked_downloads(name, version)
+            .await
             .context(format!("Failed to get crate data for {name}-{version}"))?;
 
         Ok(ReleaseData {
@@ -88,7 +90,7 @@ impl RegistryApi {
     }
 
     /// Get release_time, yanked and downloads from the registry's API
-    fn get_release_time_yanked_downloads(
+    async fn get_release_time_yanked_downloads(
         &self,
         name: &str,
         version: &str,
@@ -117,11 +119,20 @@ impl RegistryApi {
             downloads: i32,
         }
 
-        let response: Response = retry(
-            || Ok(self.client.get(url.clone()).send()?.error_for_status()?),
+        let response: Response = retry_async(
+            || async {
+                Ok(self
+                    .client
+                    .get(url.clone())
+                    .send()
+                    .await?
+                    .error_for_status()?)
+            },
             self.max_retries,
-        )?
-        .json()?;
+        )
+        .await?
+        .json()
+        .await?;
 
         let version = Version::parse(version)?;
         let version = response
@@ -134,7 +145,7 @@ impl RegistryApi {
     }
 
     /// Fetch owners from the registry's API
-    fn get_owners(&self, name: &str) -> Result<Vec<CrateOwner>> {
+    async fn get_owners(&self, name: &str) -> Result<Vec<CrateOwner>> {
         let url = {
             let mut url = self.api_base.clone();
             url.path_segments_mut()
@@ -156,11 +167,20 @@ impl RegistryApi {
             login: Option<String>,
         }
 
-        let response: Response = retry(
-            || Ok(self.client.get(url.clone()).send()?.error_for_status()?),
+        let response: Response = retry_async(
+            || async {
+                Ok(self
+                    .client
+                    .get(url.clone())
+                    .send()
+                    .await?
+                    .error_for_status()?)
+            },
             self.max_retries,
-        )?
-        .json()?;
+        )
+        .await?
+        .json()
+        .await?;
 
         let result = response
             .users