Auto merge of #1804 - sgrif:sg-more-resilient-job-runner, r=jtgeibel

Make the job runner a bit more resilient to slow jobs or other errors

A brief incident was caused by #1798. A band-aid fix is in place, and
 #1803 (included in this branch) makes it possible to apply similar
band-aids in the future without requiring a rebuild of the code. This
commit attempts to better address the root problem though.

The short version (which is expanded on below, but not required to
understand this commit or why it's needed) is that `update_downloads`
takes longer than our job timeout to run. When we moved that task to a
background job, we did not increase the number of concurrent jobs, nor
did we increase the timeout. This meant that swirl timed out trying to
start new jobs, and our behavior in that case was to crash the process.
This would mean that `update_downloads` never completes, and remains at
the front of the queue. This PR addresses all 3 of the problematic
cases.

- Increasing concurrency
  - When this system was added, the only jobs we had were index updates.
    These want to be serial, so we set the thread pool size to 1. We
    added readme renderings, which probably should have been parallel,
    but only happen with crate publishes anyway so it was fine.
    `update_downloads` *always* takes longer than the timeout to run
    though. We can't have it block everything else while it's running.

    The main downside to this is that index updates are no longer
    guaranteed to run in serial, which means that if two crates are
    uploaded simultaneously one job will fail and will have to wait for
    a retry to update the index. In theory if a crate happened to be
    uploaded at the exact instant of the retry 7 or 8 times in a row
    this could even result in getting paged. This is exceptionally
    unlikely, and I'm not concerned about it for now. As more features
    land in swirl we may want to move index updates to their own queue
    or tweak the retry behavior on that job though.

    Swirl will eventually handle this for us by default, and we should
    use its defaults once that lands.
- Increasing the default timeout
  - 10s was a bit too aggressive. Fundamentally there is always a
    condition where we hit this timeout, and if the reason for hitting
    it is that we are receiving more jobs than we can process (either
    because of volume of jobs, or our jobs are too slow).

    The most common reason we would hit this is that all threads are
    occupied by a job which takes longer than the timeout to execute.
    Increasing the concurrency makes this less likely to occur since our
    jobs are low volume, but we were actually seeing this crash before
    the addition of `update_downloads` meaning that our other jobs are
    sometimes taking >10s to run. Increasing the concurrency beyond 2
    would make it extremely unlikely we will ever hit this, but since we
    theoretically can with a burst of crate uploads at any concurrency,
    I've also upped the timeout.
- Rebuild the runner a few times before crashing the process
  - This is the most important change, though it's the only one that
    wouldn't fix the problem by itself. The first two changes address
    why the problem occurred, this last change addresses why it placed
    us in an unrecoverable state.

    What would happen is we would time out trying to start another job
    after `update_downloads`, and then the process would crash. This
    would mean that `update_downloads` would never complete, so as soon
    as we restarted, we'd just try to run it again (I may also change
    swirl to increment the retry counter before even beginning to run
    the job, but there are issues with that which are out of scope for
    this commit to discuss).

    This commit changes the behavior to instead built a new runner
    (which means a new thread pool and DB pool) up to 5 times before
    crashing the process. This means that any spawned threads will get a
    bit more time to run before the process itself crashes, so any jobs
    clogging the runner still get a chance to complete. I've opted to
    have a hard limit on the number of failures in the runner to avoid
    potentially unbounded growth in DB connections. We do still want to
    eventually fail, since being unable to start jobs can indicate
    issues that are only solved by starting a new process or moving to
    another physical machine.

More specific technical details on the issue that are not required to review this PR, but may be interesting
--

I've written this issue up at sgrif/swirl#16
as well.

The main entry point for a Swirl runner today is `run_all_pending_jobs`.
This method is fairly low level. The intent is to eventually add a
"reasonable defaults" binary shipped with swirl, probably somewhat based
on what crates.io needs here. This method will run in a loop, attempting
to fully saturate its thread pool on each iteration. It will check the
number of availble threads, spawning that many tasks.

Each task that is spawned will quickly communicate back to the
coordinator via an mpsc channel. The coordinator keeps track of how many
messages it's expecting (we get exactly 1 message per spawned task). If
we aren't currently expecting any messages, and there are also 0
available threads, we will attempt to spawn 1 task no matter what. This
is to ensure we don't loop forever waiting for a free thread, and
respsect the given timeout.

We do this in a loop until we hear from a thread that there was no job
available, or receive an error (caused by a thread being unable to get a
DB connection, an error loading the job from the DB [which should only
happen if the DB has gone away], or if we time out waiting to hear back
at all).

That's exactly what happened in this case. We would see 1 available
thread, spawn 1 task, and have 1 pending message. The worker would
communicate back that it got a job. We'd loop. There are 0 available
threads. We are expecting 0 messages, so we spawn 1 task anyway. We are
now expecting 1 pending message. We block waiting for it. The only way
we will receive a message is for the job we started in the first
iteration to complete before the timeout. It doesn't, so
`run_all_pending_jobs` returns an error. Our runner was calling
`.expect` on that, so the process crashes.

