diff --git a/backend/windmill-worker/src/jobs.rs b/backend/windmill-worker/src/jobs.rs index 25cff17bdb..0af990b233 100644 --- a/backend/windmill-worker/src/jobs.rs +++ b/backend/windmill-worker/src/jobs.rs @@ -11,12 +11,11 @@ use sqlx::{Pool, Postgres, Transaction}; use tracing::instrument; use uuid::Uuid; use windmill_common::{error::Error, flow_status::FlowStatusModule}; -use windmill_queue::{delete_job, JobKind, QueuedJob}; +use windmill_queue::{delete_job, schedule::get_schedule_opt, JobKind, QueuedJob}; #[instrument(level = "trace", skip_all)] pub async fn add_completed_job_error( db: &Pool, - client: &windmill_api_client::Client, queued_job: &QueuedJob, logs: String, e: E, @@ -27,7 +26,6 @@ pub async fn add_completed_job_error( error_to_result(&mut output_map, &e); let a = add_completed_job( db, - client, &queued_job, false, false, @@ -51,7 +49,6 @@ pub fn error_to_result( #[instrument(level = "trace", skip_all)] pub async fn add_completed_job( db: &Pool, - client: &windmill_api_client::Client, queued_job: &QueuedJob, success: bool, skipped: bool, @@ -160,7 +157,6 @@ pub async fn add_completed_job( { tx = schedule_again_if_scheduled( tx, - client, queued_job.schedule_path.as_ref().unwrap(), queued_job.script_path.as_ref().unwrap(), &queued_job.workspace_id, @@ -175,21 +171,18 @@ pub async fn add_completed_job( #[instrument(level = "trace", skip_all)] pub async fn schedule_again_if_scheduled<'c>( mut tx: Transaction<'c, Postgres>, - client: &windmill_api_client::Client, schedule_path: &str, script_path: &str, w_id: &str, ) -> windmill_common::error::Result> { - let schedule = client - .get_schedule(w_id, schedule_path) - .await - .map_err(|_| { + let schedule = get_schedule_opt(&mut tx, w_id, schedule_path) + .await? + .ok_or_else(|| { Error::InternalErr(format!( "Could not find schedule {:?} for workspace {}", schedule_path, w_id )) - })? - .into_inner(); + })?; if schedule.enabled && script_path == schedule.script_path { tx = windmill_queue::schedule::push_scheduled_job( tx, @@ -199,7 +192,7 @@ pub async fn schedule_again_if_scheduled<'c>( edited_by: schedule.edited_by, edited_at: schedule.edited_at, schedule: schedule.schedule, - offset_: schedule.offset as _, + offset_: schedule.offset_, enabled: schedule.enabled, script_path: schedule.script_path, is_flow: schedule.is_flow, diff --git a/backend/windmill-worker/src/worker.rs b/backend/windmill-worker/src/worker.rs index 4057ce0ca2..1de5f0cb66 100644 --- a/backend/windmill-worker/src/worker.rs +++ b/backend/windmill-worker/src/worker.rs @@ -682,7 +682,6 @@ async fn handle_job_error( { let _ = add_completed_job_error( db, - client, &parent_job, format!("Unexpected error during flow job error handling:\n{err}"), err, @@ -696,7 +695,6 @@ async fn handle_job_error( } add_completed_job_error( db, - client, &job, format!("Unexpected error during job execution:\n{err}"), &err, @@ -834,7 +832,7 @@ async fn handle_queued_job( match result { Ok(r) => { - add_completed_job(db, client, &job, true, false, r.clone(), logs).await?; + add_completed_job(db, &job, true, false, r.clone(), logs).await?; if job.is_flow_step { if let Some(parent_job) = job.parent_job { update_flow_status_after_job_completion( @@ -878,7 +876,6 @@ async fn handle_queued_job( let (_, output_map) = add_completed_job_error( db, - client, &job, logs, error_message, diff --git a/backend/windmill-worker/src/worker_flow.rs b/backend/windmill-worker/src/worker_flow.rs index cf12a4ad34..ace46432a9 100644 --- a/backend/windmill-worker/src/worker_flow.rs +++ b/backend/windmill-worker/src/worker_flow.rs @@ -414,7 +414,6 @@ pub async fn update_flow_status_after_job_completion( { tx = schedule_again_if_scheduled( tx, - client, flow_job.schedule_path.as_ref().unwrap(), flow_job.script_path.as_ref().unwrap(), &w_id, @@ -435,7 +434,6 @@ pub async fn update_flow_status_after_job_completion( if flow_job.canceled { add_completed_job_error( db, - client, &flow_job, logs, &canceled_job_to_result(&flow_job), @@ -445,7 +443,6 @@ pub async fn update_flow_status_after_job_completion( } else { add_completed_job( db, - client, &flow_job, success, stop_early && skip_if_stop_early, @@ -470,7 +467,6 @@ pub async fn update_flow_status_after_job_completion( Err(err) => { let _ = add_completed_job_error( db, - client, &flow_job, "Unexpected error during flow chaining:\n".to_string(), err, @@ -982,8 +978,7 @@ async fn push_next_flow_job( let logs = "Timed out waiting to be resumed".to_string(); let result = json!({ "error": logs }); let _uuid = - add_completed_job(db, client, &flow_job, success, skipped, result, logs) - .await?; + add_completed_job(db, &flow_job, success, skipped, result, logs).await?; return Ok(()); }