Move index updates off the web server

Cargo.toml

@@ -63,6 +63,7 @@ tempdir = "0.3.7"
 parking_lot = "0.7.1"
 jemallocator = { version = "0.1.8", features = ['unprefixed_malloc_on_supported_platforms', 'profiling'] }
 jemalloc-ctl = "0.2.0"
+threadpool = "1.7"
 
 lettre = {git = "https://github.com/lettre/lettre", version = "0.9"}
 lettre_email = {git = "https://github.com/lettre/lettre", version = "0.9"}

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

migrations/2018-05-03-150523_create_jobs/down.sql

@@ -0,0 +1 @@
+DROP TABLE background_jobs;

migrations/2018-05-03-150523_create_jobs/up.sql

@@ -0,0 +1,8 @@
+CREATE TABLE background_jobs (
+  id BIGSERIAL PRIMARY KEY,
+  job_type TEXT NOT NULL,
+  data JSONB NOT NULL,
+  retries INTEGER NOT NULL DEFAULT 0,
+  last_retry TIMESTAMP NOT NULL DEFAULT '1970-01-01',
+  created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP
+);

script/ci/prune-cache.sh

@@ -7,7 +7,7 @@ du -hs target/debug
 
 crate_name="cargo-registry"
 test_name="all"
-bin_names="delete-crate delete-version populate render-readmes server test-pagerduty transfer-crates update-downloads"
+bin_names="delete-crate delete-version populate render-readmes server test-pagerduty transfer-crates update-downloads background-worker monitor"
 
 normalized_crate_name=${crate_name//-/_}
 rm -v target/debug/$normalized_crate_name-*

src/app.rs

@@ -1,12 +1,7 @@
 //! Application-wide components in a struct accessible from each request
 
 use crate::{db, util::CargoResult, Config, Env};
-use std::{
-    env,
-    path::PathBuf,
-    sync::{Arc, Mutex},
-    time::Duration,
-};
+use std::{env, path::PathBuf, sync::Arc, time::Duration};
 
 use diesel::r2d2;
 use scheduled_thread_pool::ScheduledThreadPool;
@@ -25,10 +20,8 @@ pub struct App {
     /// A unique key used with conduit_cookie to generate cookies
     pub session_key: String,
 
-    /// The crate index git repository
-    pub git_repo: Mutex<git2::Repository>,
-
     /// The location on disk of the checkout of the crate index git repository
+    /// Only used in the development environment.
     pub git_repo_checkout: PathBuf,
 
     /// The server configuration
@@ -86,13 +79,10 @@ impl App {
             .connection_customizer(Box::new(db::SetStatementTimeout(db_connection_timeout)))
             .thread_pool(thread_pool);
 
-        let repo = git2::Repository::open(&config.git_repo_checkout).unwrap();
-
         App {
             diesel_database: db::diesel_pool(&config.db_url, config.env, diesel_db_config),
             github,
             session_key: config.session_key.clone(),
-            git_repo: Mutex::new(repo),
             git_repo_checkout: config.git_repo_checkout.clone(),
             config: config.clone(),
         }

src/background/job.rs

@@ -0,0 +1,26 @@
+use diesel::PgConnection;
+use serde::{de::DeserializeOwned, Serialize};
+
+use super::storage;
+use crate::util::CargoResult;
+
+/// A background job, meant to be run asynchronously.
+pub trait Job: Serialize + DeserializeOwned {
+    /// The environment this job is run with. This is a struct you define,
+    /// which should encapsulate things like database connection pools, any
+    /// configuration, and any other static data or shared resources.
+    type Environment;
+
+    /// The key to use for storing this job, and looking it up later.
+    ///
+    /// Typically this is the name of your struct in `snake_case`
+    const JOB_TYPE: &'static str;
+
+    /// Enqueue this job to be run at some point in the future.
+    fn enqueue(self, conn: &PgConnection) -> CargoResult<()> {
+        storage::enqueue_job(conn, self)
+    }
+
+    /// The logic involved in actually performing this job.
+    fn perform(self, env: &Self::Environment) -> CargoResult<()>;
+}

src/background/mod.rs

@@ -0,0 +1,8 @@
+mod job;
+mod registry;
+mod runner;
+mod storage;
+
+pub use self::job::*;
+pub use self::registry::Registry;
+pub use self::runner::*;

src/background/registry.rs

@@ -0,0 +1,46 @@
+#![allow(clippy::new_without_default)] // https://github.com/rust-lang/rust-clippy/issues/3632
+
+use serde_json;
+use std::collections::HashMap;
+use std::panic::RefUnwindSafe;
+
+use super::Job;
+use crate::util::CargoResult;
+
+#[doc(hidden)]
+pub type PerformFn<Env> =
+    Box<dyn Fn(serde_json::Value, &Env) -> CargoResult<()> + RefUnwindSafe + Send + Sync>;
+
+#[derive(Default)]
+#[allow(missing_debug_implementations)] // Can't derive debug
+/// A registry of background jobs, used to map job types to concrege perform
+/// functions at runtime.
+pub struct Registry<Env> {
+    job_types: HashMap<&'static str, PerformFn<Env>>,
+}
+
+impl<Env> Registry<Env> {
+    /// Create a new, empty registry
+    pub fn new() -> Self {
+        Registry {
+            job_types: Default::default(),
+        }
+    }
+
+    /// Get the perform function for a given job type
+    pub fn get(&self, job_type: &str) -> Option<&PerformFn<Env>> {
+        self.job_types.get(job_type)
+    }
+
+    /// Register a new background job. This will override any existing
+    /// registries with the same `JOB_TYPE`, if one exists.
+    pub fn register<T: Job<Environment = Env>>(&mut self) {
+        self.job_types.insert(
+            T::JOB_TYPE,
+            Box::new(|data, env| {
+                let data = serde_json::from_value(data)?;
+                T::perform(data, env)
+            }),
+        );
+    }
+}