Skip running a job if the crate/version deleted

src/tests/worker/mod.rs

@@ -1,3 +1,6 @@
 mod git;
+mod readmes;
 mod rss;
+mod send_publish_notifications;
 mod sync_admins;
+mod update_default_version;

src/tests/worker/readmes.rs

@@ -0,0 +1,17 @@
+use crate::tests::util::TestApp;
+use crate::worker::jobs;
+use crates_io_worker::BackgroundJob;
+
+#[tokio::test(flavor = "multi_thread")]
+async fn skips_when_crate_deleted() -> anyhow::Result<()> {
+    let (app, _) = TestApp::full().empty().await;
+    let mut conn = app.db_conn().await;
+
+    let job =
+        jobs::RenderAndUploadReadme::new(-1, "deleted".to_string(), ".".to_string(), None, None);
+
+    job.enqueue(&mut conn).await?;
+    app.run_pending_background_jobs().await;
+
+    Ok(())
+}

src/tests/worker/send_publish_notifications.rs

@@ -0,0 +1,16 @@
+use crate::tests::util::TestApp;
+use crate::worker::jobs;
+use crates_io_worker::BackgroundJob;
+
+#[tokio::test(flavor = "multi_thread")]
+async fn skips_when_crate_deleted() -> anyhow::Result<()> {
+    let (app, _) = TestApp::full().empty().await;
+    let mut conn = app.db_conn().await;
+
+    let job = jobs::SendPublishNotificationsJob::new(-1);
+
+    job.enqueue(&mut conn).await?;
+    app.run_pending_background_jobs().await;
+
+    Ok(())
+}

src/tests/worker/update_default_version.rs

@@ -0,0 +1,16 @@
+use crate::tests::util::TestApp;
+use crate::worker::jobs;
+use crates_io_worker::BackgroundJob;
+
+#[tokio::test(flavor = "multi_thread")]
+async fn skips_when_crate_deleted() -> anyhow::Result<()> {
+    let (app, _) = TestApp::full().empty().await;
+    let mut conn = app.db_conn().await;
+
+    let job = jobs::UpdateDefaultVersion::new(-1);
+
+    job.enqueue(&mut conn).await?;
+    app.run_pending_background_jobs().await;
+
+    Ok(())
+}

src/worker/jobs/readmes.rs

@@ -5,11 +5,13 @@ use crate::tasks::spawn_blocking;
 use crate::worker::Environment;
 use crates_io_markdown::text_to_html;
 use crates_io_worker::BackgroundJob;
+use diesel::result::DatabaseErrorKind;
+use diesel::result::Error::DatabaseError;
 use diesel_async::AsyncConnection;
 use diesel_async::scoped_futures::ScopedFutureExt;
 use serde::{Deserialize, Serialize};
 use std::sync::Arc;
-use tracing::{info, instrument};
+use tracing::{info, instrument, warn};
 
 #[derive(Clone, Serialize, Deserialize)]
 pub struct RenderAndUploadReadme {
@@ -70,13 +72,39 @@ impl BackgroundJob for RenderAndUploadReadme {
         let mut conn = env.deadpool.get().await?;
         conn.transaction(|conn| {
             async move {
-                Version::record_readme_rendering(job.version_id, conn).await?;
-                let (crate_name, vers): (String, String) = versions::table
+                match Version::record_readme_rendering(job.version_id, conn).await {
+                    Ok(_) => {}
+                    Err(DatabaseError(DatabaseErrorKind::ForeignKeyViolation, ..)) => {
+                        warn!(
+                            "Skipping README rendering recording for version {}: version not found",
+                            job.version_id
+                        );
+                        return Ok(());
+                    }
+                    Err(err) => {
+                        warn!(
+                            "Failed to record README rendering for version {}: {err}",
+                            job.version_id,
+                        );
+                        return Err(err.into());
+                    }
+                }
+
+                let result = versions::table
                     .find(job.version_id)
                     .inner_join(crates::table)
                     .select((crates::name, versions::num))
-                    .first(conn)
-                    .await?;
+                    .first::<(String, String)>(conn)
+                    .await
+                    .optional()?;
+
+                let Some((crate_name, vers)) = result else {
+                    warn!(
+                        "Skipping README rendering for version {}: version not found",
+                        job.version_id
+                    );
+                    return Ok(());
+                };
 
                 tracing::Span::current().record("krate.name", tracing::field::display(&crate_name));
 

src/worker/jobs/send_publish_notifications.rs

@@ -39,7 +39,12 @@ impl BackgroundJob for SendPublishNotificationsJob {
         let mut conn = ctx.deadpool.get().await?;
 
         // Get crate name, version and other publish details
-        let publish_details = PublishDetails::for_version(version_id, &mut conn).await?;
+        let Some(publish_details) = PublishDetails::for_version(version_id, &mut conn).await?
+        else {
+            warn!("Skipping publish notifications for {version_id}: no version found");
+
+            return Ok(());
+        };
 
         let publish_time = publish_details
             .publish_time
@@ -168,13 +173,17 @@ struct PublishDetails {
 }
 
 impl PublishDetails {
-    async fn for_version(version_id: i32, conn: &mut AsyncPgConnection) -> QueryResult<Self> {
+    async fn for_version(
+        version_id: i32,
+        conn: &mut AsyncPgConnection,
+    ) -> QueryResult<Option<Self>> {
         versions::table
             .find(version_id)
             .inner_join(crates::table)
             .left_join(users::table)
             .select(PublishDetails::as_select())
             .first(conn)
             .await
+            .optional()
     }
 }

src/worker/jobs/update_default_version.rs

@@ -7,7 +7,7 @@ use diesel::prelude::*;
 use diesel_async::RunQueryDsl;
 use serde::{Deserialize, Serialize};
 use std::sync::Arc;
-use tracing::info;
+use tracing::{info, warn};
 
 #[derive(Serialize, Deserialize)]
 pub struct UpdateDefaultVersion {
@@ -32,18 +32,37 @@ impl BackgroundJob for UpdateDefaultVersion {
 
         info!("Updating default version for crate {crate_id}");
         let mut conn = ctx.deadpool.get().await?;
-        update_default_version(crate_id, &mut conn).await?;
+
+        match update_default_version(crate_id, &mut conn).await {
+            Ok(_) => {
+                info!("Successfully updated default version for crate {crate_id}");
+            }
+            Err(diesel::result::Error::NotFound) => {
+                warn!("Skipping default version update for crate {crate_id}: crate not found");
+                return Ok(());
+            }
+            Err(err) => {
+                warn!("Failed to update default version for crate {crate_id}: {err}");
+                return Err(err.into());
+            }
+        }
 
         // Get the crate name for OG image generation
-        let crate_name: String = crates::table
+        let crate_name = crates::table
             .filter(crates::id.eq(crate_id))
             .select(crates::name)
-            .first(&mut conn)
-            .await?;
+            .first::<String>(&mut conn)
+            .await
+            .optional()?;
 
-        // Generate OG image after updating default version
-        info!("Enqueueing OG image generation for crate {crate_name}");
-        GenerateOgImage::new(crate_name).enqueue(&mut conn).await?;
+        if let Some(crate_name) = crate_name {
+            // Generate OG image after updating default version
+            info!("Enqueueing OG image generation for crate {crate_name}");
+            GenerateOgImage::new(crate_name).enqueue(&mut conn).await?;
+        } else {
+            warn!("No crate found for ID {crate_id}, skipping OG image generation");
+            return Ok(());
+        }
 
         Ok(())
     }