Auto merge of #2432 - sgrif:sg-update-swirl-04-16-2020, r=jtgeibel

Update Swirl, take two

This is a second attempt at #2269, which had to be reverted shortly
after being deployed, as it caused jobs to fail to run in production.
This commit only differs from that PR in the revision of swirl used.

The cause of the issue was the change in the signature of
`update_downloads`, which does not need anything from the job
environment, and so it only takes a connection. This changed the
environment type for that job from `background_jobs::Environment` to
`()`, so our runner no longer knew how to run that job.

The smallest fix in crates.io would have been to add an unused
environment argument to `update_downloads`. However, having some jobs
not require a shared environment felt common enough that I fixed this in
swirl instead. sgrif/swirl#23 changed the
behavior so jobs with no environment (a.k.a. the environment type is
`()`) are always run, regardless of the environment type of the runner.

Since `update_downloads` is essentially a cron job, and is not enqueued
from the web server directly, this was not caught by our integration
suite. This case was caught quickly after being deployed, and only
affected a non-critical part of the service, so I've opted not to figure
out how to add this to our integration suite.

r? @jtgeibel

Author: bors

Cargo.lock

@@ -48,7 +48,7 @@ dotenv = "0.15"
 toml = "0.5"
 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 = { git = "https://github.com/sgrif/swirl.git", rev = "e87cf37" }
 serde_json = "1.0.0"
 serde = { version = "1.0.0", features = ["derive"] }
 chrono = { version = "0.4.0", features = ["serde"] }

Cargo.toml

