Auto merge of #1677 - sgrif:sg-update-swirl, r=jtgeibel

Update swirl

New features in this version:

- Codegen for jobs which reduces boilerplate when defining them
- Jobs no longer need to be registered with the runner
- Rework of how `run_all_pending_jobs` works so errors loading the job
  actually bubble up
- Configurable timeout for how long to wait for a job to begin running

We aren't fully taking advantage of the timeout yet, I'd eventually like
to try rebuilding the runner in-process a few times when an error
happens, and then crash the process if it fails a few times in a row.
This needs a bit more refactoring though (I'm not sure if we want to
keep the `Repository` in memory like we are now, and whether we want to
assume a full re-clone is needed on error). For now I've set it to
ensure our jobs don't hang forever though.

Warts in this version:

- Rust thinks imports only used in the job body are unused.
  (sgrif/swirl#6)
  - Worked around for now by moving the imports into the job body
- `swirl::Job` has to be imported by calling code
- Codegen assumes that `serde::Deserialize` and `serde::Serialize` are
  available. (sgrif/swirl#9)

Issues with this version:

- The new impl of `run_all_pending_jobs` will return an error if the DB
  is in read only mode. If the DB is read only, we couldn't have
  enqueued jobs anyway, so the workaround is "fine", but we need some
  more robust APIs in swirl to fix this. This only affects us in tests.
  - sgrif/swirl#8

Author: bors

Cargo.lock

@@ -48,10 +48,9 @@ dotenv = "0.11.0"
 toml = "0.4"
 diesel = { version = "1.4.0", features = ["postgres", "serde_json", "chrono", "r2d2"] }
 diesel_full_text_search = "1.0.0"
-swirl = { git = "https://github.com/sgrif/swirl.git", rev = "95d3a35bc39a7274335cad6d7cab64acd5eb3904" }
+swirl = { git = "https://github.com/sgrif/swirl.git", rev = "de5d8bb" }
 serde_json = "1.0.0"
-serde_derive = "1.0.0"
-serde = "1.0.0"
+serde = { version = "1.0.0", features = ["derive"] }
 chrono = { version = "0.4.0", features = ["serde"] }
 comrak = { version = "0.4.0", default-features = false }
 ammonia = "2.0.0"

Cargo.toml

@@ -1,24 +1,22 @@
 use std::panic::AssertUnwindSafe;
 use std::sync::{Mutex, MutexGuard};
-use swirl::{Builder, Runner};
 
 use crate::db::{DieselPool, DieselPooledConn};
-use crate::git::{AddCrate, Repository, Yank};
+use crate::git::Repository;
 use crate::util::errors::{CargoErrToStdErr, CargoResult};
 
-impl<'a> swirl::DieselPool<'a> for DieselPool {
+impl<'a> swirl::db::BorrowedConnection<'a> for DieselPool {
     type Connection = DieselPooledConn<'a>;
+}
+
+impl swirl::db::DieselPool for DieselPool {
     type Error = CargoErrToStdErr;
 
-    fn get(&'a self) -> Result<Self::Connection, Self::Error> {
+    fn get(&self) -> Result<swirl::db::DieselPooledConn<'_, Self>, Self::Error> {
         self.get().map_err(CargoErrToStdErr)
     }
 }
 
-pub fn job_runner(config: Builder<Environment, DieselPool>) -> Runner<Environment, DieselPool> {
-    config.register::<AddCrate>().register::<Yank>().build()
-}
-
 #[allow(missing_debug_implementations)]
 pub struct Environment {
     index: Mutex<Repository>,

src/background_jobs.rs

@@ -37,8 +37,10 @@ fn main() {
 
     let environment = Environment::new(repository, credentials, db_pool.clone());
 
-    let builder = swirl::Runner::builder(db_pool, environment).thread_count(1);
-    let runner = job_runner(builder);
+    let runner = swirl::Runner::builder(db_pool, environment)
+        .thread_count(1)
+        .job_start_timeout(Duration::from_secs(10))
+        .build();
 
     println!("Runner booted, running jobs");
 

src/bin/background-worker.rs

@@ -6,9 +6,6 @@
 
 #![deny(warnings)]
 
-#[macro_use]
-extern crate serde_derive;
-
 mod on_call;
 
 use cargo_registry::{db, util::CargoResult};

src/bin/monitor.rs

@@ -2,7 +2,7 @@ use cargo_registry::util::{internal, CargoResult};
 
 use reqwest::{header, StatusCode as Status};
 
-#[derive(Serialize, Debug)]
+#[derive(serde::Serialize, Debug)]
 #[serde(rename_all = "snake_case", tag = "event_type")]
 pub enum Event {
     Trigger {
@@ -54,14 +54,14 @@ impl Event {
     }
 }
 
-#[derive(Serialize, Debug)]
+#[derive(serde::Serialize, Debug)]
 struct FullEvent {
     service_key: String,
     #[serde(flatten)]
     event: Event,
 }
 
-#[derive(Deserialize, Debug)]
+#[derive(serde::Deserialize, Debug)]
 struct InvalidEvent {
     message: String,
     errors: Vec<String>,

src/bin/on_call/mod.rs

@@ -6,7 +6,7 @@
 #![deny(warnings)]
 
 #[macro_use]
-extern crate serde_derive;
+extern crate serde;
 
 use cargo_registry::{
     db,

src/bin/render-readmes.rs

@@ -7,9 +7,6 @@
 
 #![deny(warnings)]
 
-#[macro_use]
-extern crate serde_derive;
-
 mod on_call;
 
 use std::env::args;

src/bin/test-pagerduty.rs

@@ -1,18 +1,17 @@
 //! Functionality related to publishing a new crate or version of a crate.
 
+use hex::ToHex;
 use std::collections::HashMap;
 use std::sync::Arc;
-
-use hex::ToHex;
-
-use crate::git;
-use crate::render;
-use crate::util::{internal, ChainError, Maximums};
-use crate::util::{read_fill, read_le_u32};
+use swirl::Job;
 
 use crate::controllers::prelude::*;
+use crate::git;
 use crate::models::dependency;
 use crate::models::{Badge, Category, Keyword, NewCrate, NewVersion, Rights, User};
+use crate::render;
+use crate::util::{internal, CargoError, ChainError, Maximums};
+use crate::util::{read_fill, read_le_u32};
 use crate::views::{EncodableCrateUpload, GoodCrate, PublishWarnings};
 
 /// Handles the `PUT /crates/new` route.
@@ -196,12 +195,15 @@ pub fn publish(req: &mut dyn Request) -> CargoResult<Response> {
             yanked: Some(false),
             links,
         };
-        git::add_crate(&conn, git_crate).chain_error(|| {
-            internal(&format_args!(
-                "could not add crate `{}` to the git repo",
-                name
-            ))
-        })?;
+        git::add_crate(git_crate)
+            .enqueue(&conn)
+            .map_err(|e| CargoError::from_std_error(e))
+            .chain_error(|| {
+                internal(&format_args!(
+                    "could not add crate `{}` to the git repo",
+                    name
+                ))
+            })?;
 
         // Now that we've come this far, we're committed!
         crate_bomb.path = None;

src/controllers/krate/publish.rs

@@ -1,12 +1,12 @@
 //! Endpoints for yanking and unyanking specific versions of crates
 
-use crate::controllers::prelude::*;
+use swirl::Job;
 
+use super::version_and_crate;
+use crate::controllers::prelude::*;
 use crate::git;
-
 use crate::models::Rights;
-
-use super::version_and_crate;
+use crate::util::CargoError;
 
 /// Handles the `DELETE /crates/:crate_id/:version/yank` route.
 /// This does not delete a crate version, it makes the crate
@@ -36,7 +36,9 @@ fn modify_yank(req: &mut dyn Request, yanked: bool) -> CargoResult<Response> {
         return Err(human("must already be an owner to yank or unyank"));
     }
 
-    git::yank(&conn, krate.name, version, yanked)?;
+    git::yank(krate.name, version, yanked)
+        .enqueue(&conn)
+        .map_err(|e| CargoError::from_std_error(e))?;
 
     #[derive(Serialize)]
     struct R {