Used crate ID for background jobs

rust-toolchain.toml

@@ -1,2 +1,3 @@
 [toolchain]
 channel = "1.88.0"
+components = ["rust-analyzer"]

src/bin/crates-admin/delete_crate.rs

@@ -83,32 +83,37 @@ pub async fn run(opts: Opts) -> anyhow::Result<()> {
     let available_at = now + chrono::TimeDelta::hours(opts.availability_delay);
 
     for name in &crate_names {
-        if let Some(crate_info) = existing_crates.iter().find(|info| info.name == *name) {
-            let id = crate_info.id;
-
-            let deleted_crate = NewDeletedCrate::builder(name)
-                .created_at(&crate_info.created_at)
-                .deleted_at(&now)
-                .deleted_by(deleted_by.id)
-                .maybe_message(opts.message.as_deref())
-                .available_at(&available_at)
-                .build();
-
-            info!("{name}: Deleting crate from the database…");
-            let result = conn
-                .transaction(|conn| delete_from_database(conn, id, deleted_crate).scope_boxed())
-                .await;
-
-            if let Err(error) = result {
-                warn!(%id, "{name}: Failed to delete crate from the database: {error}");
+        let crate_id =
+            if let Some(crate_info) = existing_crates.iter().find(|info| info.name == *name) {
+                let id = crate_info.id;
+
+                let deleted_crate = NewDeletedCrate::builder(name)
+                    .created_at(&crate_info.created_at)
+                    .deleted_at(&now)
+                    .deleted_by(deleted_by.id)
+                    .maybe_message(opts.message.as_deref())
+                    .available_at(&available_at)
+                    .build();
+
+                info!("{name}: Deleting crate from the database…");
+                let result = conn
+                    .transaction(|conn| delete_from_database(conn, id, deleted_crate).scope_boxed())
+                    .await;
+
+                if let Err(error) = result {
+                    warn!(%id, "{name}: Failed to delete crate from the database: {error}");
+                };
+
+                Some(id)
+            } else {
+                info!("{name}: Skipped missing crate");
+
+                None
             };
-        } else {
-            info!("{name}: Skipped missing crate");
-        };
 
         info!("{name}: Enqueuing background jobs…");
-        let git_index_job = jobs::SyncToGitIndex::new(name);
-        let sparse_index_job = jobs::SyncToSparseIndex::new(name);
+        let git_index_job = jobs::SyncToGitIndex::new_maybe_deleted(crate_id, name);
+        let sparse_index_job = jobs::SyncToSparseIndex::new_maybe_deleted(crate_id, name);
         let delete_from_storage_job = jobs::DeleteCrateFromStorage::new(name.into());
 
         if let Err(error) = tokio::try_join!(

src/bin/crates-admin/delete_version.rs

@@ -95,8 +95,8 @@ pub async fn run(opts: Opts) -> anyhow::Result<()> {
     let crate_name = &opts.crate_name;
 
     info!(%crate_name, "Enqueuing index sync jobs");
-    let git_index_job = jobs::SyncToGitIndex::new(crate_name);
-    let sparse_index_job = jobs::SyncToSparseIndex::new(crate_name);
+    let git_index_job = jobs::SyncToGitIndex::new(crate_id, crate_name);
+    let sparse_index_job = jobs::SyncToSparseIndex::new(crate_id, crate_name);
 
     if let Err(error) = tokio::try_join!(
         git_index_job.enqueue(&mut conn),

src/bin/crates-admin/enqueue_job.rs

@@ -6,7 +6,7 @@ use crates_io::worker::jobs;
 use crates_io_worker::BackgroundJob;
 use diesel::dsl::exists;
 use diesel::prelude::*;
-use diesel_async::RunQueryDsl;
+use diesel_async::{AsyncPgConnection, RunQueryDsl};
 
 #[derive(clap::Parser, Debug)]
 #[command(
@@ -152,10 +152,12 @@ pub async fn run(command: Command) -> Result<()> {
             jobs::rss::SyncCratesFeed.enqueue(&mut conn).await?;
         }
         Command::SyncToGitIndex { name } => {
-            jobs::SyncToGitIndex::new(name).enqueue(&mut conn).await?;
+            jobs::SyncToGitIndex::new_maybe_deleted(crate_id(&name, &mut conn).await?, name)
+                .enqueue(&mut conn)
+                .await?;
         }
         Command::SyncToSparseIndex { name } => {
-            jobs::SyncToSparseIndex::new(name)
+            jobs::SyncToSparseIndex::new_maybe_deleted(crate_id(&name, &mut conn).await?, name)
                 .enqueue(&mut conn)
                 .await?;
         }
@@ -173,3 +175,12 @@ pub async fn run(command: Command) -> Result<()> {
 
     Ok(())
 }
+
+async fn crate_id(name: &str, conn: &mut AsyncPgConnection) -> QueryResult<Option<i32>> {
+    crates::table
+        .filter(crates::name.eq(name))
+        .select(crates::id)
+        .first::<i32>(conn)
+        .await
+        .optional()
+}

src/bin/crates-admin/yank_version.rs

@@ -67,8 +67,8 @@ async fn yank(opts: Opts, conn: &mut AsyncPgConnection) -> anyhow::Result<()> {
         .execute(conn)
         .await?;
 
-    let git_index_job = SyncToGitIndex::new(&krate.name);
-    let sparse_index_job = SyncToSparseIndex::new(&krate.name);
+    let git_index_job = SyncToGitIndex::new(krate.id, &krate.name);
+    let sparse_index_job = SyncToSparseIndex::new(krate.id, &krate.name);
     let update_default_version_job = UpdateDefaultVersion::new(krate.id);
 
     tokio::try_join!(

src/controllers/krate/delete.rs

@@ -132,8 +132,8 @@ pub async fn delete_crate(
                 .execute(conn)
                 .await?;
 
-            let git_index_job = jobs::SyncToGitIndex::new(&krate.name);
-            let sparse_index_job = jobs::SyncToSparseIndex::new(&krate.name);
+            let git_index_job = jobs::SyncToGitIndex::new(krate.id, &krate.name);
+            let sparse_index_job = jobs::SyncToSparseIndex::new(krate.id, &krate.name);
             let delete_from_storage_job = jobs::DeleteCrateFromStorage::new(path.name);
 
             tokio::try_join!(

src/controllers/krate/publish.rs

@@ -611,10 +611,10 @@ pub async fn publish(app: AppState, req: Parts, body: Body) -> AppResult<Json<Go
             .await
             .map_err(|e| internal(format!("failed to upload crate: {e}")))?;
 
-        let git_index_job = jobs::SyncToGitIndex::new(&krate.name);
-        let sparse_index_job = jobs::SyncToSparseIndex::new(&krate.name);
+        let git_index_job = jobs::SyncToGitIndex::new(krate.id, &krate.name);
+        let sparse_index_job = jobs::SyncToSparseIndex::new(krate.id, &krate.name);
         let publish_notifications_job = SendPublishNotificationsJob::new(version.id);
-        let crate_feed_job = jobs::rss::SyncCrateFeed::new(krate.name.clone());
+        let crate_feed_job = jobs::rss::SyncCrateFeed::new(krate.id, krate.name.clone());
         let updates_feed_job = jobs::rss::SyncUpdatesFeed;
 
         tokio::try_join!(

src/controllers/version/update.rs

@@ -178,8 +178,8 @@ pub async fn perform_version_yank_update(
         .insert(conn)
         .await?;
 
-    let git_index_job = SyncToGitIndex::new(&krate.name);
-    let sparse_index_job = SyncToSparseIndex::new(&krate.name);
+    let git_index_job = SyncToGitIndex::new(krate.id, &krate.name);
+    let sparse_index_job = SyncToSparseIndex::new(krate.id, &krate.name);
     let update_default_version_job = UpdateDefaultVersion::new(krate.id);
 
     tokio::try_join!(

src/tests/worker/git.rs

@@ -58,9 +58,13 @@ async fn index_smoke_test() {
             .await
     );
 
-    assert_ok!(jobs::SyncToGitIndex::new("serde").enqueue(&mut conn).await);
     assert_ok!(
-        jobs::SyncToSparseIndex::new("serde")
+        jobs::SyncToGitIndex::new(krate.id, "serde")
+            .enqueue(&mut conn)
+            .await
+    );
+    assert_ok!(
+        jobs::SyncToSparseIndex::new(krate.id, "serde")
             .enqueue(&mut conn)
             .await
     );

src/tests/worker/rss/sync_crate_feed.rs

@@ -19,7 +19,13 @@ async fn test_sync_crate_feed() -> anyhow::Result<()> {
     create_version(&mut conn, "foo", "1.1.0", "2024-06-22T08:30:01Z").await?;
     create_version(&mut conn, "foo", "1.2.0", "2024-06-22T15:57:19Z").await?;
 
-    let job = jobs::rss::SyncCrateFeed::new("foo".to_string());
+    let crate_id = crates::table
+        .select(crates::id)
+        .filter(crates::name.eq("foo"))
+        .get_result::<i32>(&mut conn)
+        .await?;
+
+    let job = jobs::rss::SyncCrateFeed::new(crate_id, "foo".to_string());
     job.enqueue(&mut conn).await?;
 
     app.run_pending_background_jobs().await;

src/worker/jobs/index/sync.rs

@@ -13,13 +13,22 @@ use tracing::{debug, info, instrument};
 
 #[derive(Serialize, Deserialize)]
 pub struct SyncToGitIndex {
+    crate_id: Option<i32>,
     krate: String,
 }
 
 impl SyncToGitIndex {
-    pub fn new(krate: impl Into<String>) -> Self {
+    pub fn new(crate_id: i32, krate: impl Into<String>) -> Self {
         let krate = krate.into();
-        Self { krate }
+        Self {
+            crate_id: Some(crate_id),
+            krate,
+        }
+    }
+
+    pub fn new_maybe_deleted(crate_id: Option<i32>, krate: impl Into<String>) -> Self {
+        let krate = krate.into();
+        Self { crate_id, krate }
     }
 }
 
@@ -81,13 +90,22 @@ impl BackgroundJob for SyncToGitIndex {
 
 #[derive(Serialize, Deserialize)]
 pub struct SyncToSparseIndex {
+    crate_id: Option<i32>,
     krate: String,
 }
 
 impl SyncToSparseIndex {
-    pub fn new(krate: impl Into<String>) -> Self {
+    pub fn new(crate_id: i32, krate: impl Into<String>) -> Self {
+        let krate = krate.into();
+        Self {
+            crate_id: Some(crate_id),
+            krate,
+        }
+    }
+
+    pub fn new_maybe_deleted(crate_id: Option<i32>, krate: impl Into<String>) -> Self {
         let krate = krate.into();
-        Self { krate }
+        Self { crate_id, krate }
     }
 }
 

src/worker/jobs/rss/sync_crate_feed.rs

@@ -22,12 +22,16 @@ const NUM_ITEMS: i64 = 10;
 
 #[derive(Serialize, Deserialize)]
 pub struct SyncCrateFeed {
+    crate_id: Option<i32>,
     name: String,
 }
 
 impl SyncCrateFeed {
-    pub fn new(name: String) -> Self {
-        Self { name }
+    pub fn new(crate_id: i32, name: String) -> Self {
+        Self {
+            crate_id: Some(crate_id),
+            name,
+        }
     }
 }