get schedule from the database directly instead of the client

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<E: ToString + std::fmt::Debug>(
     db: &Pool<Postgres>,
-    client: &windmill_api_client::Client,
     queued_job: &QueuedJob,
     logs: String,
     e: E,
@@ -27,7 +26,6 @@ pub async fn add_completed_job_error<E: ToString + std::fmt::Debug>(
     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<E: ToString + std::fmt::Debug>(
 #[instrument(level = "trace", skip_all)]
 pub async fn add_completed_job(
     db: &Pool<Postgres>,
-    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<Transaction<'c, Postgres>> {
-    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,

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,

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(());
             }