Add a binary for running background jobs and monitoring their status

These binaries are pretty simple (and a bit spaghetti) for the time
being, since we have so little actually using the background job
framework. The runner is meant to be run on a worker dyno continuously,
while the monitor should be run by a cron-like tool roughly every 5
minutes.

We should move `update-downloads` to be scheduled as well, since it
spends so little time actually working.

Author: sgrif

Procfile

@@ -1,2 +1,3 @@
 web: bin/diesel migration run && bin/start-nginx ./target/release/server
 worker: ./target/release/update-downloads daemon 300
+background_worker: ./target/release/background-worker

src/bin/background-worker.rs

@@ -0,0 +1,47 @@
+// Runs enqueued background jobs
+//
+// This binary will loop until interrupted. Every second, it will attempt to
+// run any jobs in the background queue. Panics if attempting to count
+// available jobs fails.
+//
+// Usage:
+//      cargo run --bin background-worker
+
+#![deny(warnings)]
+
+use cargo_registry::{background, background_jobs::*, db};
+use diesel::r2d2;
+use std::env;
+use std::thread::sleep;
+use std::time::Duration;
+
+fn main() {
+    let config = cargo_registry::Config::default();
+
+    let username = env::var("GIT_HTTP_USER");
+    let password = env::var("GIT_HTTP_PWD");
+    let credentials = match (username, password) {
+        (Ok(u), Ok(p)) => Some((u, p)),
+        _ => None,
+    };
+    let environment = Environment {
+        index_location: config.index_location,
+        credentials,
+    };
+
+    // We're only using 1 thread, so we only need 1 connection
+    let db_config = r2d2::Pool::builder().max_size(1);
+    let db_pool = db::diesel_pool(&config.db_url, db_config);
+
+    let builder = background::Runner::builder(db_pool, environment).thread_count(1);
+    let runner = job_runner(builder);
+
+    println!("Runner booted, running jobs");
+
+    loop {
+        runner
+            .run_all_pending_jobs()
+            .expect("Could not begin running jobs");
+        sleep(Duration::from_secs(1));
+    }
+}

src/bin/monitor.rs

@@ -0,0 +1,73 @@
+//! Checks for any invariants we expect to be true, and pages whoever is on call
+//! if they are not.
+//!
+//! Usage:
+//!     cargo run --bin monitor
+
+#![deny(warnings)]
+
+#[macro_use]
+extern crate serde_derive;
+
+mod on_call;
+
+use cargo_registry::{db, util::CargoResult};
+use diesel::prelude::*;
+use std::env;
+
+fn main() -> CargoResult<()> {
+    let conn = db::connect_now()?;
+
+    check_stalled_background_jobs(&conn)?;
+    Ok(())
+}
+
+fn check_stalled_background_jobs(conn: &PgConnection) -> CargoResult<()> {
+    use cargo_registry::schema::background_jobs::dsl::*;
+    use diesel::dsl::*;
+
+    const BACKGROUND_JOB_KEY: &str = "background_jobs";
+
+    println!("Checking for stalled background jobs");
+
+    let max_job_time = env::var("MAX_JOB_TIME")
+        .map(|s| s.parse::<i32>().unwrap())
+        .unwrap_or(15);
+
+    let stalled_job_count = background_jobs
+        .filter(created_at.lt(now - max_job_time.minutes()))
+        .count()
+        .get_result::<i64>(conn)?;
+
+    let event = if stalled_job_count > 0 {
+        on_call::Event::Trigger {
+            incident_key: Some(BACKGROUND_JOB_KEY.into()),
+            description: format!(
+                "{} jobs have been in the queue for more than {} minutes",
+                stalled_job_count, max_job_time
+            ),
+        }
+    } else {
+        on_call::Event::Resolve {
+            incident_key: BACKGROUND_JOB_KEY.into(),
+            description: Some("No stalled background jobs".into()),
+        }
+    };
+
+    log_and_trigger_event(event)?;
+    Ok(())
+}
+
+fn log_and_trigger_event(event: on_call::Event) -> CargoResult<()> {
+    match event {
+        on_call::Event::Trigger {
+            ref description, ..
+        } => println!("Paging on-call: {}", description),
+        on_call::Event::Resolve {
+            description: Some(ref description),
+            ..
+        } => println!("{}", description),
+        _ => {} // noop
+    }
+    event.send()
+}

src/bin/on_call/mod.rs

@@ -10,6 +10,7 @@ pub enum Event {
         incident_key: Option<String>,
         description: String,
     },
+    #[allow(dead_code)] // Not all binaries create Acknowledge events
     Acknowledge {
         incident_key: String,
         description: Option<String>,