@@ -24,8 +24,6 @@ impl swirl::db::DieselPool for DieselPool {
 #[allow(missing_debug_implementations)]
 pub struct Environment {
     index: Arc<Mutex<Repository>>,
-    // FIXME: https://github.com/sfackler/r2d2/pull/70
-    pub connection_pool: AssertUnwindSafe<DieselPool>,
     pub uploader: Uploader,
     http_client: AssertUnwindSafe<Client>,
 }
@@ -36,46 +34,29 @@ impl Clone for Environment {
     fn clone(&self) -> Self {
         Self {
             index: self.index.clone(),
-            connection_pool: AssertUnwindSafe(self.connection_pool.0.clone()),
             uploader: self.uploader.clone(),
             http_client: AssertUnwindSafe(self.http_client.0.clone()),
         }
     }
 }
 
 impl Environment {
-    pub fn new(
-        index: Repository,
-        connection_pool: DieselPool,
-        uploader: Uploader,
-        http_client: Client,
-    ) -> Self {
-        Self::new_shared(
-            Arc::new(Mutex::new(index)),
-            connection_pool,
-            uploader,
-            http_client,
-        )
+    pub fn new(index: Repository, uploader: Uploader, http_client: Client) -> Self {
+        Self::new_shared(Arc::new(Mutex::new(index)), uploader, http_client)
     }
 
     pub fn new_shared(
         index: Arc<Mutex<Repository>>,
-        connection_pool: DieselPool,
         uploader: Uploader,
         http_client: Client,
     ) -> Self {
         Self {
             index,
-            connection_pool: AssertUnwindSafe(connection_pool),
             uploader,
             http_client: AssertUnwindSafe(http_client),
         }
     }
 
-    pub fn connection(&self) -> Result<DieselPooledConn<'_>, PoolError> {
-        self.connection_pool.get()
-    }
-
     pub fn lock_index(&self) -> Result<MutexGuard<'_, Repository>, PerformError> {
         let repo = self.index.lock().unwrap_or_else(PoisonError::into_inner);
         repo.reset_head()?;

src/background_jobs.rs

@@ -24,6 +24,7 @@ fn main() {
     println!("Booting runner");
 
     let config = cargo_registry::Config::default();
+    let db_url = db::connection_url(&config.db_url);
 
     let job_start_timeout = dotenv::var("BACKGROUND_JOB_TIMEOUT")
         .unwrap_or_else(|_| "30".into())
@@ -39,22 +40,11 @@ fn main() {
     println!("Index cloned");
 
     let build_runner = || {
-        // 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 environment = Environment::new_shared(
-            repository.clone(),
-            db_pool.clone(),
-            config.uploader.clone(),
-            Client::new(),
-        );
-        swirl::Runner::builder(db_pool, environment)
-            .thread_count(2)
+        let environment =
+            Environment::new_shared(repository.clone(), config.uploader.clone(), Client::new());
+        let db_config = r2d2::Pool::builder().min_idle(Some(0));
+        swirl::Runner::builder(environment)
+            .connection_pool_builder(&db_url, db_config)
             .job_start_timeout(Duration::from_secs(job_start_timeout))
             .build()
     };

src/bin/background-worker.rs

@@ -183,8 +183,7 @@ pub fn publish(req: &mut dyn RequestExt) -> EndpointResult {
                     .unwrap_or_else(|| String::from("README.md")),
                 repo,
             )
-            .enqueue(&conn)
-            .map_err(|e| AppError::from_std_error(e))?;
+            .enqueue(&conn)?;
         }
 
         let cksum = app
@@ -204,9 +203,7 @@ pub fn publish(req: &mut dyn RequestExt) -> EndpointResult {
             yanked: Some(false),
             links,
         };
-        git::add_crate(git_crate)
-            .enqueue(&conn)
-            .map_err(|e| AppError::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/krate/publish.rs

@@ -44,9 +44,7 @@ fn modify_yank(req: &mut dyn RequestExt, yanked: bool) -> EndpointResult {
 
     insert_version_owner_action(&conn, version.id, user.id, ids.api_token_id(), action)?;
 
-    git::yank(krate.name, version, yanked)
-        .enqueue(&conn)
-        .map_err(|e| AppError::from_std_error(e))?;
+    git::yank(krate.name, version, yanked).enqueue(&conn)?;
 
     ok_true()
 }

src/controllers/version/yank.rs

@@ -63,22 +63,25 @@ pub fn connect_now() -> ConnectionResult<PgConnection> {
     PgConnection::establish(&url.to_string())
 }
 
-pub fn diesel_pool(
-    url: &str,
-    env: Env,
-    config: r2d2::Builder<ConnectionManager<PgConnection>>,
-) -> DieselPool {
+pub fn connection_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");
     }
+    url.into_string()
+}
 
+pub fn diesel_pool(
+    url: &str,
+    env: Env,
+    config: r2d2::Builder<ConnectionManager<PgConnection>>,
+) -> DieselPool {
+    let url = connection_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/db.rs

@@ -289,6 +289,7 @@ pub fn add_crate(env: &Environment, krate: Crate) -> Result<(), PerformError> {
 /// push the changes.
 #[swirl::background_job]
 pub fn yank(
+    conn: &PgConnection,
     env: &Environment,
     krate: String,
     version: Version,
@@ -299,8 +300,6 @@ pub fn yank(
     let repo = env.lock_index()?;
     let dst = repo.index_file(&krate);
 
-    let conn = env.connection()?;
-
     conn.transaction(|| {
         let yanked_in_db = versions::table
             .find(version.id)

src/git.rs

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

src/render.rs

@@ -1,5 +1,4 @@
 use crate::{
-    background_jobs::Environment,
     models::VersionDownload,
     schema::{crates, metadata, version_downloads, versions},
 };
@@ -8,8 +7,7 @@ use diesel::prelude::*;
 use swirl::PerformError;
 
 #[swirl::background_job]
-pub fn update_downloads(env: &Environment) -> Result<(), PerformError> {
-    let conn = env.connection()?;
+pub fn update_downloads(conn: &PgConnection) -> Result<(), PerformError> {
     update(&conn)?;
     Ok(())
 }

src/tasks/update_downloads.rs

@@ -77,7 +77,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
@@ -192,7 +192,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");
     }
 
@@ -223,24 +223,23 @@ impl TestAppBuilder {
         let (app, middle) = crate::build_app(self.config, self.proxy);
 
         let runner = if self.build_job_runner {
-            let connection_pool = app.primary_database.clone();
             let repository_config = RepositoryConfig {
                 index_location: Url::from_file_path(&git::bare()).unwrap(),
                 credentials: Credentials::Missing,
             };
             let index = WorkerRepository::open(&repository_config).expect("Could not clone index");
             let environment = Environment::new(
                 index,
-                connection_pool.clone(),
                 app.config.uploader.clone(),
                 app.http_client().clone(),
             );
 
             Some(
-                Runner::builder(connection_pool, environment)
+                Runner::builder(environment)
                     // We only have 1 connection in tests, so trying to run more than
                     // 1 job concurrently will just block
                     .thread_count(1)
+                    .connection_pool(app.primary_database.clone())
                     .job_start_timeout(Duration::from_secs(5))
                     .build(),
             )

src/tests/util.rs

@@ -97,10 +97,6 @@ impl dyn AppError {
         self.get_type_id() == TypeId::of::<T>()
     }
 
-    pub fn from_std_error(err: Box<dyn Error + Send>) -> Box<dyn AppError> {
-        Self::try_convert(&*err).unwrap_or_else(|| internal(&err))
-    }
-
     fn try_convert(err: &(dyn Error + Send + 'static)) -> Option<Box<Self>> {
         match err.downcast_ref() {
             Some(DieselError::NotFound) => Some(Box::new(NotFound)),