worker: Implement sync_to_git/sparse_index() jobs

Author: Turbo87

src/background_jobs.rs

@@ -18,6 +18,8 @@ pub enum Job {
     IndexAddCrate(IndexAddCrateJob),
     IndexSquash,
     IndexSyncToHttp(IndexSyncToHttpJob),
+    SyncToGitIndex(SyncToIndexJob),
+    SyncToSparseIndex(SyncToIndexJob),
     IndexUpdateYanked(IndexUpdateYankedJob),
     NormalizeIndex(NormalizeIndexJob),
     RenderAndUploadReadme(RenderAndUploadReadmeJob),
@@ -47,8 +49,22 @@ impl Job {
     const INDEX_UPDATE_YANKED: &str = "sync_yanked";
     const NORMALIZE_INDEX: &str = "normalize_index";
     const RENDER_AND_UPLOAD_README: &str = "render_and_upload_readme";
+    const SYNC_TO_GIT_INDEX: &str = "sync_to_git_index";
+    const SYNC_TO_SPARSE_INDEX: &str = "sync_to_sparse_index";
     const UPDATE_DOWNLOADS: &str = "update_downloads";
 
+    pub fn sync_to_git_index<T: ToString>(krate: T) -> Job {
+        Job::SyncToGitIndex(SyncToIndexJob {
+            krate: krate.to_string(),
+        })
+    }
+
+    pub fn sync_to_sparse_index<T: ToString>(krate: T) -> Job {
+        Job::SyncToSparseIndex(SyncToIndexJob {
+            krate: krate.to_string(),
+        })
+    }
+
     fn as_type_str(&self) -> &'static str {
         match self {
             Job::DailyDbMaintenance => Self::DAILY_DB_MAINTENANCE,
@@ -59,6 +75,8 @@ impl Job {
             Job::IndexUpdateYanked(_) => Self::INDEX_UPDATE_YANKED,
             Job::NormalizeIndex(_) => Self::NORMALIZE_INDEX,
             Job::RenderAndUploadReadme(_) => Self::RENDER_AND_UPLOAD_README,
+            Job::SyncToGitIndex(_) => Self::SYNC_TO_GIT_INDEX,
+            Job::SyncToSparseIndex(_) => Self::SYNC_TO_SPARSE_INDEX,
             Job::UpdateDownloads => Self::UPDATE_DOWNLOADS,
         }
     }
@@ -73,6 +91,8 @@ impl Job {
             Job::IndexUpdateYanked(inner) => serde_json::to_value(inner),
             Job::NormalizeIndex(inner) => serde_json::to_value(inner),
             Job::RenderAndUploadReadme(inner) => serde_json::to_value(inner),
+            Job::SyncToGitIndex(inner) => serde_json::to_value(inner),
+            Job::SyncToSparseIndex(inner) => serde_json::to_value(inner),
             Job::UpdateDownloads => Ok(serde_json::Value::Null),
         }
     }
@@ -101,6 +121,8 @@ impl Job {
             Self::INDEX_UPDATE_YANKED => Job::IndexUpdateYanked(from_value(value)?),
             Self::NORMALIZE_INDEX => Job::NormalizeIndex(from_value(value)?),
             Self::RENDER_AND_UPLOAD_README => Job::RenderAndUploadReadme(from_value(value)?),
+            Self::SYNC_TO_GIT_INDEX => Job::SyncToGitIndex(from_value(value)?),
+            Self::SYNC_TO_SPARSE_INDEX => Job::SyncToSparseIndex(from_value(value)?),
             Self::UPDATE_DOWNLOADS => Job::UpdateDownloads,
             job_type => Err(PerformError::from(format!("Unknown job type {job_type}")))?,
         })
@@ -136,6 +158,8 @@ impl Job {
                 args.base_url.as_deref(),
                 args.pkg_path_in_vcs.as_deref(),
             ),
+            Job::SyncToGitIndex(args) => worker::sync_to_git_index(env, conn, &args.krate),
+            Job::SyncToSparseIndex(args) => worker::sync_to_sparse_index(env, conn, &args.krate),
             Job::UpdateDownloads => worker::perform_update_downloads(&mut *fresh_connection(pool)?),
         }
     }
@@ -172,6 +196,11 @@ pub struct IndexSyncToHttpJob {
     pub(super) crate_name: String,
 }
 
