Update swirl

Cargo.toml

@@ -48,7 +48,7 @@ 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 = "de5d8bb" }
+swirl = { path = "../swirl/swirl" }
 serde_json = "1.0.0"
 serde = { version = "1.0.0", features = ["derive"] }
 chrono = { version = "0.4.0", features = ["serde"] }

src/background_jobs.rs

@@ -1,30 +1,27 @@
 use std::panic::AssertUnwindSafe;
 use std::sync::{Arc, Mutex, MutexGuard, PoisonError};
-use swirl::PerformError;
 
 use crate::db::{DieselPool, DieselPooledConn};
 use crate::git::Repository;
 use crate::uploaders::Uploader;
-use crate::util::errors::{CargoErrToStdErr, CargoResult};
+use crate::util::errors::CargoResult;
 
 impl<'a> swirl::db::BorrowedConnection<'a> for DieselPool {
     type Connection = DieselPooledConn<'a>;
 }
 
 impl swirl::db::DieselPool for DieselPool {
-    type Error = CargoErrToStdErr;
+    type Error = diesel::r2d2::PoolError;
 
     fn get(&self) -> Result<swirl::db::DieselPooledConn<'_, Self>, Self::Error> {
-        self.get().map_err(CargoErrToStdErr)
+        self.get()
     }
 }
 
 #[allow(missing_debug_implementations)]
 pub struct Environment {
     index: Arc<Mutex<Repository>>,
     pub credentials: Option<(String, String)>,
-    // FIXME: https://github.com/sfackler/r2d2/pull/70
-    pub connection_pool: AssertUnwindSafe<DieselPool>,
     pub uploader: Uploader,
     http_client: AssertUnwindSafe<reqwest::Client>,
 }
