Merge pull request #10047 from Turbo87/default-versions-cleanup

models/default_versions: Remove sync fns

Author: Turbo87

src/bin/crates-admin/default_versions.rs

@@ -1,5 +1,5 @@
 use anyhow::Context;
-use crates_io::models::{async_update_default_version, async_verify_default_version};
+use crates_io::models::{update_default_version, verify_default_version};
 use crates_io::{db, schema::crates};
 use diesel::prelude::*;
 use diesel_async::RunQueryDsl;
@@ -33,8 +33,8 @@ pub async fn run(command: Command) -> anyhow::Result<()> {
 
     for crate_id in crate_ids.into_iter().progress_with(pb.clone()) {
         let result = match command {
-            Command::Update => async_update_default_version(crate_id, &mut conn).await,
-            Command::Verify => async_verify_default_version(crate_id, &mut conn).await,
+            Command::Update => update_default_version(crate_id, &mut conn).await,
+            Command::Verify => verify_default_version(crate_id, &mut conn).await,
         };
 
         if let Err(error) = result {

src/bin/crates-admin/delete_version.rs

@@ -3,12 +3,12 @@ use anyhow::Context;
 use crates_io::models::update_default_version;
 use crates_io::schema::crates;
 use crates_io::storage::Storage;
-use crates_io::tasks::spawn_blocking;
 use crates_io::worker::jobs;
 use crates_io::{db, schema::versions};
 use crates_io_worker::BackgroundJob;
-use diesel::{Connection, ExpressionMethods, QueryDsl};
-use diesel_async::async_connection_wrapper::AsyncConnectionWrapper;
+use diesel::prelude::*;
+use diesel_async::scoped_futures::ScopedFutureExt;
+use diesel_async::{AsyncConnection, RunQueryDsl};
 
 #[derive(clap::Parser, Debug)]
 #[command(
@@ -36,17 +36,12 @@ pub async fn run(opts: Opts) -> anyhow::Result<()> {
 
     let store = Storage::from_environment();
 
-    let crate_id: i32 = {
-        use diesel_async::RunQueryDsl;
-
-        crates::table
-            .select(crates::id)
-            .filter(crates::name.eq(&opts.crate_name))
-            .first(&mut conn)
-            .await
-            .context("Failed to look up crate id from the database")
-    }?;
-
+    let crate_id: i32 = crates::table
+        .select(crates::id)
+        .filter(crates::name.eq(&opts.crate_name))
+        .first(&mut conn)
+        .await
+        .context("Failed to look up crate id from the database")?;
     {
         let crate_name = &opts.crate_name;
 
@@ -64,57 +59,51 @@ pub async fn run(opts: Opts) -> anyhow::Result<()> {
         }
     }
 
-    let opts = spawn_blocking(move || {
-        use diesel::RunQueryDsl;
-
-        let conn: &mut AsyncConnectionWrapper<_> = &mut conn.into();
-
+    let opts = conn.transaction(|conn| async move {
         let crate_name = &opts.crate_name;
 
-        conn.transaction(|conn| {
-            info!(%crate_name, %crate_id, versions = ?opts.versions, "Deleting versions from the database");
-            let result = diesel::delete(
-                versions::table
-                    .filter(versions::crate_id.eq(crate_id))
-                    .filter(versions::num.eq_any(&opts.versions)),
-            )
-            .execute(conn);
-
-            match result {
-                Ok(num_deleted) if num_deleted == opts.versions.len() => {}
-                Ok(num_deleted) => {
-                    warn!(
-                        %crate_name,
-                        "Deleted only {num_deleted} of {num_expected} versions from the database",
-                        num_expected = opts.versions.len()
-                    );
-                }
-                Err(error) => {
-                    warn!(%crate_name, ?error, "Failed to delete versions from the database")
-                }
+        info!(%crate_name, %crate_id, versions = ?opts.versions, "Deleting versions from the database");
+        let result = diesel::delete(
+            versions::table
+                .filter(versions::crate_id.eq(crate_id))
+                .filter(versions::num.eq_any(&opts.versions)),
+        )
+        .execute(conn).await;
+
+        match result {
+            Ok(num_deleted) if num_deleted == opts.versions.len() => {}
+            Ok(num_deleted) => {
+                warn!(
+                    %crate_name,
+                    "Deleted only {num_deleted} of {num_expected} versions from the database",
+                    num_expected = opts.versions.len()
+                );
             }
-
-            info!(%crate_name, %crate_id, "Updating default version in the database");
-            if let Err(error) = update_default_version(crate_id, conn) {
-                warn!(%crate_name, %crate_id, ?error, "Failed to update default version");
+            Err(error) => {
+                warn!(%crate_name, ?error, "Failed to delete versions from the database")
             }
-
-            Ok::<_, anyhow::Error>(())
-        })?;
-
-        info!(%crate_name, "Enqueuing index sync jobs");
-        if let Err(error) = jobs::SyncToGitIndex::new(crate_name).enqueue(conn) {
-            warn!(%crate_name, ?error, "Failed to enqueue SyncToGitIndex job");
         }
-        if let Err(error) = jobs::SyncToSparseIndex::new(crate_name).enqueue(conn) {
-            warn!(%crate_name, ?error, "Failed to enqueue SyncToSparseIndex job");
+
+        info!(%crate_name, %crate_id, "Updating default version in the database");
+        if let Err(error) = update_default_version(crate_id, conn).await {
+            warn!(%crate_name, %crate_id, ?error, "Failed to update default version");
         }
 
         Ok::<_, anyhow::Error>(opts)
-    }).await??;
+    }.scope_boxed()).await?;
 
     let crate_name = &opts.crate_name;
 
+    info!(%crate_name, "Enqueuing index sync jobs");
+    let job = jobs::SyncToGitIndex::new(crate_name);
+    if let Err(error) = job.async_enqueue(&mut conn).await {
+        warn!(%crate_name, ?error, "Failed to enqueue SyncToGitIndex job");
+    }
+    let job = jobs::SyncToSparseIndex::new(crate_name);
+    if let Err(error) = job.async_enqueue(&mut conn).await {
+        warn!(%crate_name, ?error, "Failed to enqueue SyncToSparseIndex job");
+    }
+
     for version in &opts.versions {
         debug!(%crate_name, %version, "Deleting crate file from S3");
         if let Err(error) = store.delete_crate_file(crate_name, version).await {

src/models.rs

@@ -1,10 +1,7 @@
 pub use self::action::{NewVersionOwnerAction, VersionAction, VersionOwnerAction};
 pub use self::category::{Category, CrateCategory, NewCategory};
 pub use self::crate_owner_invitation::{CrateOwnerInvitation, NewCrateOwnerInvitationOutcome};
-pub use self::default_versions::{
-    async_update_default_version, async_verify_default_version, update_default_version,
-    verify_default_version,
-};
+pub use self::default_versions::{update_default_version, verify_default_version};
 pub use self::deleted_crate::NewDeletedCrate;
 pub use self::dependency::{Dependency, DependencyKind, ReverseDependency};
 pub use self::download::VersionDownload;

src/models/default_versions.rs

@@ -1,8 +1,7 @@
 use crate::schema::{default_versions, versions};
 use crate::sql::SemverVersion;
-use crate::util::diesel::prelude::*;
-use crate::util::diesel::Conn;
-use diesel_async::AsyncPgConnection;
+use diesel::prelude::*;
+use diesel_async::{AsyncPgConnection, RunQueryDsl};
 
 /// A subset of the columns of the `versions` table.
 ///
@@ -57,13 +56,11 @@ impl Ord for Version {
 ///
 /// The default version is then written to the `default_versions` table.
 #[instrument(skip(conn))]
-pub async fn async_update_default_version(
+pub async fn update_default_version(
     crate_id: i32,
     conn: &mut AsyncPgConnection,
 ) -> QueryResult<()> {
-    use diesel_async::RunQueryDsl;
-
-    let default_version = async_calculate_default_version(crate_id, conn).await?;
+    let default_version = calculate_default_version(crate_id, conn).await?;
 
     debug!(
         "Updating default version to {} (id: {})…",
@@ -84,49 +81,13 @@ pub async fn async_update_default_version(
     Ok(())
 }
 
-/// Updates the `default_versions` table entry for the specified crate.
-///
-/// This function first loads all versions of the crate from the database,
-/// then determines the default version based on the following criteria:
-///
-/// 1. The highest non-prerelease version that is not yanked.
-/// 2. The highest non-yanked version.
-/// 3. The highest version.
-///
-/// The default version is then written to the `default_versions` table.
-#[instrument(skip(conn))]
-pub fn update_default_version(crate_id: i32, conn: &mut impl Conn) -> QueryResult<()> {
-    use diesel::RunQueryDsl;
-
-    let default_version = calculate_default_version(crate_id, conn)?;
-
-    debug!(
-        "Updating default version to {} (id: {})…",
-        default_version.num, default_version.id
-    );
-
-    diesel::insert_into(default_versions::table)
-        .values((
-            default_versions::crate_id.eq(crate_id),
-            default_versions::version_id.eq(default_version.id),
-        ))
-        .on_conflict(default_versions::crate_id)
-        .do_update()
-        .set(default_versions::version_id.eq(default_version.id))
-        .execute(conn)?;
-
-    Ok(())
-}
-
 /// Verifies that the default version for the specified crate is up-to-date.
 #[instrument(skip(conn))]
-pub async fn async_verify_default_version(
+pub async fn verify_default_version(
     crate_id: i32,
     conn: &mut AsyncPgConnection,
 ) -> QueryResult<()> {
-    use diesel_async::RunQueryDsl;
-
-    let calculated = async_calculate_default_version(crate_id, conn).await?;
+    let calculated = calculate_default_version(crate_id, conn).await?;
 
     let saved = default_versions::table
         .select(default_versions::version_id)
@@ -154,44 +115,11 @@ pub async fn async_verify_default_version(
     Ok(())
 }
 
-/// Verifies that the default version for the specified crate is up-to-date.
-#[instrument(skip(conn))]
-pub fn verify_default_version(crate_id: i32, conn: &mut impl Conn) -> QueryResult<()> {
-    use diesel::RunQueryDsl;
-
-    let calculated = calculate_default_version(crate_id, conn)?;
-
-    let saved = default_versions::table
-        .select(default_versions::version_id)
-        .filter(default_versions::crate_id.eq(crate_id))
-        .first::<i32>(conn)
-        .optional()?;
-
-    if let Some(saved) = saved {
-        if saved == calculated.id {
-            debug!("Default version for crate {crate_id} is up to date");
-        } else {
-            warn!(
-                "Default version for crate {crate_id} is outdated (expected: {saved}, actual: {})",
-                calculated.id,
-            );
-        }
-    } else {
-        warn!(
-            "Default version for crate {crate_id} is missing (expected: {})",
-            calculated.id
-        );
-    }
-
-    Ok(())
-}
-
-async fn async_calculate_default_version(
+async fn calculate_default_version(
     crate_id: i32,
     conn: &mut AsyncPgConnection,
 ) -> QueryResult<Version> {
     use diesel::result::Error::NotFound;
-    use diesel_async::RunQueryDsl;
 
     debug!("Loading all versions for the crate…");
     let versions = versions::table
@@ -205,21 +133,6 @@ async fn async_calculate_default_version(
     versions.into_iter().max().ok_or(NotFound)
 }
 
-fn calculate_default_version(crate_id: i32, conn: &mut impl Conn) -> QueryResult<Version> {
-    use diesel::result::Error::NotFound;
-    use diesel::RunQueryDsl;
-
-    debug!("Loading all versions for the crate…");
-    let versions = versions::table
-        .filter(versions::crate_id.eq(crate_id))
-        .select(Version::as_returning())
-        .load::<Version>(conn)?;
-
-    debug!("Found {} versions", versions.len());
-
-    versions.into_iter().max().ok_or(NotFound)
-}
-
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -331,19 +244,16 @@ mod tests {
         buf
     }
 
-    fn create_crate(name: &str, conn: &mut impl Conn) -> i32 {
-        use diesel::RunQueryDsl;
-
+    async fn create_crate(name: &str, conn: &mut AsyncPgConnection) -> i32 {
         diesel::insert_into(crates::table)
             .values(crates::name.eq(name))
             .returning(crates::id)
             .get_result(conn)
+            .await
             .unwrap()
     }
 
-    fn create_version(crate_id: i32, num: &str, conn: &mut impl Conn) {
-        use diesel::RunQueryDsl;
-
+    async fn create_version(crate_id: i32, num: &str, conn: &mut AsyncPgConnection) {
         diesel::insert_into(versions::table)
             .values((
                 versions::crate_id.eq(crate_id),
@@ -353,36 +263,36 @@ mod tests {
                 versions::crate_size.eq(0),
             ))
             .execute(conn)
+            .await
             .unwrap();
     }
 
-    fn get_default_version(crate_id: i32, conn: &mut impl Conn) -> String {
-        use diesel::RunQueryDsl;
-
+    async fn get_default_version(crate_id: i32, conn: &mut AsyncPgConnection) -> String {
         default_versions::table
             .inner_join(versions::table)
             .select(versions::num)
             .filter(default_versions::crate_id.eq(crate_id))
             .first(conn)
+            .await
             .unwrap()
     }
 
-    #[test]
-    fn test_update_default_version() {
+    #[tokio::test]
+    async fn test_update_default_version() {
         let test_db = TestDatabase::new();
-        let conn = &mut test_db.connect();
+        let conn = &mut test_db.async_connect().await;
 
-        let crate_id = create_crate("foo", conn);
-        create_version(crate_id, "1.0.0", conn);
+        let crate_id = create_crate("foo", conn).await;
+        create_version(crate_id, "1.0.0", conn).await;
 
-        update_default_version(crate_id, conn).unwrap();
-        assert_eq!(get_default_version(crate_id, conn), "1.0.0");
+        update_default_version(crate_id, conn).await.unwrap();
+        assert_eq!(get_default_version(crate_id, conn).await, "1.0.0");
 
-        create_version(crate_id, "1.1.0", conn);
-        create_version(crate_id, "1.0.1", conn);
-        assert_eq!(get_default_version(crate_id, conn), "1.0.0");
+        create_version(crate_id, "1.1.0", conn).await;
+        create_version(crate_id, "1.0.1", conn).await;
+        assert_eq!(get_default_version(crate_id, conn).await, "1.0.0");
 
-        update_default_version(crate_id, conn).unwrap();
-        assert_eq!(get_default_version(crate_id, conn), "1.1.0");
+        update_default_version(crate_id, conn).await.unwrap();
+        assert_eq!(get_default_version(crate_id, conn).await, "1.1.0");
     }
 }