+#[derive(Serialize, Deserialize)]
+pub struct SyncToIndexJob {
+    pub(super) krate: String,
+}
+
 #[derive(Serialize, Deserialize)]
 pub struct IndexUpdateYankedJob {
     pub(super) krate: String,

src/worker/git.rs

@@ -1,14 +1,15 @@
 use crate::background_jobs::{
     Environment, IndexAddCrateJob, IndexSyncToHttpJob, IndexUpdateYankedJob, Job, NormalizeIndexJob,
 };
+use crate::models;
 use crate::schema;
 use crate::swirl::PerformError;
 use anyhow::Context;
 use cargo_registry_index::{Crate, Repository};
 use chrono::Utc;
 use diesel::prelude::*;
-use std::fs::{self, OpenOptions};
-use std::io::{BufRead, BufReader, ErrorKind};
+use std::fs::{self, File, OpenOptions};
+use std::io::{BufRead, BufReader, ErrorKind, Write};
 use std::process::Command;
 
 #[instrument(skip_all, fields(krate.name = ?krate.name, krate.vers = ?krate.vers))]
@@ -74,6 +75,98 @@ pub fn update_crate_index(crate_name: String) -> Job {
     Job::IndexSyncToHttp(IndexSyncToHttpJob { crate_name })
 }
 
+/// Regenerates or removes an index file for a single crate
+#[instrument(skip_all, fields(krate.name = ?krate))]
+pub fn sync_to_git_index(
+    env: &Environment,
+    conn: &mut PgConnection,
+    krate: &str,
+) -> Result<(), PerformError> {
+    info!("Syncing to git index");
+
+    let new = get_index_data(krate, conn).context("Failed to get index data")?;
+
+    let repo = env.lock_index()?;
+    let dst = repo.index_file(krate);
+
+    // Read the previous crate contents
+    let old = match fs::read_to_string(&dst) {
+        Ok(content) => Some(content),
+        Err(error) if error.kind() == ErrorKind::NotFound => None,
+        Err(error) => return Err(error.into()),
+    };
+
+    match (old, new) {
+        (None, Some(new)) => {
+            fs::create_dir_all(dst.parent().unwrap())?;
+            let mut file = File::create(&dst)?;
+            file.write_all(new.as_bytes())?;
+            repo.commit_and_push(&format!("Creating crate `{}`", krate), &dst)?;
+        }
+        (Some(old), Some(new)) if old != new => {
+            let mut file = File::create(&dst)?;
+            file.write_all(new.as_bytes())?;
+            repo.commit_and_push(&format!("Updating crate `{}`", krate), &dst)?;
+        }
+        (Some(_old), None) => {
+            fs::remove_file(&dst)?;
+            repo.commit_and_push(&format!("Deleting crate `{}`", krate), &dst)?;
+        }
+        _ => debug!("Skipping sync because index is up-to-date"),
+    }
+
+    Ok(())
+}
+
+/// Regenerates or removes an index file for a single crate
+#[instrument(skip_all, fields(krate.name = ?krate))]
+pub fn sync_to_sparse_index(
+    env: &Environment,
+    conn: &mut PgConnection,
+    krate: &str,
+) -> Result<(), PerformError> {
+    info!("Syncing to sparse index");
+
+    let content = get_index_data(krate, conn).context("Failed to get index data")?;
+
+    env.uploader
+        .sync_index(env.http_client(), krate, content)
+        .context("Failed to sync index data")?;
+
+    if let Some(cloudfront) = env.cloudfront() {
+        let path = Repository::relative_index_file_for_url(krate);
+
+        info!(%path, "Invalidating index file on CloudFront");
+        cloudfront
+            .invalidate(env.http_client(), &path)
+            .context("Failed to invalidate CloudFront")?;
+    }
+
+    Ok(())
+}
+
+#[instrument(skip_all, fields(krate.name = ?name))]
+pub fn get_index_data(name: &str, conn: &mut PgConnection) -> anyhow::Result<Option<String>> {
+    debug!("Looking up crate by name");
+    let Some(krate): Option<models::Crate> = models::Crate::by_exact_name(name).first(conn).optional()? else {
+        return Ok(None);
+    };
+
+    debug!("Gathering remaining index data");
+    let crates = krate
+        .index_metadata(conn)
+        .context("Failed to gather index metadata")?;
+
+    debug!("Serializing index data");
+    let mut bytes = Vec::new();
+    cargo_registry_index::write_crates(&crates, &mut bytes)
+        .context("Failed to serialize index metadata")?;
+
+    let str = String::from_utf8(bytes).context("Failed to decode index metadata as utf8")?;
+
+    Ok(Some(str))
+}
+
 /// 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

src/worker/mod.rs

@@ -20,7 +20,7 @@ pub(crate) use daily_db_maintenance::perform_daily_db_maintenance;
 pub(crate) use dump_db::perform_dump_db;
 pub(crate) use git::{
     perform_index_add_crate, perform_index_squash, perform_index_sync_to_http,
-    perform_index_update_yanked, perform_normalize_index,
+    perform_index_update_yanked, perform_normalize_index, sync_to_git_index, sync_to_sparse_index,
 };
 pub(crate) use readmes::perform_render_and_upload_readme;
 pub(crate) use update_downloads::perform_update_downloads;