Auto merge of #1670 - sgrif:sg-read-only-db, r=jtgeibel

Allow the database to be set in read only mode

This is in preparation for our next database upgrade. Last time we
needed to do this, we took the whole site down. This means that for 30
minutes, `cargo build` just didn't work. This is no longer something we
consider acceptable. While we could just take everything down *except*
`cargo build`, we can do better.

The upgrade process is basically:

- Set up a new replica of the primary database
- Wait until the replica is "caught up" (< 200 commits behind)
- Take down the site
- Upgrade the replica in place
- Verify the upgrade was successful
- Swap over to the replica
- Bring the site back up

There's no reason that we actually need to bring the site down though,
we just need to stop writing to the primary during the upgrade process.

This does that by setting the default transaction mode to READ ONLY, and
handling the error we will get back if we try to do a write.
Unfortunately, Diesel doesn't expose any API to give us the underlying
database error code, so we have to match on the error message instead.

This code assumes that:

- We are never manually setting a transaction to `READ WRITE`
- We are never manually setting a transaction to `READ ONLY`, and
  expecting write errors to bubble up to the user

We're not currently doing either of those things, and I can't imagine
that we'll start doing either one in the future.

During testing I found some issues with how this interacts with token
auth. The middleware was assuming that *any* error from
`User::find_by_api_token` was `diesel::NotFound`, and would proceed to
run the request (this finally explains a spurious failure I've been
getting about a poisoned transaction, it was probably a deadlock in the
API key update)

Because our tests run in a single transaction, we need to wrap that
update to ensure that the error doesn't poison the transaction. If we
didn't do anything else, we'd just assume that any user trying to do
something with token auth was not logged in, and would give a 403
instead of a 503. To fix this, I've set it up to fall back to a simple
find if the update fails.

This does *not* make the download endpoint work when we're in read only
mode, there is a separate issue for that which will come in a separate
PR. I have, however, added an explicit test to ensure it works in read
only mode.

Author: bors

src/app.rs

@@ -69,14 +69,19 @@ impl App {
             (_, Env::Test) => 1,
             _ => 30,
         };
+        let read_only_mode = dotenv::var("READ_ONLY_MODE").is_ok();
+        let connection_config = db::ConnectionConfig {
+            statement_timeout: db_connection_timeout,
+            read_only: read_only_mode,
+        };
 
         let thread_pool = Arc::new(ScheduledThreadPool::new(db_helper_threads));
 
         let diesel_db_config = r2d2::Pool::builder()
             .max_size(db_pool_size)
             .min_idle(db_min_idle)
             .connection_timeout(Duration::from_secs(db_connection_timeout))
