From 01ba579646512a9d599dc79a5380eb2681945e2e Mon Sep 17 00:00:00 2001 From: Jamesbarford Date: Mon, 7 Jul 2025 10:38:26 +0100 Subject: [PATCH 1/3] Chore; comment out marking in progress, just log out for now --- site/src/job_queue.rs | 9 ++++++--- 1 file changed, 6 insertions(+), 3 deletions(-) diff --git a/site/src/job_queue.rs b/site/src/job_queue.rs index b359e80ba..b15d08c46 100644 --- a/site/src/job_queue.rs +++ b/site/src/job_queue.rs @@ -265,9 +265,12 @@ async fn enqueue_next_job(conn: &mut dyn database::pool::Connection) -> anyhow:: if let Some(request) = queue.into_iter().next() { if request.status != BenchmarkRequestStatus::InProgress { - // TODO: actually enqueue the jobs - conn.update_benchmark_request_status(&request, BenchmarkRequestStatus::InProgress) - .await?; + log::info!("{:?} would have been marked as InProgress", request); + // TODO: + // - Uncomment this code + // - Actually enqueue the jobs + // conn.update_benchmark_request_status(&request, BenchmarkRequestStatus::InProgress) + // .await?; } } From 74586b5bb4c4bde5cb0da6bd53ddf6bce94ecce2 Mon Sep 17 00:00:00 2001 From: Jamesbarford Date: Mon, 7 Jul 2025 10:50:45 +0100 Subject: [PATCH 2/3] Add a flag for inserting into the database; `true` for tests, `false` for actual use --- site/src/job_queue.rs | 39 ++++++++++++++++++++++++--------------- 1 file changed, 24 insertions(+), 15 deletions(-) diff --git a/site/src/job_queue.rs b/site/src/job_queue.rs index b15d08c46..7c2c9d0ef 100644 --- a/site/src/job_queue.rs +++ b/site/src/job_queue.rs @@ -252,7 +252,10 @@ pub async fn build_queue( } /// Enqueue the job into the job_queue -async fn enqueue_next_job(conn: &mut dyn database::pool::Connection) -> anyhow::Result<()> { +async fn enqueue_next_job( + conn: &mut dyn database::pool::Connection, + should_add_to_db: bool, +) -> anyhow::Result<()> { // We draw back all completed requests let completed: HashSet = conn .get_benchmark_requests_by_status(&[BenchmarkRequestStatus::Completed]) @@ -265,12 +268,15 @@ async fn enqueue_next_job(conn: &mut dyn database::pool::Connection) -> anyhow:: if let Some(request) = queue.into_iter().next() { if request.status != BenchmarkRequestStatus::InProgress { - log::info!("{:?} would have been marked as InProgress", request); // TODO: - // - Uncomment this code + // - Remove this if condition // - Actually enqueue the jobs - // conn.update_benchmark_request_status(&request, BenchmarkRequestStatus::InProgress) - // .await?; + if !should_add_to_db { + log::info!("{:?} would have been marked as InProgress", request); + } else { + conn.update_benchmark_request_status(&request, BenchmarkRequestStatus::InProgress) + .await?; + } } } @@ -278,13 +284,16 @@ async fn enqueue_next_job(conn: &mut dyn database::pool::Connection) -> anyhow:: } /// For queueing jobs, add the jobs you want to queue to this function -async fn cron_enqueue_jobs(site_ctxt: &Arc) -> anyhow::Result<()> { +async fn cron_enqueue_jobs( + site_ctxt: &Arc, + should_add_to_db: bool, +) -> anyhow::Result<()> { let mut conn = site_ctxt.conn().await; // Put the master commits into the `benchmark_requests` queue create_benchmark_request_master_commits(site_ctxt, &*conn).await?; // Put the releases into the `benchmark_requests` queue create_benchmark_request_releases(&*conn).await?; - enqueue_next_job(&mut *conn).await?; + enqueue_next_job(&mut *conn, should_add_to_db).await?; Ok(()) } @@ -300,7 +309,7 @@ pub async fn cron_main(site_ctxt: Arc>>>, seconds: u let guard = ctxt.read(); guard.as_ref().cloned() } { - match cron_enqueue_jobs(&ctxt_clone).await { + match cron_enqueue_jobs(&ctxt_clone, false).await { Ok(_) => log::info!("Cron job executed at: {:?}", std::time::SystemTime::now()), Err(e) => log::error!("Cron job failed to execute {}", e), } @@ -422,7 +431,7 @@ mod tests { run_postgres_test(|ctx| async { let mut db = ctx.db_client().connection().await; - enqueue_next_job(&mut *db).await?; + enqueue_next_job(&mut *db, true).await?; let in_progress = get_in_progress(&*db).await; @@ -443,7 +452,7 @@ mod tests { db_insert_requests(&*db, &[parent, child]).await; - enqueue_next_job(&mut *db).await?; + enqueue_next_job(&mut *db, true).await?; let in_progress = get_in_progress(&*db).await; @@ -462,7 +471,7 @@ mod tests { db_insert_requests(&*db, &[release]).await; - enqueue_next_job(&mut *db).await?; + enqueue_next_job(&mut *db, true).await?; let in_progress = get_in_progress(&*db).await; @@ -486,7 +495,7 @@ mod tests { let m2 = create_master("new", "y", 4, "days1"); db_insert_requests(&*db, &[c1, c2, m1, m2]).await; - enqueue_next_job(&mut *db).await?; + enqueue_next_job(&mut *db, true).await?; let in_progress = get_in_progress(&*db).await; @@ -506,7 +515,7 @@ mod tests { let orphan = create_master("orphan", "gone", 42, "days1"); db_insert_requests(&*db, &[orphan]).await; - enqueue_next_job(&mut *db).await?; + enqueue_next_job(&mut *db, true).await?; let in_progress = get_in_progress(&*db).await; assert_eq!(in_progress.unwrap().tag(), "orphan"); @@ -559,7 +568,7 @@ mod tests { ]; db_insert_requests(&*db, &requests).await; - enqueue_next_job(&mut *db).await?; + enqueue_next_job(&mut *db, true).await?; // The oldest release ("v0.8.0") outranks everything else let in_progress = get_in_progress(&*db).await; @@ -606,7 +615,7 @@ mod tests { ]; db_insert_requests(&*db, &requests).await; - enqueue_next_job(&mut *db).await?; + enqueue_next_job(&mut *db, true).await?; // The oldest release ("v0.8.0") outranks everything else let in_progress = get_in_progress(&*db).await; From 141e018ad881f4487229f51d5963f00f8abe147f Mon Sep 17 00:00:00 2001 From: Jamesbarford Date: Mon, 7 Jul 2025 10:56:45 +0100 Subject: [PATCH 3/3] remove flag and all tests --- site/src/job_queue.rs | 234 ++---------------------------------------- 1 file changed, 7 insertions(+), 227 deletions(-) diff --git a/site/src/job_queue.rs b/site/src/job_queue.rs index 7c2c9d0ef..ebcd4b37e 100644 --- a/site/src/job_queue.rs +++ b/site/src/job_queue.rs @@ -252,10 +252,7 @@ pub async fn build_queue( } /// Enqueue the job into the job_queue -async fn enqueue_next_job( - conn: &mut dyn database::pool::Connection, - should_add_to_db: bool, -) -> anyhow::Result<()> { +async fn enqueue_next_job(conn: &mut dyn database::pool::Connection) -> anyhow::Result<()> { // We draw back all completed requests let completed: HashSet = conn .get_benchmark_requests_by_status(&[BenchmarkRequestStatus::Completed]) @@ -269,14 +266,10 @@ async fn enqueue_next_job( if let Some(request) = queue.into_iter().next() { if request.status != BenchmarkRequestStatus::InProgress { // TODO: - // - Remove this if condition + // - Uncomment // - Actually enqueue the jobs - if !should_add_to_db { - log::info!("{:?} would have been marked as InProgress", request); - } else { - conn.update_benchmark_request_status(&request, BenchmarkRequestStatus::InProgress) - .await?; - } + // conn.update_benchmark_request_status(&request, BenchmarkRequestStatus::InProgress) + // .await?; } } @@ -284,16 +277,13 @@ async fn enqueue_next_job( } /// For queueing jobs, add the jobs you want to queue to this function -async fn cron_enqueue_jobs( - site_ctxt: &Arc, - should_add_to_db: bool, -) -> anyhow::Result<()> { +async fn cron_enqueue_jobs(site_ctxt: &Arc) -> anyhow::Result<()> { let mut conn = site_ctxt.conn().await; // Put the master commits into the `benchmark_requests` queue create_benchmark_request_master_commits(site_ctxt, &*conn).await?; // Put the releases into the `benchmark_requests` queue create_benchmark_request_releases(&*conn).await?; - enqueue_next_job(&mut *conn, should_add_to_db).await?; + enqueue_next_job(&mut *conn).await?; Ok(()) } @@ -309,7 +299,7 @@ pub async fn cron_main(site_ctxt: Arc>>>, seconds: u let guard = ctxt.read(); guard.as_ref().cloned() } { - match cron_enqueue_jobs(&ctxt_clone, false).await { + match cron_enqueue_jobs(&ctxt_clone).await { Ok(_) => log::info!("Cron job executed at: {:?}", std::time::SystemTime::now()), Err(e) => log::error!("Cron job failed to execute {}", e), } @@ -399,16 +389,6 @@ mod tests { } } - /// Get an `InProgress` item out of the `benchmark_requests` table. In - /// practice this is the job that has been enqueued. - async fn get_in_progress(conn: &dyn database::pool::Connection) -> Option { - conn.get_benchmark_requests_by_status(&[BenchmarkRequestStatus::InProgress]) - .await - .unwrap() - .first() - .cloned() - } - fn queue_order_matches(queue: &[BenchmarkRequest], expected: &[&str]) { let queue_shas: Vec<&str> = queue.iter().map(|req| req.tag()).collect(); assert_eq!(queue_shas, expected) @@ -425,206 +405,6 @@ mod tests { } } - /// Nothing to do, empty table - #[tokio::test] - async fn enqueue_next_job_no_jobs() { - run_postgres_test(|ctx| async { - let mut db = ctx.db_client().connection().await; - - enqueue_next_job(&mut *db, true).await?; - - let in_progress = get_in_progress(&*db).await; - - assert!(in_progress.is_none()); - Ok(ctx) - }) - .await; - } - - /// Parent completed -> child is picked - #[tokio::test] - async fn get_next_benchmark_request_completed_parent() { - run_postgres_test(|ctx| async { - let mut db = ctx.db_client().connection().await; - let parent = - create_master("a", "x", 1, "days5").with_status(BenchmarkRequestStatus::Completed); - let child = create_master("b", "a", 1, "days5"); - - db_insert_requests(&*db, &[parent, child]).await; - - enqueue_next_job(&mut *db, true).await?; - - let in_progress = get_in_progress(&*db).await; - - assert_eq!(in_progress.unwrap().tag(), "b"); - Ok(ctx) - }) - .await; - } - - /// Release (no parent) is always eligible - #[tokio::test] - async fn get_next_benchmark_request_no_parent_release() { - run_postgres_test(|ctx| async { - let mut db = ctx.db_client().connection().await; - let release = create_release("v1.2.3", "days2"); - - db_insert_requests(&*db, &[release]).await; - - enqueue_next_job(&mut *db, true).await?; - - let in_progress = get_in_progress(&*db).await; - - assert_eq!(in_progress.unwrap().tag(), "v1.2.3"); - Ok(ctx) - }) - .await; - } - - /// Parent exists but is older -> parent gets picked - #[tokio::test] - async fn get_next_benchmark_request_oldest_first() { - run_postgres_test(|ctx| async { - let mut db = ctx.db_client().connection().await; - let c1 = create_master("x", "x", 1, "days521") - .with_status(BenchmarkRequestStatus::Completed); - let c2 = create_master("y", "y", 2, "days521") - .with_status(BenchmarkRequestStatus::Completed); - - let m1 = create_master("old", "x", 3, "days45"); - let m2 = create_master("new", "y", 4, "days1"); - - db_insert_requests(&*db, &[c1, c2, m1, m2]).await; - enqueue_next_job(&mut *db, true).await?; - - let in_progress = get_in_progress(&*db).await; - - assert_eq!(in_progress.unwrap().tag(), "old"); - Ok(ctx) - }) - .await; - } - - /// Parent SHA missing entirely -> child is ready - #[cfg(unix)] // test will not panic on windows and would be skipped entirely - #[tokio::test] - #[should_panic(expected = "No commit is ready for benchmarking")] - async fn get_next_benchmark_request_missing_parent() { - run_postgres_test(|ctx| async { - let mut db = ctx.db_client().connection().await; - let orphan = create_master("orphan", "gone", 42, "days1"); - - db_insert_requests(&*db, &[orphan]).await; - enqueue_next_job(&mut *db, true).await?; - - let in_progress = get_in_progress(&*db).await; - assert_eq!(in_progress.unwrap().tag(), "orphan"); - - Ok(ctx) - }) - .await; - } - - #[tokio::test] - async fn get_next_benchmark_request_large_mixture() { - run_postgres_test(|ctx| async { - let mut db = ctx.db_client().connection().await; - // Fresh parents that will unblock some children - let parent_master = create_master("parent_m", "x", 911, "days5") - .with_status(BenchmarkRequestStatus::Completed); - let parent_try = create_try("parent_t", "x", 888, "days4") - .with_status(BenchmarkRequestStatus::Completed); - let parent_master_two = create_master("gp", "x", 922, "days5") - .with_status(BenchmarkRequestStatus::Completed); - let parent_master_three = create_master("blocked_p", "x", 932, "days5") - .with_status(BenchmarkRequestStatus::Completed); - - // Two releases, the older one should win overall - let rel_old = create_release("v0.8.0", "days40"); // 40days old - let rel_new = create_release("v1.0.0", "days10"); - - // Ready masters (parents completed) - let master_low_pr = create_master("m_low", "parent_m", 1, "days12"); - let master_high_pr = create_master("m_high", "parent_m", 7, "days8"); - - let blocked_parent = create_master("blocked_p", "gp", 0, "days3"); - let master_blocked = create_master("blocked_c", "blocked_p", 0, "days1"); - - // A try commit that is ready - let try_ready = create_try("t_ready", "parent_t", 42, "days2"); - - let requests = vec![ - parent_master, - parent_master_two, - parent_master_three, - parent_try, - master_high_pr, - master_low_pr, - master_blocked, - blocked_parent, - try_ready, - rel_old, - rel_new, - ]; - - db_insert_requests(&*db, &requests).await; - enqueue_next_job(&mut *db, true).await?; - - // The oldest release ("v0.8.0") outranks everything else - let in_progress = get_in_progress(&*db).await; - assert_eq!(in_progress.unwrap().tag(), "v0.8.0"); - Ok(ctx) - }) - .await; - } - - #[tokio::test] - async fn get_next_benchmark_request_large_mixture_no_release() { - run_postgres_test(|ctx| async { - let mut db = ctx.db_client().connection().await; - // Fresh parents that will unblock some children - let parent_master = create_master("parent_m", "x", 8, "days5") - .with_status(BenchmarkRequestStatus::Completed); - let parent_try = create_try("parent_t", "x", 9, "days4") - .with_status(BenchmarkRequestStatus::Completed); - let parent_master_two = create_master("gp", "x", 10, "days5") - .with_status(BenchmarkRequestStatus::Completed); - let parent_master_three = create_master("blocked_p", "x", 11, "days5") - .with_status(BenchmarkRequestStatus::Completed); - - // Ready masters (parents completed) - let m1 = create_master("m_low", "parent_m", 3, "days12"); - let m2 = create_master("m_high", "parent_m", 7, "days8"); - - let m3 = create_master("B", "gp", 1, "days3"); - let m4 = create_master("C", "blocked_p", 2, "days1"); - - // A try commit that is ready - let t1 = create_try("t_ready", "parent_t", 42, "days2"); - - let requests = vec![ - parent_master, - parent_master_two, - parent_master_three, - parent_try, - m2, - m1, - m4, - m3, - t1, - ]; - - db_insert_requests(&*db, &requests).await; - enqueue_next_job(&mut *db, true).await?; - - // The oldest release ("v0.8.0") outranks everything else - let in_progress = get_in_progress(&*db).await; - assert_eq!(in_progress.unwrap().tag(), "B"); - Ok(ctx) - }) - .await; - } - #[tokio::test] async fn queue_ordering() { run_postgres_test(|ctx| async {