@@ -36,7 +33,6 @@ impl Clone for Environment {
         Self {
             index: self.index.clone(),
             credentials: self.credentials.clone(),
-            connection_pool: AssertUnwindSafe(self.connection_pool.0.clone()),
             uploader: self.uploader.clone(),
             http_client: AssertUnwindSafe(self.http_client.0.clone()),
         }
@@ -47,14 +43,12 @@ impl Environment {
     pub fn new(
         index: Repository,
         credentials: Option<(String, String)>,
-        connection_pool: DieselPool,
         uploader: Uploader,
         http_client: reqwest::Client,
     ) -> Self {
         Self {
             index: Arc::new(Mutex::new(index)),
             credentials,
-            connection_pool: AssertUnwindSafe(connection_pool),
             uploader,
             http_client: AssertUnwindSafe(http_client),
         }
@@ -66,12 +60,6 @@ impl Environment {
             .map(|(u, p)| (u.as_str(), p.as_str()))
     }
 
-    pub fn connection(&self) -> Result<DieselPooledConn<'_>, PerformError> {
-        self.connection_pool
-            .get()
-            .map_err(|e| CargoErrToStdErr(e).into())
-    }
-
     pub fn lock_index(&self) -> CargoResult<MutexGuard<'_, Repository>> {
         let repo = self.index.lock().unwrap_or_else(PoisonError::into_inner);
         repo.reset_head()?;

src/bin/background-worker.rs

@@ -14,24 +14,14 @@
 
 use cargo_registry::git::Repository;
 use cargo_registry::{background_jobs::*, db};
-use diesel::r2d2;
 use std::thread::sleep;
 use std::time::Duration;
 
 fn main() {
     println!("Booting runner");
 
     let config = cargo_registry::Config::default();
-
-    // 2x the thread pool size -- not all our jobs need a DB connection,
-    // but we want to always be able to run our jobs in parallel, rather
-    // than adjusting based on how many concurrent jobs need a connection.
-    // Eventually swirl will do this for us, and this will be the default
-    // -- we should just let it do a thread pool size of CPU count, and a
-    // a connection pool size of 2x that when that lands.
-    let db_config = r2d2::Pool::builder().max_size(4);
-    let db_pool = db::diesel_pool(&config.db_url, config.env, db_config);
-
+    let db_url = db::database_url(&config.db_url);
     let username = dotenv::var("GIT_HTTP_USER");
     let password = dotenv::var("GIT_HTTP_PWD");
     let credentials = match (username, password) {
@@ -51,13 +41,13 @@ fn main() {
     let environment = Environment::new(
         repository,
         credentials,
-        db_pool.clone(),
         config.uploader,
         reqwest::Client::new(),
     );
 
     let build_runner = || {
-        swirl::Runner::builder(db_pool.clone(), environment.clone())
+        swirl::Runner::builder(environment.clone())
+            .database_url(db_url.clone())
             .thread_count(2)
             .job_start_timeout(Duration::from_secs(job_start_timeout))
             .build()

src/bin/enqueue-job.rs

@@ -1,29 +1,20 @@
-use cargo_registry::util::{human, CargoError, CargoResult};
+use cargo_registry::util::{human, CargoResult};
 use cargo_registry::{db, env, tasks};
-use diesel::PgConnection;
+use swirl::Job;
 
 fn main() -> CargoResult<()> {
     let conn = db::connect_now()?;
     let mut args = std::env::args().skip(1);
     match &*args.next().unwrap_or_default() {
-        "update_downloads" => tasks::update_downloads().enqueue(&conn),
+        "update_downloads" => tasks::update_downloads().enqueue(&conn)?,
         "dump_db" => {
             let database_url = args.next().unwrap_or_else(|| env("DATABASE_URL"));
             let target_name = args
                 .next()
                 .unwrap_or_else(|| String::from("db-dump.tar.gz"));
-            tasks::dump_db(database_url, target_name).enqueue(&conn)
+            tasks::dump_db(database_url, target_name).enqueue(&conn)?
         }
-        other => Err(human(&format!("Unrecognized job type `{}`", other))),
+        other => return Err(human(&format!("Unrecognized job type `{}`", other))),
     }
+    Ok(())
 }
-
-/// Helper to map the `PerformError` returned by `swirl::Job::enqueue()` to a
-/// `CargoError`. Can be removed once `map_err()` isn't needed any more.
-trait Enqueue: swirl::Job {
-    fn enqueue(self, conn: &PgConnection) -> CargoResult<()> {
-        <Self as swirl::Job>::enqueue(self, conn).map_err(|e| CargoError::from_std_error(e))
-    }
-}
-
-impl<J: swirl::Job> Enqueue for J {}

src/controllers/krate/publish.rs

@@ -10,7 +10,7 @@ use crate::models::dependency;
 use crate::models::{Badge, Category, Keyword, NewCrate, NewVersion, Rights, User};
 use crate::render;
 use crate::util::{read_fill, read_le_u32};
-use crate::util::{CargoError, ChainError, Maximums};
+use crate::util::{ChainError, Maximums};
 use crate::views::{EncodableCrateUpload, GoodCrate, PublishWarnings};
 
 /// Handles the `PUT /crates/new` route.
@@ -167,8 +167,7 @@ pub fn publish(req: &mut dyn Request) -> CargoResult<Response> {
                     .unwrap_or_else(|| String::from("README.md")),
                 repo,
             )
-            .enqueue(&conn)
-            .map_err(|e| CargoError::from_std_error(e))?;
+            .enqueue(&conn)?;
         }
 
         let cksum = app
@@ -189,9 +188,7 @@ pub fn publish(req: &mut dyn Request) -> CargoResult<Response> {
             yanked: Some(false),
             links,
         };
-        git::add_crate(git_crate)
-            .enqueue(&conn)
-            .map_err(|e| CargoError::from_std_error(e))?;
+        git::add_crate(git_crate).enqueue(&conn)?;
 
         // The `other` field on `PublishWarnings` was introduced to handle a temporary warning
         // that is no longer needed. As such, crates.io currently does not return any `other`

src/controllers/version/yank.rs

@@ -6,7 +6,6 @@ use super::version_and_crate;
 use crate::controllers::prelude::*;
 use crate::git;
 use crate::models::Rights;
-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,9 +35,7 @@ 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(krate.name, version, yanked)
-        .enqueue(&conn)
-        .map_err(|e| CargoError::from_std_error(e))?;
+    git::yank(krate.name, version, yanked).enqueue(&conn)?;
 
     #[derive(Serialize)]
     struct R {

src/db.rs

@@ -18,7 +18,7 @@ pub enum DieselPool {
 }
 
 impl DieselPool {
-    pub fn get(&self) -> CargoResult<DieselPooledConn<'_>> {
+    pub fn get(&self) -> Result<DieselPooledConn<'_>, diesel::r2d2::PoolError> {
         match self {
             DieselPool::Pool(pool) => Ok(DieselPooledConn::Pool(pool.get()?)),
             DieselPool::Test(conn) => Ok(DieselPooledConn::Test(conn.lock())),
@@ -57,29 +57,30 @@ impl Deref for DieselPooledConn<'_> {
 }
 
 pub fn connect_now() -> ConnectionResult<PgConnection> {
-    let mut url = Url::parse(&crate::env("DATABASE_URL")).expect("Invalid database URL");
+    let url = database_url(&crate::env("DATABASE_URL"));
+    PgConnection::establish(&url)
+}
+
+pub fn database_url(url: &str) -> String {
+    let mut url = Url::parse(url).expect("Invalid database URL");
     if dotenv::var("HEROKU").is_ok() && !url.query_pairs().any(|(k, _)| k == "sslmode") {
         url.query_pairs_mut().append_pair("sslmode", "require");
     }
-    PgConnection::establish(&url.to_string())
+    url.into_string()
 }
 
 pub fn diesel_pool(
     url: &str,
     env: Env,
     config: r2d2::Builder<ConnectionManager<PgConnection>>,
 ) -> DieselPool {
-    let mut url = Url::parse(url).expect("Invalid database URL");
-    if dotenv::var("HEROKU").is_ok() && !url.query_pairs().any(|(k, _)| k == "sslmode") {
-        url.query_pairs_mut().append_pair("sslmode", "require");
-    }
+    let url = database_url(url);
 
     if env == Env::Test {
-        let conn =
-            PgConnection::establish(&url.into_string()).expect("failed to establish connection");
+        let conn = PgConnection::establish(&url).expect("failed to establish connection");
         DieselPool::test_conn(conn)
     } else {
-        let manager = ConnectionManager::new(url.into_string());
+        let manager = ConnectionManager::new(url);
         DieselPool::Pool(config.build(manager).unwrap())
     }
 }

src/git.rs

@@ -156,6 +156,7 @@ pub fn add_crate(env: &Environment, krate: Crate) -> Result<(), PerformError> {
 #[swirl::background_job]
 pub fn yank(
     env: &Environment,
+    conn: &PgConnection,
     krate: String,
     version: Version,
     yanked: bool,
@@ -165,8 +166,6 @@ pub fn yank(
     let repo = env.lock_index().map_err(std_error_no_send)?;
     let dst = repo.index_file(&krate);
 
-    let conn = env.connection()?;
-
     conn.transaction(|| {
         let yanked_in_db = versions::table
             .find(version.id)

src/render.rs

@@ -234,6 +234,7 @@ pub fn readme_to_html(text: &str, filename: &str, base_url: Option<&str>) -> Str
 #[swirl::background_job]
 pub fn render_and_upload_readme(
     env: &Environment,
+    conn: &PgConnection,
     version_id: i32,
     text: String,
     file_name: String,
@@ -244,7 +245,6 @@ pub fn render_and_upload_readme(
     use diesel::prelude::*;
 
     let rendered = readme_to_html(&text, &file_name, base_url.as_ref().map(String::as_str));
-    let conn = env.connection()?;
 
     conn.transaction(|| {
         Version::record_readme_rendering(version_id, &conn)?;

src/tasks/update_downloads.rs

@@ -8,9 +8,8 @@ use diesel::prelude::*;
 use swirl::PerformError;
 
 #[swirl::background_job]
-pub fn update_downloads(env: &Environment) -> Result<(), PerformError> {
-    let conn = env.connection()?;
-    update(&conn)?;
+pub fn update_downloads(_env: &Environment, conn: &PgConnection) -> Result<(), PerformError> {
+    update(conn)?;
     Ok(())
 }
 

src/tests/util.rs

@@ -71,7 +71,7 @@ impl Drop for TestAppInner {
         // Lazily run any remaining jobs
         if let Some(runner) = &self.runner {
             runner.run_all_pending_jobs().expect("Could not run jobs");
-            runner.assert_no_failed_jobs().expect("Failed jobs remain");
+            runner.check_for_failed_jobs().expect("Failed jobs remain");
         }
 
         // Manually verify that all jobs have completed successfully
@@ -184,7 +184,7 @@ impl TestApp {
 
         runner.run_all_pending_jobs().expect("Could not run jobs");
         runner
-            .assert_no_failed_jobs()
+            .check_for_failed_jobs()
             .expect("Could not determine if jobs failed");
     }
 
@@ -214,13 +214,13 @@ impl TestAppBuilder {
             let environment = Environment::new(
                 index,
                 None,
-                connection_pool.clone(),
                 app.config.uploader.clone(),
                 app.http_client().clone(),
             );
 
             Some(
-                Runner::builder(connection_pool, environment)
+                Runner::builder(environment)
+                    .connection_pool(connection_pool)
                     // We only have 1 connection in tests, so trying to run more than
                     // 1 job concurrently will just block
                     .thread_count(1)

src/util/errors.rs

@@ -51,19 +51,15 @@ impl dyn CargoError {
         self.get_type_id() == TypeId::of::<T>()
     }
 
-    pub fn from_std_error(err: Box<dyn Error + Send>) -> Box<dyn CargoError> {
-        Self::try_convert(&*err).unwrap_or_else(|| internal(&err))
-    }
-
-    fn try_convert(err: &(dyn Error + Send + 'static)) -> Option<Box<Self>> {
+    fn try_convert(err: &(dyn Error + 'static)) -> Option<Box<Self>> {
         match err.downcast_ref() {
             Some(DieselError::NotFound) => Some(Box::new(NotFound)),
             Some(DieselError::DatabaseError(_, info))
                 if info.message().ends_with("read-only transaction") =>
             {
                 Some(Box::new(ReadOnlyMode))
             }
-            _ => None,
+            _ => err.source().and_then(Self::try_convert),
         }
     }
 }