-            .connection_customizer(Box::new(db::SetStatementTimeout(db_connection_timeout)))
+            .connection_customizer(Box::new(connection_config))
             .thread_pool(thread_pool);
 
         App {

src/db.rs

@@ -100,15 +100,26 @@ impl<T: Request + ?Sized> RequestTransaction for T {
 }
 
 #[derive(Debug, Clone, Copy)]
-pub struct SetStatementTimeout(pub u64);
+pub struct ConnectionConfig {
+    pub statement_timeout: u64,
+    pub read_only: bool,
+}
 
-impl CustomizeConnection<PgConnection, r2d2::Error> for SetStatementTimeout {
+impl CustomizeConnection<PgConnection, r2d2::Error> for ConnectionConfig {
     fn on_acquire(&self, conn: &mut PgConnection) -> Result<(), r2d2::Error> {
         use diesel::sql_query;
 
-        sql_query(format!("SET statement_timeout = {}", self.0 * 1000))
-            .execute(conn)
-            .map_err(r2d2::Error::QueryError)?;
+        sql_query(format!(
+            "SET statement_timeout = {}",
+            self.statement_timeout * 1000
+        ))
+        .execute(conn)
+        .map_err(r2d2::Error::QueryError)?;
+        if self.read_only {
+            sql_query("SET default_transaction_read_only = 't'")
+                .execute(conn)
+                .map_err(r2d2::Error::QueryError)?;
+        }
         Ok(())
     }
 }

src/git.rs

@@ -12,7 +12,7 @@ use url::Url;
 use crate::background_jobs::Environment;
 use crate::models::{DependencyKind, Version};
 use crate::schema::versions;
-use crate::util::errors::{internal, std_error_no_send, CargoResult};
+use crate::util::errors::{std_error_no_send, CargoError, CargoResult};
 
 #[derive(Serialize, Deserialize, Debug)]
 pub struct Crate {
@@ -159,7 +159,9 @@ impl Job for AddCrate {
 }
 
 pub fn add_crate(conn: &PgConnection, krate: Crate) -> CargoResult<()> {
-    AddCrate { krate }.enqueue(conn).map_err(|e| internal(&e))
+    AddCrate { krate }
+        .enqueue(conn)
+        .map_err(|e| CargoError::from_std_error(e))
 }
 
 #[derive(Serialize, Deserialize)]
@@ -239,5 +241,5 @@ pub fn yank(conn: &PgConnection, krate: String, version: Version, yanked: bool)
         yanked,
     }
     .enqueue(conn)
-    .map_err(|e| internal(&e))
+    .map_err(|e| CargoError::from_std_error(e))
 }

src/middleware/current_user.rs

@@ -45,7 +45,9 @@ impl Middleware for CurrentUser {
             // Otherwise, look for an `Authorization` header on the request
             // and try to find a user in the database with a matching API token
             let user = if let Some(headers) = req.headers().find("Authorization") {
-                User::find_by_api_token(&conn, headers[0]).ok()
+                User::find_by_api_token(&conn, headers[0])
+                    .optional()
+                    .map_err(|e| Box::new(e) as Box<dyn Error + Send>)?
             } else {
                 None
             };

src/middleware/run_pending_background_jobs.rs

@@ -13,6 +13,10 @@ impl Middleware for RunPendingBackgroundJobs {
         req: &mut dyn Request,
         res: Result<Response, Box<dyn Error + Send>>,
     ) -> Result<Response, Box<dyn Error + Send>> {
+        if response_is_error(&res) {
+            return res;
+        }
+
         let app = req.app();
         let connection_pool = app.diesel_database.clone();
         let repo = Repository::open(&app.config.index_location).expect("Could not clone index");
@@ -28,3 +32,10 @@ impl Middleware for RunPendingBackgroundJobs {
         res
     }
 }
+
+fn response_is_error(res: &Result<Response, Box<dyn Error + Send>>) -> bool {
+    match res {
+        Ok(res) => res.status.0 >= 400,
+        Err(_) => true,
+    }
+}

src/models/user.rs

@@ -106,18 +106,26 @@ impl<'a> NewUser<'a> {
 
 impl User {
     /// Queries the database for a user with a certain `api_token` value.
-    pub fn find_by_api_token(conn: &PgConnection, token_: &str) -> CargoResult<User> {
+    pub fn find_by_api_token(conn: &PgConnection, token_: &str) -> QueryResult<User> {
         use crate::schema::api_tokens::dsl::{api_tokens, last_used_at, revoked, token, user_id};
-        use crate::schema::users::dsl::{id, users};
         use diesel::update;
+
         let tokens = api_tokens
             .filter(token.eq(token_))
             .filter(revoked.eq(false));
-        let user_id_ = update(tokens)
-            .set(last_used_at.eq(now.nullable()))
-            .returning(user_id)
-            .get_result::<i32>(conn)?;
-        Ok(users.filter(id.eq(user_id_)).get_result(conn)?)
+
+        // If the database is in read only mode, we can't update last_used_at.
+        // Try updating in a new transaction, if that fails, fall back to reading
+        let user_id_ = conn
+            .transaction(|| {
+                update(tokens)
+                    .set(last_used_at.eq(now.nullable()))
+                    .returning(user_id)
+                    .get_result::<i32>(conn)
+            })
+            .or_else(|_| tokens.select(user_id).first(conn))?;
+
+        users::table.find(user_id_).first(conn)
     }
 
     pub fn owning(krate: &Crate, conn: &PgConnection) -> CargoResult<Vec<Owner>> {

src/tests/all.rs

@@ -68,6 +68,7 @@ mod git;
 mod keyword;
 mod krate;
 mod owners;
+mod read_only_mode;
 mod record;
 mod schema_details;
 mod server;

src/tests/read_only_mode.rs

@@ -0,0 +1,56 @@
+use crate::builders::CrateBuilder;
+use crate::{RequestHelper, TestApp};
+use diesel::prelude::*;
+
+#[test]
+fn can_hit_read_only_endpoints_in_read_only_mode() {
+    let (app, anon) = TestApp::init().empty();
+    app.db(set_read_only).unwrap();
+    anon.get::<()>("/api/v1/crates").assert_status(200);
+}
+
+#[test]
+fn cannot_hit_endpoint_which_writes_db_in_read_only_mode() {
+    let (app, _, user, token) = TestApp::init().with_token();
+    app.db(|conn| {
+        CrateBuilder::new("foo_yank_read_only", user.as_model().id)
+            .version("1.0.0")
+            .expect_build(conn);
+        set_read_only(conn).unwrap();
+    });
+    token
+        .delete::<()>("/api/v1/crates/foo_yank_read_only/1.0.0/yank")
+        .assert_status(503);
+}
+
+#[test]
+#[ignore] // Will be implicitly fixed by #1387, no need to special case here
+fn can_download_crate_in_read_only_mode() {
+    let (app, anon, user) = TestApp::with_proxy().with_user();
+
+    app.db(|conn| {
+        CrateBuilder::new("foo_download_read_only", user.as_model().id)
+            .version("1.0.0")
+            .expect_build(conn);
+        set_read_only(conn).unwrap();
+    });
+
+    anon.get::<()>("/api/v1/crates/foo_download_read_only/1.0.0/download")
+        .assert_status(302);
+
+    // We're in read only mode so the download should not have been counted
+    app.db(|conn| {
+        use cargo_registry::schema::version_downloads::dsl::*;
+        use diesel::dsl::sum;
+
+        let dl_count = version_downloads
+            .select(sum(downloads))
+            .get_result::<Option<i64>>(conn);
+        assert_eq!(Ok(None), dl_count);
+    })
+}
+
+fn set_read_only(conn: &PgConnection) -> QueryResult<()> {
+    diesel::sql_query("SET TRANSACTION READ ONLY").execute(conn)?;
+    Ok(())
+}

src/util/errors.rs

@@ -49,6 +49,22 @@ impl dyn CargoError {
     pub fn is<T: Any>(&self) -> bool {
         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>> {
+        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,
+        }
+    }
 }
 
 impl CargoError for Box<dyn CargoError> {
@@ -155,13 +171,9 @@ impl<E: Error + Send + 'static> CargoError for E {
     }
 }
 
-impl<E: Any + Error + Send + 'static> From<E> for Box<dyn CargoError> {
+impl<E: Error + Send + 'static> From<E> for Box<dyn CargoError> {
     fn from(err: E) -> Box<dyn CargoError> {
-        if let Some(DieselError::NotFound) = Any::downcast_ref::<DieselError>(&err) {
-            Box::new(NotFound)
-        } else {
-            Box::new(err)
-        }
+        CargoError::try_convert(&err).unwrap_or_else(|| Box::new(err))
     }
 }
 // =============================================================================
@@ -340,3 +352,34 @@ pub fn std_error(e: Box<dyn CargoError>) -> Box<dyn Error + Send> {
 pub fn std_error_no_send(e: Box<dyn CargoError>) -> Box<dyn Error> {
     Box::new(CargoErrToStdErr(e))
 }
+
+#[derive(Debug, Clone, Copy)]
+pub struct ReadOnlyMode;
+
+impl CargoError for ReadOnlyMode {
+    fn description(&self) -> &str {
+        "tried to write in read only mode"
+    }
+
+    fn response(&self) -> Option<Response> {
+        let mut response = json_response(&Bad {
+            errors: vec![StringError {
+                detail: "Crates.io is currently in read-only mode for maintenance. \
+                         Please try again later."
+                    .to_string(),
+            }],
+        });
+        response.status = (503, "Service Unavailable");
+        Some(response)
+    }
+
+    fn human(&self) -> bool {
+        true
+    }
+}
+
+impl fmt::Display for ReadOnlyMode {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        "Tried to write in read only mode".fmt(f)
+    }
+}