Merge pull request #40 from spacestation13/cancelshit

adds job cancelling for commit spam, updates spacemandmm, updates

Author: jupyterkat

Cargo.lock

@@ -1,23 +1,22 @@
 [package]
 name = "diffbot_lib"
 version = "0.1.0"
-edition = "2021"
+edition = "2024"
 license = "MIT"
 
 [dependencies]
 serde = { version = "1.0.219", features = ["derive"] }
-octocrab = "0.44.0"
+octocrab = "0.44.1"
 eyre = "0.6.12"
 derive_builder = "0.20.2"
-chrono = "0.4.40"
-sha2 = "0.10.8"
+chrono = "0.4.41"
+sha2 = "0.10.9"
 hmac = "0.12.1"
 hex = "0.4.3"
 tracing = "0.1"
 tracing-subscriber = { version = "0.3", features = ["fmt", "time"] }
 tracing-panic = "0.1.2"
 tracing-loki = "0.2.6"
 flume = "0.11.1"
-async-fs = "2.1.2"
 
-actix-web = "4.10.2"
+actix-web = "4.11.0"

crates/diffbot_lib/Cargo.toml

@@ -11,12 +11,6 @@ pub type JobRunner = fn(Job) -> Result<CheckOutputs>;
 
 pub type JobSender<T> = Sender<T>;
 
