index-watcher: handle deletion requests when the crate is only queued

Author: syphar

.sqlx/query-4b13fe2c8df2b8b8bf019344313b2bc6442482a604cf90fb6106154f8e69a1c2.json

@@ -245,6 +245,39 @@ impl AsyncBuildQueue {
         .await?
         .is_some())
     }
+
+    async fn remove_crate_from_queue(&self, name: &str) -> Result<()> {
+        let mut conn = self.db.get_async().await?;
+        sqlx::query!(
+            "DELETE
+             FROM queue
+             WHERE name = $1
+             ",
+            name
+        )
+        .execute(&mut *conn)
+        .await?;
+
+        Ok(())
+    }
+
+    async fn remove_version_from_queue(&self, name: &str, version: &Version) -> Result<()> {
+        let mut conn = self.db.get_async().await?;
+        sqlx::query!(
+            "DELETE
+             FROM queue
+             WHERE
+                name = $1 AND
+                version = $2
+             ",
+            name,
+            version as _,
+        )
+        .execute(&mut *conn)
+        .await?;
+
+        Ok(())
+    }
 }
 
 /// Locking functions.
@@ -331,19 +364,22 @@ impl AsyncBuildQueue {
                 }
 
                 self.queue_crate_invalidation(&mut conn, krate).await;
