implement CDN invalidation queue & background worker

Author: syphar

src/bin/cratesfyi.rs

@@ -108,6 +108,8 @@ enum CommandLine {
             value_enum
         )]
         repository_stats_updater: Toggle,
+        #[arg(long = "cdn-invalidator", default_value = "enabled", value_enum)]
+        cdn_invalidator: Toggle,
     },
 
     StartBuildServer,
@@ -143,10 +145,14 @@ impl CommandLine {
             } => subcommand.handle_args(ctx, skip_if_exists)?,
             Self::StartRegistryWatcher {
                 repository_stats_updater,
+                cdn_invalidator,
             } => {
                 if repository_stats_updater == Toggle::Enabled {
                     docs_rs::utils::daemon::start_background_repository_stats_updater(&ctx)?;
                 }
+                if cdn_invalidator == Toggle::Enabled {
+                    docs_rs::utils::daemon::start_background_cdn_invalidator(&ctx)?;
+                }
 
                 docs_rs::utils::watch_registry(ctx.build_queue()?, ctx.config()?, ctx.index()?)?;
             }
@@ -586,7 +592,6 @@ impl Context for BinContext {
             self.pool()?,
             self.metrics()?,
             self.config()?,
-            self.cdn()?,
             self.storage()?,
         );
         fn storage(self) -> Storage = Storage::new(
@@ -595,7 +600,10 @@ impl Context for BinContext {
             self.config()?,
             self.runtime()?,
         )?;
-        fn cdn(self) -> CdnBackend = CdnBackend::new(&self.config()?, &self.runtime()?);
+        fn cdn(self) -> CdnBackend = CdnBackend::new(
+            &self.config()?,
+            &self.runtime()?,
+        );
         fn config(self) -> Config = Config::from_env()?;
         fn metrics(self) -> Metrics = Metrics::new()?;
         fn runtime(self) -> Runtime = {

src/build_queue.rs

@@ -1,4 +1,4 @@
-use crate::cdn::{self, CdnBackend};
+use crate::cdn;
 use crate::db::{delete_crate, Pool};
 use crate::docbuilder::PackageKind;
 use crate::error::Result;
@@ -27,7 +27,6 @@ pub(crate) struct QueuedCrate {
 #[derive(Debug)]
 pub struct BuildQueue {
     config: Arc<Config>,
-    cdn: Arc<CdnBackend>,
     storage: Arc<Storage>,
     pub(crate) db: Pool,
     metrics: Arc<Metrics>,
@@ -39,13 +38,11 @@ impl BuildQueue {
         db: Pool,
         metrics: Arc<Metrics>,
         config: Arc<Config>,
-        cdn: Arc<CdnBackend>,
         storage: Arc<Storage>,
     ) -> Self {
         BuildQueue {
             max_attempts: config.build_attempts.into(),
             config,
-            cdn,
             db,
             metrics,
             storage,
@@ -191,7 +188,9 @@ impl BuildQueue {
             )
         });
         self.metrics.total_builds.inc();
-        if let Err(err) = cdn::invalidate_crate(&self.config, &self.cdn, &to_process.name) {
+        if let Err(err) =
+            cdn::queue_crate_invalidation(&mut transaction, &self.config, &to_process.name)
+        {
             report_error(&err);
         }
 
@@ -321,7 +320,7 @@ impl BuildQueue {
                     ),
                     Err(err) => report_error(&err),
                 }
-                if let Err(err) = cdn::invalidate_crate(&self.config, &self.cdn, krate) {
+                if let Err(err) = cdn::queue_crate_invalidation(&mut *conn, &self.config, krate) {
                     report_error(&err);
                 }
                 continue;
@@ -366,7 +365,9 @@ impl BuildQueue {
                     yanked.is_some(),
                 );
 
-                if let Err(err) = cdn::invalidate_crate(&self.config, &self.cdn, &release.name) {
+                if let Err(err) =
+                    cdn::queue_crate_invalidation(&mut *conn, &self.config, &release.name)
+                {
                     report_error(&err);
                 }
             }
@@ -570,11 +571,7 @@ mod tests {
             assert_eq!(metrics.failed_builds.get(), 1);
 
             // no invalidations were run since we don't have a distribution id configured
-            assert!(matches!(*env.cdn(), CdnBackend::Dummy(_)));
-            if let CdnBackend::Dummy(ref invalidation_requests) = *env.cdn() {
-                let ir = invalidation_requests.lock().unwrap();
-                assert!(ir.is_empty());
-            }
+            assert!(cdn::queued_or_active_crate_invalidations(&mut *env.db().conn())?.is_empty());
 
             Ok(())
         })
@@ -593,55 +590,31 @@ mod tests {
             queue.add_crate("will_succeed", "1.0.0", -1, None)?;
             queue.add_crate("will_fail", "1.0.0", 0, None)?;
 
-            assert!(matches!(*env.cdn(), CdnBackend::Dummy(_)));
-            if let CdnBackend::Dummy(ref invalidation_requests) = *env.cdn() {
-                let ir = invalidation_requests.lock().unwrap();
-                assert!(ir.is_empty());
-            }
+            let mut conn = env.db().conn();
+            cdn::queued_or_active_crate_invalidations(&mut *conn)?.is_empty();
 
             queue.process_next_crate(|krate| {
                 assert_eq!("will_succeed", krate.name);
                 Ok(())
             })?;
-            if let CdnBackend::Dummy(ref invalidation_requests) = *env.cdn() {
-                let ir = invalidation_requests.lock().unwrap();
-                assert_eq!(
-                    *ir,
-                    [
-                        ("distribution_id_web".into(), "/will_succeed*".into()),
-                        ("distribution_id_web".into(), "/crate/will_succeed*".into()),
-                        (
-                            "distribution_id_static".into(),
-                            "/rustdoc/will_succeed*".into()
-                        ),
-                    ]
-                );
-            }
+
+            let queued_invalidations = cdn::queued_or_active_crate_invalidations(&mut *conn)?;
+            assert_eq!(queued_invalidations.len(), 3);
+            assert!(queued_invalidations
+                .iter()
+                .all(|i| i.krate == "will_succeed"));
 
             queue.process_next_crate(|krate| {
                 assert_eq!("will_fail", krate.name);
                 anyhow::bail!("simulate a failure");
             })?;
-            if let CdnBackend::Dummy(ref invalidation_requests) = *env.cdn() {
-                let ir = invalidation_requests.lock().unwrap();
-                assert_eq!(
-                    *ir,
-                    [
-                        ("distribution_id_web".into(), "/will_succeed*".into()),
-                        ("distribution_id_web".into(), "/crate/will_succeed*".into()),
-                        (
-                            "distribution_id_static".into(),
-                            "/rustdoc/will_succeed*".into()
-                        ),
-                        ("distribution_id_web".into(), "/will_fail*".into()),
-                        ("distribution_id_web".into(), "/crate/will_fail*".into()),
-                        (
-                            "distribution_id_static".into(),
-                            "/rustdoc/will_fail*".into()
-                        ),
-                    ]
-                );
-            }
+
+            let queued_invalidations = cdn::queued_or_active_crate_invalidations(&mut *conn)?;
+            assert_eq!(queued_invalidations.len(), 6);
+            assert!(queued_invalidations
+                .iter()
+                .skip(3)
+                .all(|i| i.krate == "will_fail"));
 
             Ok(())
         })