-#[derive(Serialize, Deserialize, Debug, Clone)]
-pub enum JobType {
-    GithubJob(Box<Job>),
-    CleanupJob,
-}
-
 #[derive(Serialize, Deserialize, Debug, Clone)]
 pub struct Job {
     pub repo: github_types::Repository,

crates/diffbot_lib/src/job/types.rs

@@ -2,5 +2,4 @@ pub mod github;
 pub mod job;
 pub mod logger;
 pub mod verify;
-pub use async_fs;
 pub use tracing;

crates/diffbot_lib/src/lib.rs

@@ -1,32 +1,33 @@
 [package]
 name = "icondiffbot2"
 version = "0.1.0"
-edition = "2021"
+edition = "2024"
 license = "MIT"
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
 [dependencies]
 serde = { version = "1.0.219", features = ["derive"] }
-serde_json = "1.0.140"
-octocrab = "0.44.0"
+serde_json = "1.0.143"
+octocrab = "0.44.1"
 dmm-tools = { git = "https://github.com/jupyterkat/SpacemanDMM/" }
 jsonwebtoken = "9.3.1"
-reqwest = "0.12.15"
+reqwest = "0.12.23"
 diffbot_lib = { path = "../diffbot_lib" }
 eyre = "0.6.12"
 simple-eyre = "0.3.1"
-rayon = "1.10.0"
-toml = "0.8.20"
-ahash = "0.8.11"
-hashbrown = { version = "0.15.2", features = ["rayon"] }
+rayon = "1.11.0"
+toml = "0.9.5"
+ahash = "0.8.12"
+hashbrown = { version = "0.15.5", features = ["rayon"] }
 tracing-loki = "0.2.6"
 flume = "0.11.1"
-mysql_async = "0.35.1"
+mysql_async = "0.36.1"
 time = "0.3.41"
 secrecy = "0.10.3"
-percent-encoding = "2.3.1"
+percent-encoding = "2.3.2"
+dashmap = "6.1.0"
 
-actix-web = "4.10.2"
+actix-web = "4.11.0"
 actix-files = "0.6.6"
 
 [target.'cfg(not(target_env = "msvc"))'.dependencies]

crates/icondiffbot2/Cargo.toml

@@ -1,3 +1,4 @@
+use actix_web::web::Data;
 use diffbot_lib::{
     github::{
         github_api::CheckRun,
@@ -12,15 +13,14 @@ use octocrab::models::InstallationId;
 
 use mysql_async::{params, prelude::Queryable};
 
-use crate::DataJobSender;
+use crate::JobScheduler;
 
 async fn handle_pull_request(
     payload: PullRequestEventPayload,
-    job_sender: DataJobSender,
-    pool: actix_web::web::Data<Option<mysql_async::Pool>>,
+    scheduler: &JobScheduler,
+    sender: &flume::Sender<(String, u64)>,
+    pool: Option<&mysql_async::Pool>,
 ) -> Result<()> {
-    let pool = pool.get_ref();
-
     match payload.action.as_str() {
         "opened" | "synchronize" => {
             let check_run = CheckRun::create(
@@ -37,9 +37,9 @@ async fn handle_pull_request(
                 payload.pull_request.number,
             );
 
-            let num_icons = handle_pull(payload, job_sender, check_run).await?;
+            let num_icons = handle_pull(payload, scheduler, sender, check_run).await?;
 
-            if let Some(ref pool) = pool {
+            if let Some(pool) = pool {
                 let mut conn = match pool.get_conn().await {
                     Ok(conn) => conn,
                     Err(e) => {
@@ -81,7 +81,9 @@ async fn handle_pull_request(
             Ok(())
         }
         "closed" => {
-            if let Some(ref pool) = pool {
+            scheduler.remove(&(payload.repository.full_name(), payload.pull_request.number));
+
+            if let Some(pool) = pool {
                 let mut conn = match pool.get_conn().await {
                     Ok(conn) => conn,
                     Err(e) => {
@@ -106,7 +108,7 @@ async fn handle_pull_request(
                 {
                     tracing::error!("{:?}", e);
                 };
-            };
+            }
             Ok(())
         }
         _ => Ok(()),
@@ -115,7 +117,8 @@ async fn handle_pull_request(
 
 async fn handle_pull(
     payload: PullRequestEventPayload,
-    job_sender: DataJobSender,
+    scheduler: &JobScheduler,
+    sender: &flume::Sender<(String, u64)>,
     check_run: CheckRun,
 ) -> Result<usize> {
     if payload
@@ -188,6 +191,9 @@ async fn handle_pull(
 
     check_run.mark_queued().await?;
 
+    let scheduler_entry = (payload.repository.full_name(), payload.pull_request.number);
+    let scheduler_entry_clone = (payload.repository.full_name(), payload.pull_request.number);
+
     let pull = payload.pull_request;
     let installation = payload.installation;
 
@@ -201,7 +207,20 @@ async fn handle_pull(
         installation: InstallationId(installation.id),
     };
 
-    job_sender.send_async(job).await?;
+    if let Some(old_job) = match scheduler.entry(scheduler_entry) {
+        dashmap::Entry::Occupied(mut entry) => Some(entry.insert(job)),
+        dashmap::Entry::Vacant(entry) => {
+            entry.insert(job);
+            None
+        }
+    } {
+        _ = old_job
+            .check_run
+            .mark_failed("Check cancelled, a later commit has triggered the check")
+            .await;
+    }
+
+    sender.send_async(scheduler_entry_clone).await?;
 
     Ok(num_icons_diffed)
 }
@@ -210,9 +229,15 @@ async fn handle_pull(
 pub async fn process_github_payload_actix(
     event: diffbot_lib::github::github_api::GithubEvent,
     payload: String,
-    job_sender: DataJobSender,
-    pool: actix_web::web::Data<Option<mysql_async::Pool>>,
+    pool: Data<Option<mysql_async::Pool>>,
+    scheduler: Data<JobScheduler>,
+    sender: Data<flume::Sender<(String, u64)>>,
 ) -> actix_web::Result<&'static str> {
+    let (pool, scheduler, sender) = (
+        pool.get_ref().as_ref(),
+        scheduler.get_ref(),
+        sender.get_ref(),
+    );
     // TODO: Handle reruns
     if event.0 != "pull_request" {
         return Ok("Not a pull request event");
@@ -231,7 +256,7 @@ pub async fn process_github_payload_actix(
 
     let payload: PullRequestEventPayload = serde_json::from_str(&payload)?;
 
-    handle_pull_request(payload, job_sender, pool)
+    handle_pull_request(payload, scheduler, sender, pool)
         .await
         .map_err(actix_web::error::ErrorBadRequest)?;
 

crates/icondiffbot2/src/github_processor.rs

@@ -1,11 +1,11 @@
 use crate::{
-    sha::{sha_to_iconfile, status_to_sha, IconFileWithName},
-    table_builder::OutputTableBuilder,
     CONFIG,
+    sha::{IconFileWithName, sha_to_iconfile, status_to_sha},
+    table_builder::OutputTableBuilder,
 };
 use diffbot_lib::{github::github_types::CheckOutputs, job::types::Job, tracing};
-use dmm_tools::dmi::render::{IconRenderer, RenderType};
 use dmm_tools::dmi::State;
+use dmm_tools::dmi::render::{IconRenderer, RenderType};
 use eyre::{Context, Result};
 use hashbrown::HashSet;
 use rayon::prelude::*;
@@ -344,14 +344,13 @@ fn full_render(job: &Job, target: &IconFileWithName) -> Result<Vec<((usize, Stri
     let vec: Vec<((usize, String), String)> = icon
         .metadata
         .states
-        .par_values()
-        .map(|vec| {
+        .values()
+        .flat_map(|vec| {
             vec.iter()
                 .enumerate()
                 .map(|(duplication_index, (_, state))| (duplication_index, state))
                 .collect::<Vec<_>>()
         })
-        .flatten()
         .map(|(idx, state)| {
             render_state(&prefix, target, (idx, state), &renderer)
                 .with_context(|| format!("Failed to render state {}", state.name))

crates/icondiffbot2/src/job_processor.rs

@@ -5,14 +5,11 @@ mod runner;
 mod sha;
 mod table_builder;
 
-use diffbot_lib::{
-    async_fs,
-    job::types::{Job, JobSender},
-};
+use diffbot_lib::job::types::Job;
 use mysql_async::prelude::Queryable;
 use octocrab::OctocrabBuilder;
 use serde::Deserialize;
-use std::sync::OnceLock;
+use std::sync::{Arc, OnceLock};
 use std::{
     fs::File,
     io::Read,
@@ -23,7 +20,7 @@ use std::{
 #[global_allocator]
 static GLOBAL: tikv_jemallocator::Jemalloc = tikv_jemallocator::Jemalloc;
 
-pub type DataJobSender = actix_web::web::Data<JobSender<Job>>;
+pub type JobScheduler = Arc<dashmap::DashMap<(String, u64), Job, ahash::RandomState>>;
 
 #[actix_web::get("/")]
 async fn index() -> &'static str {
@@ -132,9 +129,8 @@ async fn main() -> eyre::Result<()> {
     simple_eyre::install().expect("Eyre handler installation failed!");
     // init_global_subscriber();
 
-    let config_path = Path::new(".").join("config.toml");
-    let config =
-        init_config(&config_path).unwrap_or_else(|_| panic!("Failed to read {config_path:?}"));
+    let config_path = std::path::Path::new(".").join("config").join("config.toml");
+    let config = init_config(&config_path).unwrap();
 
     let (layer, tasks) = if let Some(ref loki_config) = config.grafana_loki {
         let (layer, tasks) = tracing_loki::builder()
@@ -164,9 +160,7 @@ async fn main() -> eyre::Result<()> {
     );
     let reqwest_client = reqwest::Client::new();
 
-    async_fs::create_dir_all("./images").await.unwrap();
-
-    let (job_sender, job_receiver) = flume::unbounded();
+    std::fs::create_dir_all("./images").unwrap();
 
     let pool = config
         .db_url
@@ -191,16 +185,18 @@ async fn main() -> eyre::Result<()> {
         .await?;
     }
 
+    let (sender, receiver) = flume::unbounded();
+
+    let scheduler: JobScheduler = Default::default();
+
     actix_web::rt::spawn(runner::handle_jobs(
         "IconDiffBot2",
-        job_receiver,
+        scheduler.clone(),
+        receiver,
         reqwest_client,
     ));
 
-    let job_sender: DataJobSender = actix_web::web::Data::new(job_sender);
-
     actix_web::HttpServer::new(move || {
-        let pool = actix_web::web::Data::new(pool.clone());
         use actix_web::web::{FormConfig, PayloadConfig};
         //absolutely rancid
         let (form_config, string_config) = config.web.limits.as_ref().map_or(
@@ -215,8 +211,9 @@ async fn main() -> eyre::Result<()> {
         actix_web::App::new()
             .app_data(form_config)
             .app_data(string_config)
-            .app_data(job_sender.clone())
-            .app_data(pool)
+            .app_data(actix_web::web::Data::new(pool.clone()))
+            .app_data(actix_web::web::Data::new(scheduler.clone()))
+            .app_data(actix_web::web::Data::new(sender.clone()))
             .service(index)
             .service(github_processor::process_github_payload_actix)
             .service(actix_files::Files::new("/images", "./images"))

crates/icondiffbot2/src/main.rs

@@ -1,22 +1,27 @@
 use std::time::Duration;
 
+use crate::JobScheduler;
+
 use super::job_processor::do_job;
 use diffbot_lib::job::types::Job;
 
 use diffbot_lib::tracing;
 
 pub async fn handle_jobs<S: AsRef<str>>(
     name: S,
-    job_receiver: flume::Receiver<Job>,
+    scheduler: JobScheduler,
+    receiver: flume::Receiver<(String, u64)>,
     client: reqwest::Client,
 ) {
     loop {
-        match job_receiver.recv_async().await {
-            Ok(job) => {
-                tracing::info!("Job received from queue");
-                job_handler(name.as_ref(), job, client.clone()).await;
+        match receiver.recv_async().await {
+            Ok(entry) => {
+                if let Some((_, job)) = scheduler.remove(&entry) {
+                    tracing::info!("Job received from queue");
+                    job_handler(name.as_ref(), job, client.clone()).await;
+                }
             }
-            Err(err) => tracing::error!("{err}"),
+            Err(e) => tracing::error!("{e}"),
         }
     }
 }

crates/icondiffbot2/src/runner.rs

@@ -2,10 +2,7 @@ use actix_web::rt::Runtime;
 use diffbot_lib::{github::github_types::ChangeType, job::types::Job};
 use dmm_tools::dmi::IconFile;
 use eyre::{Context, Result};
-use std::{
-    collections::hash_map::DefaultHasher,
-    hash::{Hash, Hasher},
-};
+use std::hash::{Hash, Hasher};
 
 use crate::downloading::download_url;
 
@@ -53,7 +50,7 @@ fn get_if_exists(
             .wrap_err_with(|| format!("Failed to download file {filename:?}"))
     })?;
 
-    let mut hasher = DefaultHasher::new();
+    let mut hasher = ahash::AHasher::default();
     raw.hash(&mut hasher);
     let hash = hasher.finish();