This shows several issues both in the configuration that was being used
by crates.io, and also in Swirl itself. I discussed the configuration
issues above, but there are also questions WRT Swirl's design. The first
issue is whether this case should be separated from not getting a
response from the worker at all. The latter should *never* happen under
reasonable circumstances, so my gut is that we can assume if it does
happen it was due to this case...

The second issue is that this was put us in an unrecoverable state
rather than causing one class of issues to fail to run. This could be
prevented by increasing the retry counter outside of a transaction
before running the job. This has issues though, which are out of scope
for this commit, but basically boil down to introducing non-atomic
pieces to an otherwise atomic operation.

Author: bors

src/background_jobs.rs

@@ -1,5 +1,5 @@
 use std::panic::AssertUnwindSafe;
-use std::sync::{Mutex, MutexGuard, PoisonError};
+use std::sync::{Arc, Mutex, MutexGuard, PoisonError};
 use swirl::PerformError;
 
 use crate::db::{DieselPool, DieselPooledConn};
@@ -21,14 +21,28 @@ impl swirl::db::DieselPool for DieselPool {
 
 #[allow(missing_debug_implementations)]
 pub struct Environment {
-    index: Mutex<Repository>,
+    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>,
 }
 
+// FIXME: AssertUnwindSafe should be `Clone`, this can be replaced with
+// `#[derive(Clone)]` if that is fixed in the standard lib
+impl Clone for Environment {
+    fn clone(&self) -> Self {
+        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()),
+        }
+    }
+}
+
 impl Environment {
     pub fn new(
         index: Repository,
@@ -38,7 +52,7 @@ impl Environment {
         http_client: reqwest::Client,
     ) -> Self {
         Self {
-            index: Mutex::new(index),
+            index: Arc::new(Mutex::new(index)),
             credentials,
             connection_pool: AssertUnwindSafe(connection_pool),
             uploader,

src/bin/background-worker.rs

@@ -1,11 +1,14 @@
-// Runs enqueued background jobs
-//
-// This binary will loop until interrupted. Every second, it will attempt to
-// run any jobs in the background queue. Panics if attempting to count
-// available jobs fails.
-//
-// Usage:
-//      cargo run --bin background-worker
+//! Runs enqueued background jobs
+//!
+//! This binary will loop until interrupted. It will run all jobs in the
+//! background queue, sleeping for 1 second whenever the queue is empty. If we
+//! are unable to spawn workers to run jobs (either because we couldn't connect
+//! to the DB, an error occurred while loading, or we just never heard back from
+//! the worker thread), we will rebuild the runner and try again up to 5 times.
+//! After the 5th occurrance, we will panic.
+//!
+//! Usage:
+//!      cargo run --bin background-worker
 
 #![deny(warnings, clippy::all, rust_2018_idioms)]
 
@@ -20,8 +23,13 @@ fn main() {
 
     let config = cargo_registry::Config::default();
 
-    // We're only using 1 thread, so we only need 2 connections
-    let db_config = r2d2::Pool::builder().max_size(2);
+    // 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 username = dotenv::var("GIT_HTTP_USER");
@@ -32,7 +40,7 @@ fn main() {
     };
 
     let job_start_timeout = dotenv::var("BACKGROUND_JOB_TIMEOUT")
-        .unwrap_or_else(|_| "10".into())
+        .unwrap_or_else(|_| "30".into())
         .parse()
         .expect("Invalid value for `BACKGROUND_JOB_TIMEOUT`");
 
@@ -48,17 +56,31 @@ fn main() {
         reqwest::Client::new(),
     );
 
-    let runner = swirl::Runner::builder(db_pool, environment)
-        .thread_count(1)
-        .job_start_timeout(Duration::from_secs(job_start_timeout))
-        .build();
+    let build_runner = || {
+        swirl::Runner::builder(db_pool.clone(), environment.clone())
+            .thread_count(2)
+            .job_start_timeout(Duration::from_secs(job_start_timeout))
+            .build()
+    };
+    let mut runner = build_runner();
 
     println!("Runner booted, running jobs");
 
+    let mut failure_count = 0;
+
     loop {
-        runner
-            .run_all_pending_jobs()
-            .expect("Could not begin running jobs");
+        if let Err(e) = runner.run_all_pending_jobs() {
+            failure_count += 1;
+            if failure_count < 5 {
+                eprintln!(
+                    "Error running jobs (n = {}) -- retrying: {:?}",
+                    failure_count, e,
+                );
+                runner = build_runner();
+            } else {
+                panic!("Failed to begin running jobs 5 times. Restarting the process");
+            }
+        }
         sleep(Duration::from_secs(1));
     }
 }