Auto merge of #4108 - Turbo87:worker-module, r=JohnTitor

Extract `worker` module

This PR renames the `tasks` module to `worker`, to make it clearer that these things are running on the background worker. It also moves the other swirl tasks outside of the `tasks` module into the new `worker` module.

Author: bors

src/bin/enqueue-job.rs

@@ -1,7 +1,7 @@
 #![warn(clippy::all, rust_2018_idioms)]
 
 use anyhow::{anyhow, Result};
-use cargo_registry::{db, env, git, tasks};
+use cargo_registry::{db, env, worker};
 use diesel::prelude::*;
 use swirl::schema::background_jobs::dsl::*;
 use swirl::Job;
@@ -25,18 +25,18 @@ fn main() -> Result<()> {
                 println!("Did not enqueue update_downloads, existing job already in progress");
                 Ok(())
             } else {
-                Ok(tasks::update_downloads().enqueue(&conn)?)
+                Ok(worker::update_downloads().enqueue(&conn)?)
             }
         }
         "dump_db" => {
             let database_url = args.next().unwrap_or_else(|| env("READ_ONLY_REPLICA_URL"));
             let target_name = args
                 .next()
                 .unwrap_or_else(|| String::from("db-dump.tar.gz"));
-            Ok(tasks::dump_db(database_url, target_name).enqueue(&conn)?)
+            Ok(worker::dump_db(database_url, target_name).enqueue(&conn)?)
         }
-        "daily_db_maintenance" => Ok(tasks::daily_db_maintenance().enqueue(&conn)?),
-        "squash_index" => Ok(git::squash_index().enqueue(&conn)?),
+        "daily_db_maintenance" => Ok(worker::daily_db_maintenance().enqueue(&conn)?),
+        "squash_index" => Ok(worker::squash_index().enqueue(&conn)?),
         other => Err(anyhow!("Unrecognized job type `{}`", other)),
     }
 }

src/controllers/krate/publish.rs

@@ -13,8 +13,8 @@ use crate::models::{
     insert_version_owner_action, Badge, Category, Crate, DependencyKind, Keyword, NewCrate,
     NewVersion, Rights, VersionAction,
 };
+use crate::worker;
 
-use crate::render;
 use crate::schema::*;
 use crate::util::errors::{cargo_err, AppResult};
 use crate::util::{read_fill, read_le_u32, LimitErrorReader, Maximums};