+                self.remove_crate_from_queue(krate).await?;
                 continue;
             }
 
             if let Some(release) = change.version_deleted() {
+                let version: Version = release
+                    .version
+                    .parse()
+                    .context("couldn't parse release version as semver")?;
+
                 match delete_version(
                     &mut conn,
                     &self.storage,
                     &self.config,
                     &release.name,
-                    &release
-                        .version
-                        .parse()
-                        .context("couldn't parse release version as semver")?,
+                    &version,
                 )
                 .await
                 .with_context(|| {
@@ -361,6 +397,8 @@ impl AsyncBuildQueue {
 
                 self.queue_crate_invalidation(&mut conn, &release.name)
                     .await;
+                self.remove_version_from_queue(&release.name, &version)
+                    .await?;
                 continue;
             }
 
@@ -849,9 +887,8 @@ FROM crates AS c
 mod tests {
     use super::*;
     use crate::db::types::BuildStatus;
-    use crate::test::{FakeBuild, TestEnvironment, V1, V2};
+    use crate::test::{FakeBuild, KRATE, TestEnvironment, V1, V2};
     use chrono::Utc;
-
     use std::time::Duration;
 
     #[tokio::test(flavor = "multi_thread")]
@@ -1719,4 +1756,48 @@ mod tests {
 
         Ok(())
     }
+
+    #[tokio::test(flavor = "multi_thread")]
+    async fn test_delete_version_from_queue() -> Result<()> {
+        let env = TestEnvironment::new().await?;
+
+        let queue = env.async_build_queue();
+        assert_eq!(queue.pending_count().await?, 0);
+
+        queue.add_crate(KRATE, &V1, 0, None).await?;
+        queue.add_crate(KRATE, &V2, 0, None).await?;
+
+        assert_eq!(queue.pending_count().await?, 2);
+        queue.remove_version_from_queue(KRATE, &V1).await?;
+
+        assert_eq!(queue.pending_count().await?, 1);
+
+        // only v2 remains
+        if let [k] = queue.queued_crates().await?.as_slice() {
+            assert_eq!(k.name, KRATE);
+            assert_eq!(k.version, V2);
+        } else {
+            panic!("expected only one queued crate");
+        }
+
+        Ok(())
+    }
+
+    #[tokio::test(flavor = "multi_thread")]
+    async fn test_delete_crate_from_queue() -> Result<()> {
+        let env = TestEnvironment::new().await?;
+
+        let queue = env.async_build_queue();
+        assert_eq!(queue.pending_count().await?, 0);
+
+        queue.add_crate(KRATE, &V1, 0, None).await?;
+        queue.add_crate(KRATE, &V2, 0, None).await?;
+
+        assert_eq!(queue.pending_count().await?, 2);
+        queue.remove_crate_from_queue(KRATE).await?;
+
+        assert_eq!(queue.pending_count().await?, 0);
+
+        Ok(())
+    }
 }

.sqlx/query-f8b389df3451e4b5e6539e9260ba6340edf69c7dba22e667aedd510e868b0f00.json

@@ -15,20 +15,17 @@ use super::{CrateId, update_latest_version_id};
 static LIBRARY_STORAGE_PATHS_TO_DELETE: &[&str] = &["rustdoc", "rustdoc-json", "sources"];
 static OTHER_STORAGE_PATHS_TO_DELETE: &[&str] = &["sources"];
 
-#[derive(Debug, thiserror::Error)]
-enum CrateDeletionError {
-    #[error("crate is missing: {0}")]
-    MissingCrate(String),
-}
-
 #[context("error trying to delete crate {name} from database")]
 pub async fn delete_crate(
     conn: &mut sqlx::PgConnection,
     storage: &AsyncStorage,
     config: &Config,
     name: &str,
 ) -> Result<()> {
-    let crate_id = get_id(conn, name).await?;
+    let Some(crate_id) = get_id(conn, name).await? else {
+        return Ok(());
+    };
+
     let is_library = delete_crate_from_database(conn, name, crate_id).await?;
     // #899
     let paths = if is_library {
@@ -68,7 +65,11 @@ pub async fn delete_version(
     name: &str,
     version: &Version,
 ) -> Result<()> {
-    let is_library = delete_version_from_database(conn, name, version).await?;
+    let Some(crate_id) = get_id(conn, name).await? else {
+        return Ok(());
+    };
+
+    let is_library = delete_version_from_database(conn, crate_id, version).await?;
     let paths = if is_library {
         LIBRARY_STORAGE_PATHS_TO_DELETE
     } else {
@@ -105,7 +106,7 @@ pub async fn delete_version(
     Ok(())
 }
 
-async fn get_id(conn: &mut sqlx::PgConnection, name: &str) -> Result<CrateId> {
+async fn get_id(conn: &mut sqlx::PgConnection, name: &str) -> Result<Option<CrateId>> {
     Ok(sqlx::query_scalar!(
         r#"
         SELECT id as "id: CrateId"
@@ -115,8 +116,7 @@ async fn get_id(conn: &mut sqlx::PgConnection, name: &str) -> Result<CrateId> {
         name
     )
     .fetch_optional(&mut *conn)
-    .await?
-    .ok_or_else(|| CrateDeletionError::MissingCrate(name.into()))?)
+    .await?)
 }
 
 // metaprogramming!
@@ -131,10 +131,9 @@ const METADATA: &[(&str, &str)] = &[
 /// Returns whether this release was a library
 async fn delete_version_from_database(
     conn: &mut sqlx::PgConnection,
-    name: &str,
+    crate_id: CrateId,
     version: &Version,
 ) -> Result<bool> {
-    let crate_id = get_id(conn, name).await?;
     let mut transaction = conn.begin().await?;
     for &(table, column) in METADATA {
         sqlx::query(
@@ -152,20 +151,6 @@ async fn delete_version_from_database(
 
     update_latest_version_id(&mut transaction, crate_id).await?;
 
-    let paths = if is_library {
-        LIBRARY_STORAGE_PATHS_TO_DELETE
-    } else {
-        OTHER_STORAGE_PATHS_TO_DELETE
-    };
-    for prefix in paths {
-        sqlx::query!(
-            "DELETE FROM files WHERE path LIKE $1;",
-            format!("{prefix}/{name}/{version}/%"),
-        )
-        .execute(&mut *transaction)
-        .await?;
-    }
-
     transaction.commit().await?;
     Ok(is_library)
 }
@@ -224,7 +209,7 @@ mod tests {
     use crate::db::ReleaseId;
     use crate::registry_api::{CrateOwner, OwnerKind};
     use crate::storage::{CompressionAlgorithm, rustdoc_json_path};
-    use crate::test::{V1, V2, async_wrapper, fake_release_that_failed_before_build};
+    use crate::test::{KRATE, V1, V2, async_wrapper, fake_release_that_failed_before_build};
     use test_case::test_case;
 
     async fn crate_exists(conn: &mut sqlx::PgConnection, name: &str) -> Result<bool> {
@@ -542,4 +527,35 @@ mod tests {
             Ok(())
         })
     }
+
+    #[tokio::test(flavor = "multi_thread")]
+    async fn test_delete_missing_crate_doesnt_error() -> Result<()> {
+        let env = crate::test::TestEnvironment::new().await?;
+
+        let db = env.async_db();
+        let mut conn = db.async_conn().await;
+
+        assert!(!crate_exists(&mut conn, KRATE).await?);
+        delete_crate(&mut conn, env.async_storage(), env.config(), KRATE).await?;
+
+        assert!(!crate_exists(&mut conn, KRATE).await?);
+
+        Ok(())
+    }
+
+    #[tokio::test(flavor = "multi_thread")]
+    async fn test_delete_missing_version_doesnt_error() -> Result<()> {
+        let env = crate::test::TestEnvironment::new().await?;
+
+        let db = env.async_db();
+        let mut conn = db.async_conn().await;
+
+        assert!(!crate_exists(&mut conn, KRATE).await?);
+
+        delete_version(&mut conn, env.async_storage(), env.config(), KRATE, &V1).await?;
+
+        assert!(!crate_exists(&mut conn, KRATE).await?);
+
+        Ok(())
+    }
 }

src/build_queue.rs

@@ -41,6 +41,8 @@ use tokio::{runtime, task::block_in_place};
 use tower::ServiceExt;
 use tracing::error;
 
+// testing krate name constants
+pub(crate) const KRATE: &str = "krate";
 // some versions as constants for tests
 pub(crate) const V0_1: Version = Version::new(0, 1, 0);
 pub(crate) const V1: Version = Version::new(1, 0, 0);