Yanking only updates the database after the index is updated

Previously, we'd immediately update the database and then queue the
index update to be run later. Jobs aren't guaranteed to run in the
order they were queued (though in practice they likely will). This means
that if someone yanked and unyanked a crate in rapid succession, the
database may end up permanently out of sync with the index.

With this change, the final result may still be out of order, but the
database will be in sync with the index whatever the outcome is. Since
rapidly yanking and unyanking isn't something we expect users to do, and
even if they did jobs should be running in order under normal
circumstances, I don't think we need to do anything to ensure more
consistency than this.

Author: sgrif

src/background_jobs.rs

@@ -1,7 +1,10 @@
 use url::Url;
+use std::panic::AssertUnwindSafe;
 
 use crate::background::{Builder, Runner};
+use crate::db::{DieselPool, DieselPooledConn};
 use crate::git::{AddCrate, Yank};
+use crate::util::CargoResult;
 
 pub fn job_runner(config: Builder<Environment>) -> Runner<Environment> {
     config.register::<AddCrate>().register::<Yank>().build()
@@ -11,12 +14,30 @@ pub fn job_runner(config: Builder<Environment>) -> Runner<Environment> {
 pub struct Environment {
     pub index_location: Url,
     pub credentials: Option<(String, String)>,
+    // FIXME: https://github.com/sfackler/r2d2/pull/70
+    pub connection_pool: AssertUnwindSafe<DieselPool>,
 }
 
 impl Environment {
+    pub fn new(
+        index_location: Url,
+        credentials: Option<(String, String)>,
+        connection_pool: DieselPool,
+    ) -> Self {
+        Self {
+            index_location,
+            credentials,
+            connection_pool: AssertUnwindSafe(connection_pool),
+        }
+    }
+
     pub fn credentials(&self) -> Option<(&str, &str)> {
         self.credentials
             .as_ref()
             .map(|(u, p)| (u.as_str(), p.as_str()))
     }
+
+    pub fn connection(&self) -> CargoResult<DieselPooledConn> {
+        self.connection_pool.0.get().map_err(Into::into)
+    }
 }

src/bin/background-worker.rs

@@ -18,20 +18,21 @@ use std::time::Duration;
 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(1);
+    let db_pool = db::diesel_pool(&config.db_url, config.env, db_config);
+
     let username = env::var("GIT_HTTP_USER");
     let password = env::var("GIT_HTTP_PWD");
     let credentials = match (username, password) {
         (Ok(u), Ok(p)) => Some((u, p)),
         _ => None,
     };
-    let environment = Environment {
-        index_location: config.index_location,
+    let environment = Environment::new(
+        config.index_location,
         credentials,
-    };
-
-    // We're only using 1 thread, so we only need 1 connection
-    let db_config = r2d2::Pool::builder().max_size(1);
-    let db_pool = db::diesel_pool(&config.db_url, config.env, db_config);
+        db_pool.clone(),
+    );
 
     let builder = background::Runner::builder(db_pool, environment).thread_count(1);
     let runner = job_runner(builder);

src/controllers/version/yank.rs

@@ -3,10 +3,8 @@
 use crate::controllers::prelude::*;
 
 use crate::git;
-use crate::util::errors::CargoError;
 
 use crate::models::Rights;
-use crate::schema::*;
 
 use super::version_and_crate;
 
@@ -39,13 +37,7 @@ fn modify_yank(req: &mut dyn Request, yanked: bool) -> CargoResult<Response> {
     }
 
     if version.yanked != yanked {
-        conn.transaction::<_, Box<dyn CargoError>, _>(|| {
-            diesel::update(&version)
-                .set(versions::yanked.eq(yanked))
-                .execute(&*conn)?;
-            git::yank(&conn, krate.name, &version.num, yanked)?;
-            Ok(())
-        })?;
+        git::yank(&conn, krate.name, version, yanked)?;
     }
 
     #[derive(Serialize)]

src/git.rs

@@ -10,7 +10,8 @@ use url::Url;
 
 use crate::background::Job;
 use crate::background_jobs::Environment;
-use crate::models::DependencyKind;
+use crate::models::{DependencyKind, Version};
+use crate::schema::versions;
 use crate::util::{internal, CargoResult};
 
 #[derive(Serialize, Deserialize, Debug)]
@@ -155,7 +156,7 @@ pub fn add_crate(conn: &PgConnection, krate: Crate) -> CargoResult<()> {
 #[derive(Serialize, Deserialize)]
 pub struct Yank {
     krate: String,
-    version: String,
+    version: Version,
     yanked: bool,
 }
 
@@ -168,12 +169,13 @@ impl Job for Yank {
         let dst = repo.index_file(&self.krate);
 
         let prev = fs::read_to_string(&dst)?;
+        let version = self.version.num.to_string();
         let new = prev
             .lines()
             .map(|line| {
                 let mut git_crate = serde_json::from_str::<Crate>(line)
                     .map_err(|_| internal(&format_args!("couldn't decode: `{}`", line)))?;
-                if git_crate.name != self.krate || git_crate.vers != self.version {
+                if git_crate.name != self.krate || git_crate.vers != version {
                     return Ok(line.to_string());
                 }
                 git_crate.yanked = Some(self.yanked);
@@ -188,11 +190,17 @@ impl Job for Yank {
                 "{} crate `{}#{}`",
                 if self.yanked { "Yanking" } else { "Unyanking" },
                 self.krate,
-                self.version
+                self.version.num
             ),
             &repo.relative_index_file(&self.krate),
             env.credentials(),
-        )
+        )?;
+
+        diesel::update(&self.version)
+            .set(versions::yanked.eq(self.yanked))
+            .execute(&*env.connection()?)?;
+
+        Ok(())
     }
 }
 
@@ -203,12 +211,12 @@ impl Job for Yank {
 pub fn yank(
     conn: &PgConnection,
     krate: String,
-    version: &semver::Version,
+    version: Version,
     yanked: bool,
 ) -> CargoResult<()> {
     Yank {
         krate,
-        version: version.to_string(),
+        version,
         yanked,
     }
     .enqueue(conn)

src/middleware/run_pending_background_jobs.rs

@@ -13,10 +13,11 @@ impl Middleware for RunPendingBackgroundJobs {
     ) -> Result<Response, Box<dyn Error + Send>> {
         let app = req.app();
         let connection_pool = app.diesel_database.clone();
-        let environment = Environment {
-            index_location: app.config.index_location.clone(),
-            credentials: None,
-        };
+        let environment = Environment::new(
+            app.config.index_location.clone(),
+            None,
+            app.diesel_database.clone(),
+        );
 
         let config = Runner::builder(connection_pool, environment);
         let runner = job_runner(config);

src/models/version.rs

@@ -10,7 +10,7 @@ use crate::schema::*;
 use crate::views::{EncodableVersion, EncodableVersionLinks};
 
 // Queryable has a custom implementation below
-#[derive(Clone, Identifiable, Associations, Debug, Queryable)]
+#[derive(Clone, Identifiable, Associations, Debug, Queryable, Deserialize, Serialize)]
 #[belongs_to(Crate)]
 pub struct Version {
     pub id: i32,