Auto merge of #2539 - jtgeibel:there-can-only-be-one, r=pietroalbini

Only enqueue 1 update_downloads job at a time

This ensures that if an `update_downloads` job is already running, a
duplicate job will not be enqueued. Currently, when multiple jobs are
running in parallel, they end up doing duplicate work resulting in
temporary overcounts that must be corrected in the next run. The
concurrent tasks also slow down the overall process and can result in
runaway performance problems as further jobs are spawned.

This commit also updates the monitoring to specifically check if the
update downloads job runs for too long (120 minutes by default). The
main check for stalled jobs will not trigger for `update_downloads` as
the row is locked for the duration of the job (and `skip_locked` is used
in that query).

r? @pietroalbini

Author: bors

src/bin/enqueue-job.rs

@@ -1,6 +1,8 @@
 #![deny(clippy::all)]
 
 use cargo_registry::{db, env, tasks, util::Error};
+use diesel::prelude::*;
+use swirl::schema::background_jobs::dsl::*;
 use swirl::Job;
 
 fn main() -> Result<(), Error> {
@@ -11,7 +13,20 @@ fn main() -> Result<(), Error> {
     println!("Enqueueing background job: {}", job);
 
     match &*job {
-        "update_downloads" => Ok(tasks::update_downloads().enqueue(&conn)?),
+        "update_downloads" => {
+            let count: i64 = background_jobs
+                .filter(job_type.eq("update_downloads"))
+                .count()
+                .get_result(&conn)
+                .unwrap();
+
+            if count > 0 {
+                println!("Did not enqueue update_downloads, existing job already in progress");
+                Ok(())
+            } else {
+                Ok(tasks::update_downloads().enqueue(&conn)?)
+            }
+        }
         "dump_db" => {
             let database_url = args.next().unwrap_or_else(|| env("READ_ONLY_REPLICA_URL"));
             let target_name = args

src/bin/monitor.rs

@@ -14,20 +14,30 @@ use diesel::prelude::*;
 fn main() -> Result<(), Error> {
     let conn = db::connect_now()?;
 
-    check_stalled_background_jobs(&conn)?;
+    check_failing_background_jobs(&conn)?;
+    check_stalled_update_downloads(&conn)?;
     check_spam_attack(&conn)?;
     Ok(())
 }
 
-fn check_stalled_background_jobs(conn: &PgConnection) -> Result<(), Error> {
+/// Check for old background jobs that are not currently running.
+///
+/// This check includes `skip_locked` in the query and will only trigger on
+/// enqueued jobs that have attempted to run and have failed (and are in the
+/// queue awaiting a retry).
+///
+/// Within the default 15 minute time, a job should have already had several
+/// failed retry attempts.
+fn check_failing_background_jobs(conn: &PgConnection) -> Result<(), Error> {
     use cargo_registry::schema::background_jobs::dsl::*;
     use diesel::dsl::*;
     use diesel::sql_types::Integer;
 
     const EVENT_KEY: &str = "background_jobs";
 
-    println!("Checking for stalled background jobs");
+    println!("Checking for failed background jobs");
 
+    // Max job execution time in minutes
     let max_job_time = dotenv::var("MAX_JOB_TIME")
         .map(|s| s.parse::<i32>().unwrap())
         .unwrap_or(15);
@@ -59,6 +69,44 @@ fn check_stalled_background_jobs(conn: &PgConnection) -> Result<(), Error> {
     Ok(())
 }
 
+/// Check for an `update_downloads` job that has run longer than expected
+fn check_stalled_update_downloads(conn: &PgConnection) -> Result<(), Error> {
+    use cargo_registry::schema::background_jobs::dsl::*;
+    use chrono::{DateTime, NaiveDateTime, Utc};
+
+    const EVENT_KEY: &str = "update_downloads_stalled";
+
+    println!("Checking for stalled background jobs");
+
+    // Max job execution time in minutes
+    let max_job_time = dotenv::var("MONITOR_MAX_UPDATE_DOWNLOADS_TIME")
+        .map(|s| s.parse::<u32>().unwrap() as i64)
+        .unwrap_or(120);
+
+    let start_time = background_jobs
+        .filter(job_type.eq("update_downloads"))
+        .select(created_at)
+        .first::<NaiveDateTime>(conn);
+
+    if let Ok(start_time) = start_time {
+        let start_time = DateTime::<Utc>::from_utc(start_time, Utc);
+        let minutes = Utc::now().signed_duration_since(start_time).num_minutes();
+
+        if minutes > max_job_time {
+            return log_and_trigger_event(on_call::Event::Trigger {
+                incident_key: Some(EVENT_KEY.into()),
+                description: format!("update_downloads job running for {} minutes", minutes),
+            });
+        }
+    };
+
+    log_and_trigger_event(on_call::Event::Resolve {
+        incident_key: EVENT_KEY.into(),
+        description: Some("No stalled update_downloads job".into()),
+    })
+}
+
+/// Check for known spam patterns
 fn check_spam_attack(conn: &PgConnection) -> Result<(), Error> {
     use cargo_registry::models::krate::canon_crate_name;
     use diesel::dsl::*;