Auto merge of #2269 - rust-lang:sg-update-swirl-2020-03-10, r=jtgeibel

Update swirl

This removes some of the boilerplate we had to add last night for
manually managing connection pools with swirl. Swirl now can
automatically manage its own connection pool. It can be given either a
database URL, a pool builder, or a fully built pool (which is used in
tests). If it's given one of the first two, it will configure the
connection pool to have a size of 2x the thread count. I've also let it
go up to the default number of threads, which is the number of CPU
cores. Since this would mean the pool size grows pretty dramatically,
I've configured it to not try to keep any idle connections (meaning it
will try to establish the connection when asked for one, not before).

Additionally, we no longer need the connection pool to live on our
environment, we can take either a connection or a pool as an argument to
a background job, and swirl will automatically take it from the pool. In
the future I'm going to try and have swirl make some more intelligent
decisions on this, so we could do something like have 2 threads only run
jobs that don't need a connection (like publish!)

Heroku doesn't guarantee the number of logical cores on standard dynos,
but it appears to have consistently been 4 for a while. This means we'll
be doubling the number of background workers, and potentially using up
to 8 database connections on worker. Now that we've moved some web
traffic to the replica, we may want to lower the pool size for the
primary db on the web servers to compensate.

Author: bors

Cargo.lock

@@ -48,7 +48,7 @@ dotenv = "0.15"
 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 = { git = "https://github.com/sgrif/swirl.git", rev = "6ef8c4cd" }
 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 Request) -> AppResult<Response> {
                     .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 Request) -> AppResult<Response> {
             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 Request, yanked: bool) -> AppResult<Response> {
 
     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(())
 }