Maintain a persistent checkout of the git index

The time a full clone takes caused too long of a delay when publishing a
crate. This instead performs a full clone when the runner boots, and
then locks and does `git fetch && git reset --hard origin/master` at the
top of every job. The reset means we can (somewhat) scale to multiple
runners, and retains unwind safety in the environment.

Note that there can still be a delay measured in minutes if a publish is
performed right after the server boots, as the job runner starting up
does not block the web server from starting.

Author: sgrif

src/background_jobs.rs

@@ -1,9 +1,9 @@
 use std::panic::AssertUnwindSafe;
-use url::Url;
+use std::sync::{Mutex, MutexGuard};
 
 use crate::background::{Builder, Runner};
 use crate::db::{DieselPool, DieselPooledConn};
-use crate::git::{AddCrate, Yank};
+use crate::git::{AddCrate, Repository, Yank};
 use crate::util::CargoResult;
 
 pub fn job_runner(config: Builder<Environment>) -> Runner<Environment> {
@@ -12,20 +12,20 @@ pub fn job_runner(config: Builder<Environment>) -> Runner<Environment> {
 
 #[allow(missing_debug_implementations)]
 pub struct Environment {
-    pub index_location: Url,
+    index: Mutex<Repository>,
     pub credentials: Option<(String, String)>,
     // FIXME: https://github.com/sfackler/r2d2/pull/70
     pub connection_pool: AssertUnwindSafe<DieselPool>,
 }
 
 impl Environment {
     pub fn new(
-        index_location: Url,
+        index: Repository,
         credentials: Option<(String, String)>,
         connection_pool: DieselPool,
     ) -> Self {
         Self {
-            index_location,
+            index: Mutex::new(index),
             credentials,
             connection_pool: AssertUnwindSafe(connection_pool),
         }
@@ -40,4 +40,11 @@ impl Environment {
     pub fn connection(&self) -> CargoResult<DieselPooledConn> {
         self.connection_pool.0.get().map_err(Into::into)
     }
+
+    pub fn lock_index(&self) -> CargoResult<MutexGuard<'_, Repository>> {
+        let repo = self.index.lock()
+            .unwrap_or_else(|e| e.into_inner());
+        repo.reset_head()?;
+        Ok(repo)
+    }
 }

src/bin/background-worker.rs

@@ -10,12 +10,15 @@
 #![deny(warnings)]
 
 use cargo_registry::{background, background_jobs::*, db};
+use cargo_registry::git::Repository;
 use diesel::r2d2;
 use std::env;
 use std::thread::sleep;
 use std::time::Duration;
 
 fn main() {
+    println!("Booting runner");
+
     let config = cargo_registry::Config::default();
 
     // We're only using 1 thread, so we only need 2 connections
@@ -28,7 +31,13 @@ fn main() {
         (Ok(u), Ok(p)) => Some((u, p)),
         _ => None,
     };
-    let environment = Environment::new(config.index_location, credentials, db_pool.clone());
+
+    println!("Cloning index");
+
+    let repository = Repository::open(&config.index_location)
+        .expect("Failed to clone index");
+
+    let environment = Environment::new(repository, credentials, db_pool.clone());
 
     let builder = background::Runner::builder(db_pool, environment).thread_count(1);
     let runner = job_runner(builder);

src/git.rs

@@ -39,13 +39,13 @@ pub struct Dependency {
     pub package: Option<String>,
 }
 
-struct Repository {
+pub struct Repository {
     checkout_path: TempDir,
     repository: git2::Repository,
 }
 
 impl Repository {
-    fn open(url: &Url) -> CargoResult<Self> {
+    pub fn open(url: &Url) -> CargoResult<Self> {
         let checkout_path = TempDir::new("git")?;
         let repository = git2::Repository::clone(url.as_str(), checkout_path.path())?;
 
@@ -120,6 +120,15 @@ impl Repository {
         }
         ref_status
     }
+
+    pub fn reset_head(&self) -> CargoResult<()> {
+        let mut origin = self.repository.find_remote("origin")?;
+        origin.fetch(&["refs/heads/*:refs/heads/*"], None, None)?;
+        let head = self.repository.head()?.target().unwrap();
+        let obj = self.repository.find_object(head, None)?;
+        self.repository.reset(&obj, git2::ResetType::Hard, None)?;
+        Ok(())
+    }
 }
 
 #[derive(Deserialize, Serialize)]
@@ -132,7 +141,7 @@ impl Job for AddCrate {
     const JOB_TYPE: &'static str = "add_crate";
 
     fn perform(self, env: &Self::Environment) -> CargoResult<()> {
-        let repo = Repository::open(&env.index_location)?;
+        let repo = env.lock_index()?;
         let dst = repo.index_file(&self.krate.name);
 
         // Add the crate to its relevant file
@@ -165,7 +174,7 @@ impl Job for Yank {
     const JOB_TYPE: &'static str = "yank";
 
     fn perform(self, env: &Self::Environment) -> CargoResult<()> {
-        let repo = Repository::open(&env.index_location)?;
+        let repo = env.lock_index()?;
         let dst = repo.index_file(&self.krate);
 
         let conn = env.connection()?;

src/middleware/run_pending_background_jobs.rs

@@ -2,6 +2,7 @@ use super::app::RequestApp;
 use super::prelude::*;
 use crate::background::Runner;
 use crate::background_jobs::*;
+use crate::git::Repository;
 
 pub struct RunPendingBackgroundJobs;
 
@@ -13,8 +14,10 @@ impl Middleware for RunPendingBackgroundJobs {
     ) -> Result<Response, Box<dyn Error + Send>> {
         let app = req.app();
         let connection_pool = app.diesel_database.clone();
+        let repo = Repository::open(&app.config.index_location)
+            .expect("Could not clone index");
         let environment = Environment::new(
-            app.config.index_location.clone(),
+            repo,
             None,
             app.diesel_database.clone(),
         );