@@ -197,7 +197,7 @@ pub fn publish(req: &mut dyn RequestExt) -> EndpointResult {
         verify_tarball(&pkg_name, &tarball, maximums.max_unpack_size)?;
 
         if let Some(readme) = new_crate.readme {
-            render::render_and_upload_readme(
+            worker::render_and_upload_readme(
                 version.id,
                 readme,
                 new_crate
@@ -223,7 +223,7 @@ pub fn publish(req: &mut dyn RequestExt) -> EndpointResult {
             yanked: Some(false),
             links,
         };
-        git::add_crate(git_crate).enqueue(&conn)?;
+        worker::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/version/yank.rs

@@ -4,9 +4,9 @@ use swirl::Job;
 
 use super::{extract_crate_name_and_semver, version_and_crate};
 use crate::controllers::cargo_prelude::*;
-use crate::git;
 use crate::models::Rights;
 use crate::models::{insert_version_owner_action, VersionAction};
+use crate::worker;
 
 /// Handles the `DELETE /crates/:crate_id/:version/yank` route.
 /// This does not delete a crate version, it makes the crate
@@ -50,7 +50,7 @@ fn modify_yank(req: &mut dyn RequestExt, yanked: bool) -> EndpointResult {
 
     insert_version_owner_action(&conn, version.id, user.id, api_token_id, action)?;
 
-    git::yank(krate.name, version, yanked).enqueue(&conn)?;
+    worker::yank(krate.name, version, yanked).enqueue(&conn)?;
 
     ok_true()
 }

src/git.rs

@@ -1,16 +1,11 @@
 use std::collections::HashMap;
-use std::fs::{self, OpenOptions};
-use std::io::prelude::*;
 use std::path::{Path, PathBuf};
 
-use chrono::Utc;
 use swirl::PerformError;
 use tempfile::TempDir;
 use url::Url;
 
-use crate::background_jobs::Environment;
-use crate::models::{DependencyKind, Version};
-use crate::schema::versions;
+use crate::models::DependencyKind;
 
 static DEFAULT_GIT_SSH_USERNAME: &str = "git";
 
@@ -142,9 +137,9 @@ impl RepositoryConfig {
 }
 
 pub struct Repository {
-    checkout_path: TempDir,
+    pub checkout_path: TempDir,
     repository: git2::Repository,
-    credentials: Credentials,
+    pub credentials: Credentials,
 }
 
 impl Repository {
@@ -179,13 +174,13 @@ impl Repository {
         })
     }
 
-    fn index_file(&self, name: &str) -> PathBuf {
+    pub fn index_file(&self, name: &str) -> PathBuf {
         self.checkout_path
             .path()
             .join(self.relative_index_file(name))
     }
 
-    fn relative_index_file(&self, name: &str) -> PathBuf {
+    pub fn relative_index_file(&self, name: &str) -> PathBuf {
         let name = name.to_lowercase();
         match name.len() {
             1 => Path::new("1").join(&name),
@@ -195,7 +190,7 @@ impl Repository {
         }
     }
 
-    fn head_oid(&self) -> Result<git2::Oid, PerformError> {
+    pub fn head_oid(&self) -> Result<git2::Oid, PerformError> {
         Ok(self.repository.head()?.target().unwrap())
     }
 
@@ -246,7 +241,7 @@ impl Repository {
         ref_status
     }
 
-    fn commit_and_push(&self, message: &str, modified_file: &Path) -> Result<(), PerformError> {
+    pub fn commit_and_push(&self, message: &str, modified_file: &Path) -> Result<(), PerformError> {
         println!("Committing and pushing \"{}\"", message);
 
         self.perform_commit_and_push(message, modified_file)
@@ -291,7 +286,7 @@ impl Repository {
     }
 
     /// Reset `HEAD` to a single commit with all the index contents, but no parent
-    fn squash_to_single_commit(&self, msg: &str) -> Result<(), PerformError> {
+    pub fn squash_to_single_commit(&self, msg: &str) -> Result<(), PerformError> {
         let tree = self.repository.find_commit(self.head_oid()?)?.tree()?;
         let sig = self.repository.signature()?;
 
@@ -308,154 +303,3 @@ impl Repository {
         Ok(())
     }
 }
-
-#[swirl::background_job]
-pub fn add_crate(env: &Environment, krate: Crate) -> Result<(), PerformError> {
-    use std::io::prelude::*;
-
-    let repo = env.lock_index()?;
-    let dst = repo.index_file(&krate.name);
-
-    // Add the crate to its relevant file
-    fs::create_dir_all(dst.parent().unwrap())?;
-    let mut file = OpenOptions::new().append(true).create(true).open(&dst)?;
-    serde_json::to_writer(&mut file, &krate)?;
-    file.write_all(b"\n")?;
-
-    let message: String = format!("Updating crate `{}#{}`", krate.name, krate.vers);
-
-    repo.commit_and_push(&message, &repo.relative_index_file(&krate.name))
-}
-
-/// Yanks or unyanks a crate version. This requires finding the index
-/// file, deserlialise the crate from JSON, change the yank boolean to
-/// `true` or `false`, write all the lines back out, and commit and
-/// push the changes.
-#[swirl::background_job]
-pub fn yank(
-    conn: &PgConnection,
-    env: &Environment,
-    krate: String,
-    version: Version,
-    yanked: bool,
-) -> Result<(), PerformError> {
-    use diesel::prelude::*;
-
-    let repo = env.lock_index()?;
-    let dst = repo.index_file(&krate);
-
-    conn.transaction(|| {
-        let yanked_in_db: bool = versions::table
-            .find(version.id)
-            .select(versions::yanked)
-            .for_update()
-            .first(&*conn)?;
-
-        if yanked_in_db == yanked {
-            // The crate is alread in the state requested, nothing to do
-            return Ok(());
-        }
-
-        let prev = fs::read_to_string(&dst)?;
-        let new = prev
-            .lines()
-            .map(|line| {
-                let mut git_crate = serde_json::from_str::<Crate>(line)
-                    .map_err(|_| format!("couldn't decode: `{}`", line))?;
-                if git_crate.name != krate || git_crate.vers != version.num {
-                    return Ok(line.to_string());
-                }
-                git_crate.yanked = Some(yanked);
-                Ok(serde_json::to_string(&git_crate)?)
-            })
-            .collect::<Result<Vec<_>, PerformError>>();
-        let new = new?.join("\n") + "\n";
-        fs::write(&dst, new.as_bytes())?;
-
-        let message: String = format!(
-            "{} crate `{}#{}`",
-            if yanked { "Yanking" } else { "Unyanking" },
-            krate,
-            version.num
-        );
-
-        repo.commit_and_push(&message, &repo.relative_index_file(&krate))?;
-
-        diesel::update(&version)
-            .set(versions::yanked.eq(yanked))
-            .execute(&*conn)?;
-
-        Ok(())
-    })
-}
-
-/// Collapse the index into a single commit, archiving the current history in a snapshot branch.
-#[swirl::background_job]
-pub fn squash_index(env: &Environment) -> Result<(), PerformError> {
-    let repo = env.lock_index()?;
-    println!("Squashing the index into a single commit.");
-
-    let now = Utc::now().format("%Y-%m-%d");
-    let original_head = repo.head_oid()?.to_string();
-    let msg = format!("Collapse index into one commit\n\n\
-        Previous HEAD was {}, now on the `snapshot-{}` branch\n\n\
-        More information about this change can be found [online] and on [this issue].\n\n\
-        [online]: https://internals.rust-lang.org/t/cargos-crate-index-upcoming-squash-into-one-commit/8440\n\
-        [this issue]: https://github.com/rust-lang/crates-io-cargo-teams/issues/47", original_head, now);
-
-    repo.squash_to_single_commit(&msg)?;
-
-    // Shell out to git because libgit2 does not currently support push leases
-
-    let key = match &repo.credentials {
-        Credentials::Ssh { key } => key,
-        Credentials::Http { .. } => {
-            return Err(String::from("squash_index: Password auth not supported").into())
-        }
-        _ => return Err(String::from("squash_index: Could not determine credentials").into()),
-    };
-
-    // When running on production, ensure the file is created in tmpfs and not persisted to disk
-    #[cfg(target_os = "linux")]
-    let mut temp_key_file = tempfile::Builder::new().tempfile_in("/dev/shm")?;
-
-    // For other platforms, default to std::env::tempdir()
-    #[cfg(not(target_os = "linux"))]
-    let mut temp_key_file = tempfile::Builder::new().tempfile()?;
-
-    temp_key_file.write_all(key.as_bytes())?;
-
-    let checkout_path = repo.checkout_path.path();
-    let output = std::process::Command::new("git")
-        .current_dir(checkout_path)
-        .env(
-            "GIT_SSH_COMMAND",
-            format!(
-                "ssh -o StrictHostKeyChecking=accept-new -i {}",
-                temp_key_file.path().display()
-            ),
-        )
-        .args(&[
-            "push",
-            // Both updates should succeed or fail together
-            "--atomic",
-            "origin",
-            // Overwrite master, but only if it server matches the expected value
-            &format!("--force-with-lease=refs/heads/master:{}", original_head),
-            // The new squashed commit is pushed to master
-            "HEAD:refs/heads/master",
-            // The previous value of HEAD is pushed to a snapshot branch
-            &format!("{}:refs/heads/snapshot-{}", original_head, now),
-        ])
-        .output()?;
-
-    if !output.status.success() {
-        let stderr = String::from_utf8_lossy(&output.stderr);
-        let message = format!("Running git command failed with: {}", stderr);
-        return Err(message.into());
-    }
-
-    println!("The index has been successfully squashed.");
-
-    Ok(())
-}

src/lib.rs

@@ -45,13 +45,12 @@ pub mod github;
 pub mod metrics;
 pub mod middleware;
 mod publish_rate_limit;
-pub mod render;
 pub mod schema;
 pub mod sql;
-pub mod tasks;
 mod test_util;
 pub mod uploaders;
 pub mod util;
+pub mod worker;
 
 pub mod controllers;
 pub mod models;

src/tasks.rs

@@ -1,5 +1,5 @@
 use crate::util::FreshSchema;
-use cargo_registry::tasks::dump_db;
+use cargo_registry::worker::dump_db;
 
 #[test]
 fn dump_db_and_reimport_dump() {

src/tests/dump_db.rs

@@ -1,6 +1,6 @@
 use std::{fs::File, path::Path};
 
-use crate::tasks::dump_db::configuration::{ColumnVisibility, TableConfig, VisibilityConfig};
+use crate::worker::dump_db::configuration::{ColumnVisibility, TableConfig, VisibilityConfig};
 use swirl::PerformError;
 
 pub fn gen_scripts(export_script: &Path, import_script: &Path) -> Result<(), PerformError> {