Use ProcessCloudfrontInvalidationQueue background job for CloudFront invalidations

Author: Turbo87

src/worker/environment.rs

@@ -3,12 +3,15 @@ use crate::cloudfront::CloudFront;
 use crate::fastly::Fastly;
 use crate::storage::Storage;
 use crate::typosquat;
+use crate::worker::jobs::ProcessCloudfrontInvalidationQueue;
 use anyhow::Context;
 use bon::Builder;
+use crates_io_database::models::CloudFrontInvalidationQueueItem;
 use crates_io_docs_rs::DocsRsClient;
 use crates_io_index::{Repository, RepositoryConfig};
 use crates_io_og_image::OgImageGenerator;
 use crates_io_team_repo::TeamRepo;
+use crates_io_worker::BackgroundJob;
 use diesel_async::AsyncPgConnection;
 use diesel_async::pooled_connection::deadpool::Pool;
 use object_store::ObjectStore;
@@ -70,9 +73,20 @@ impl Environment {
     }
 
     /// Invalidate a file in all registered CDNs.
-    pub(crate) async fn invalidate_cdns(&self, path: &str) -> anyhow::Result<()> {
-        if let Some(cloudfront) = self.cloudfront() {
-            cloudfront.invalidate(path).await.context("CloudFront")?;
+    pub(crate) async fn invalidate_cdns(
+        &self,
+        conn: &mut AsyncPgConnection,
+        path: &str,
+    ) -> anyhow::Result<()> {
+        // Queue CloudFront invalidations for batch processing instead of calling directly
+        if self.cloudfront().is_some() {
+            let paths = &[path.to_string()];
+            let result = CloudFrontInvalidationQueueItem::queue_paths(conn, paths).await;
+            result.context("Failed to queue CloudFront invalidation path")?;
+
+            // Schedule the processing job to handle the queued paths
+            let result = ProcessCloudfrontInvalidationQueue.enqueue(conn).await;
+            result.context("Failed to enqueue CloudFront invalidation processing job")?;
         }
 
         if let Some(fastly) = self.fastly() {

src/worker/jobs/dump_db.rs

@@ -47,7 +47,8 @@ impl BackgroundJob for DumpDb {
         info!("Database dump tarball uploaded");
 
         info!("Invalidating CDN caches…");
-        if let Err(error) = env.invalidate_cdns(TAR_PATH).await {
+        let mut conn = env.deadpool.get().await?;
+        if let Err(error) = env.invalidate_cdns(&mut conn, TAR_PATH).await {
             warn!("Failed to invalidate CDN caches: {error}");
         }
 
@@ -58,7 +59,7 @@ impl BackgroundJob for DumpDb {
         info!("Database dump zip file uploaded");
 
         info!("Invalidating CDN caches…");
-        if let Err(error) = env.invalidate_cdns(ZIP_PATH).await {
+        if let Err(error) = env.invalidate_cdns(&mut conn, ZIP_PATH).await {
             warn!("Failed to invalidate CDN caches: {error}");
         }
 

src/worker/jobs/generate_og_image.rs

@@ -1,7 +1,9 @@
 use crate::models::OwnerKind;
 use crate::schema::*;
 use crate::worker::Environment;
+use crate::worker::jobs::ProcessCloudfrontInvalidationQueue;
 use anyhow::Context;
+use crates_io_database::models::CloudFrontInvalidationQueueItem;
 use crates_io_og_image::{OgImageAuthorData, OgImageData};
 use crates_io_worker::BackgroundJob;
 use diesel::prelude::*;
@@ -104,11 +106,17 @@ impl BackgroundJob for GenerateOgImage {
         // Invalidate CDN cache for the OG image
         let og_image_path = format!("og-images/{crate_name}.png");
 
-        // Invalidate CloudFront CDN
-        if let Some(cloudfront) = ctx.cloudfront()
-            && let Err(error) = cloudfront.invalidate(&og_image_path).await
-        {
-            warn!("Failed to invalidate CloudFront CDN for {crate_name}: {error}");
+        // Queue CloudFront invalidation for batch processing
+        if ctx.cloudfront().is_some() {
+            let paths = std::slice::from_ref(&og_image_path);
+            let result = CloudFrontInvalidationQueueItem::queue_paths(&mut conn, paths).await;
+            if let Err(error) = result {
+                warn!("Failed to queue CloudFront invalidation for {crate_name}: {error}");
+            } else if let Err(error) = ProcessCloudfrontInvalidationQueue.enqueue(&mut conn).await {
+                warn!(
+                    "Failed to enqueue CloudFront invalidation processing job for {crate_name}: {error}"
+                );
+            }
         }
 
         // Invalidate Fastly CDN

src/worker/jobs/index/sync.rs

@@ -1,7 +1,9 @@
 use crate::index::get_index_data;
 use crate::tasks::spawn_blocking;
 use crate::worker::Environment;
+use crate::worker::jobs::ProcessCloudfrontInvalidationQueue;
 use anyhow::Context;
+use crates_io_database::models::CloudFrontInvalidationQueueItem;
 use crates_io_index::Repository;
 use crates_io_worker::BackgroundJob;
 use serde::{Deserialize, Serialize};
@@ -113,12 +115,17 @@ impl BackgroundJob for SyncToSparseIndex {
         let future = env.storage.sync_index(&self.krate, content);
         future.await.context("Failed to sync index data")?;
 
-        if let Some(cloudfront) = env.cloudfront() {
+        if env.cloudfront().is_some() {
             let path = Repository::relative_index_file_for_url(&self.krate);
 
-            info!(%path, "Invalidating index file on CloudFront");
-            let future = cloudfront.invalidate(&path);
-            future.await.context("Failed to invalidate CloudFront")?;
+            info!(%path, "Queuing index file invalidation on CloudFront");
+
+            let paths = &[path];
+            let result = CloudFrontInvalidationQueueItem::queue_paths(&mut conn, paths).await;
+            result.context("Failed to queue CloudFront invalidation path")?;
+
+            let result = ProcessCloudfrontInvalidationQueue.enqueue(&mut conn).await;
+            result.context("Failed to enqueue CloudFront invalidation processing job")?;
         }
         Ok(())
     }

src/worker/jobs/index_version_downloads_archive/mod.rs

@@ -54,10 +54,11 @@ impl BackgroundJob for IndexVersionDownloadsArchive {
         info!("index.json generated and uploaded");
 
         info!("Invalidating CDN caches…");
-        if let Err(error) = env.invalidate_cdns(INDEX_PATH).await {
+        let mut conn = env.deadpool.get().await?;
+        if let Err(error) = env.invalidate_cdns(&mut conn, INDEX_PATH).await {
             warn!("Failed to invalidate CDN caches: {error}");
         }
-        if let Err(error) = env.invalidate_cdns(INDEX_JSON_PATH).await {
+        if let Err(error) = env.invalidate_cdns(&mut conn, INDEX_JSON_PATH).await {
             warn!("Failed to invalidate CDN caches: {error}");
         }
         info!("CDN caches invalidated");

src/worker/jobs/invalidate_cdns.rs

@@ -1,10 +1,12 @@
 use std::sync::Arc;
 
 use anyhow::Context;
+use crates_io_database::models::CloudFrontInvalidationQueueItem;
 use crates_io_worker::BackgroundJob;
 use serde::{Deserialize, Serialize};
 
 use crate::worker::Environment;
+use crate::worker::jobs::ProcessCloudfrontInvalidationQueue;
 
 /// A background job that invalidates the given paths on all CDNs in use on crates.io.
 #[derive(Deserialize, Serialize)]
@@ -46,11 +48,16 @@ impl BackgroundJob for InvalidateCdns {
             }
         }
 
-        if let Some(cloudfront) = ctx.cloudfront() {
-            cloudfront
-                .invalidate_many(self.paths.clone())
-                .await
-                .context("Failed to invalidate paths on CloudFront CDN")?;
+        // Queue CloudFront invalidations for batch processing instead of calling directly
+        if ctx.cloudfront().is_some() {
+            let mut conn = ctx.deadpool.get().await?;
+
+            let result = CloudFrontInvalidationQueueItem::queue_paths(&mut conn, &self.paths).await;
+            result.context("Failed to queue CloudFront invalidation paths")?;
+
+            // Schedule the processing job to handle the queued paths
+            let result = ProcessCloudfrontInvalidationQueue.enqueue(&mut conn).await;
+            result.context("Failed to enqueue CloudFront invalidation processing job")?;
         }
 
         Ok(())

src/worker/jobs/rss/sync_crate_feed.rs

@@ -80,7 +80,7 @@ impl BackgroundJob for SyncCrateFeed {
         ctx.storage.upload_feed(&feed_id, &channel).await?;
 
         let path = object_store::path::Path::from(&feed_id);
-        if let Err(error) = ctx.invalidate_cdns(path.as_ref()).await {
+        if let Err(error) = ctx.invalidate_cdns(&mut conn, path.as_ref()).await {
             warn!("Failed to invalidate CDN caches: {error}");
         }
 

src/worker/jobs/rss/sync_crates_feed.rs

@@ -67,7 +67,7 @@ impl BackgroundJob for SyncCratesFeed {
         ctx.storage.upload_feed(&feed_id, &channel).await?;
 
         let path = object_store::path::Path::from(&feed_id);
-        if let Err(error) = ctx.invalidate_cdns(path.as_ref()).await {
+        if let Err(error) = ctx.invalidate_cdns(&mut conn, path.as_ref()).await {
             warn!("Failed to invalidate CDN caches: {error}");
         }
 

src/worker/jobs/rss/sync_updates_feed.rs

@@ -67,7 +67,7 @@ impl BackgroundJob for SyncUpdatesFeed {
         ctx.storage.upload_feed(&feed_id, &channel).await?;
 
         let path = object_store::path::Path::from(&feed_id);
-        if let Err(error) = ctx.invalidate_cdns(path.as_ref()).await {
+        if let Err(error) = ctx.invalidate_cdns(&mut conn, path.as_ref()).await {
             warn!("Failed to invalidate CDN caches: {error}");
         }