Merge pull request #9549 from Turbo87/delete-crate-job

Extract `DeleteCrateFromStorage` background job

Author: Turbo87

src/admin/delete_crate.rs

@@ -1,8 +1,8 @@
 use crate::schema::{crate_owners, teams, users};
-use crate::storage::{FeedId, Storage};
 use crate::worker::jobs;
 use crate::{admin::dialoguer, db, schema::crates};
 use anyhow::Context;
+use crates_io_worker::BackgroundJob;
 use diesel::dsl::sql;
 use diesel::prelude::*;
 use diesel::sql_types::Text;
@@ -27,8 +27,6 @@ pub struct Opts {
 pub fn run(opts: Opts) -> anyhow::Result<()> {
     let conn = &mut db::oneoff_connection().context("Failed to establish database connection")?;
 
-    let store = Storage::from_environment();
-
     let mut crate_names = opts.crate_names;
     crate_names.sort();
 
@@ -73,11 +71,6 @@ pub fn run(opts: Opts) -> anyhow::Result<()> {
         return Ok(());
     }
 
-    let rt = tokio::runtime::Builder::new_current_thread()
-        .enable_all()
-        .build()
-        .context("Failed to initialize tokio runtime")?;
-
     for name in &crate_names {
         if let Some((id, _)) = existing_crates.get(name) {
             info!("{name}: Deleting crate from the database…");
@@ -93,20 +86,10 @@ pub fn run(opts: Opts) -> anyhow::Result<()> {
             warn!("{name}: Failed to enqueue index sync jobs: {error}");
         }
 
-        info!("{name}: Deleting crate files from S3…");
-        if let Err(error) = rt.block_on(store.delete_all_crate_files(name)) {
-            warn!("{name}: Failed to delete crate files from S3: {error}");
-        }
-
-        info!("{name}: Deleting readme files from S3…");
-        if let Err(error) = rt.block_on(store.delete_all_readmes(name)) {
-            warn!("{name}: Failed to delete readme files from S3: {error}");
-        }
-
-        info!("{name}: Deleting RSS feed from S3…");
-        let feed_id = FeedId::Crate { name };
-        if let Err(error) = rt.block_on(store.delete_feed(&feed_id)) {
-            warn!("{name}: Failed to delete RSS feed from S3: {error}");
+        info!("{name}: Enqueuing DeleteCrateFromStorage job…");
+        let job = jobs::DeleteCrateFromStorage::new(name.into());
+        if let Err(error) = job.enqueue(conn) {
+            warn!("{name}: Failed to enqueue DeleteCrateFromStorage job: {error}");
         }
     }
 

src/worker/jobs/delete_crate.rs

@@ -0,0 +1,50 @@
+use crate::storage::FeedId;
+use crate::worker::Environment;
+use anyhow::Context;
+use crates_io_worker::BackgroundJob;
+use std::sync::Arc;
+use tokio::try_join;
+
+/// A background job that deletes all files associated with a crate from the storage backend.
+#[derive(Serialize, Deserialize)]
+pub struct DeleteCrateFromStorage {
+    name: String,
+}
+
+impl DeleteCrateFromStorage {
+    pub fn new(name: String) -> Self {
+        Self { name }
+    }
+}
+
+impl BackgroundJob for DeleteCrateFromStorage {
+    const JOB_NAME: &'static str = "delete_crate_from_storage";
+
+    type Context = Arc<Environment>;
+
+    async fn run(&self, ctx: Self::Context) -> anyhow::Result<()> {
+        let name = &self.name;
+
+        try_join!(
+            async {
+                info!("{name}: Deleting crate files from S3…");
+                let result = ctx.storage.delete_all_crate_files(name).await;
+                result.context("Failed to delete crate files from S3")
+            },
+            async {
+                info!("{name}: Deleting readme files from S3…");
+                let result = ctx.storage.delete_all_readmes(name).await;
+                result.context("Failed to delete readme files from S3")
+            },
+            async {
+                info!("{name}: Deleting RSS feed from S3…");
+                let feed_id = FeedId::Crate { name };
+                let result = ctx.storage.delete_feed(&feed_id).await;
+                result.context("Failed to delete RSS feed from S3")
+            }
+        )?;
+
+        info!("{name}: Successfully deleted crate from S3");
+        Ok(())
+    }
+}

src/worker/jobs/mod.rs

@@ -8,6 +8,7 @@ use std::fmt::Display;
 
 mod archive_version_downloads;
 mod daily_db_maintenance;
+mod delete_crate;
 mod downloads;
 pub mod dump_db;
 mod expiry_notification;
@@ -22,6 +23,7 @@ mod update_default_version;
 
 pub use self::archive_version_downloads::ArchiveVersionDownloads;
 pub use self::daily_db_maintenance::DailyDbMaintenance;
+pub use self::delete_crate::DeleteCrateFromStorage;
 pub use self::downloads::{
     CleanProcessedLogFiles, ProcessCdnLog, ProcessCdnLogQueue, UpdateDownloads,
 };

src/worker/mod.rs

@@ -23,6 +23,7 @@ impl RunnerExt for Runner<Arc<Environment>> {
             .register_job_type::<jobs::CheckTyposquat>()
             .register_job_type::<jobs::CleanProcessedLogFiles>()
             .register_job_type::<jobs::DailyDbMaintenance>()
+            .register_job_type::<jobs::DeleteCrateFromStorage>()
             .register_job_type::<jobs::DumpDb>()
             .register_job_type::<jobs::IndexVersionDownloadsArchive>()
             .register_job_type::<jobs::NormalizeIndex>()