diff --git a/backend/windmill-api/src/flows.rs b/backend/windmill-api/src/flows.rs index 49a143fb98..d6187db436 100644 --- a/backend/windmill-api/src/flows.rs +++ b/backend/windmill-api/src/flows.rs @@ -937,7 +937,7 @@ async fn update_flow( clear_schedule(&mut tx, &schedule.path, &w_id).await?; if schedule.enabled { - tx = push_scheduled_job(&db, tx, &schedule, None).await?; + tx = push_scheduled_job(&db, tx, &schedule, None, None).await?; } } diff --git a/backend/windmill-api/src/schedule.rs b/backend/windmill-api/src/schedule.rs index b5bb69f8ee..9eaa897341 100644 --- a/backend/windmill-api/src/schedule.rs +++ b/backend/windmill-api/src/schedule.rs @@ -277,7 +277,7 @@ async fn create_schedule( .await?; if ns.enabled.unwrap_or(true) { - tx = push_scheduled_job(&db, tx, &schedule, Some(&authed.clone().into())).await? + tx = push_scheduled_job(&db, tx, &schedule, Some(&authed.clone().into()), None).await? } tx.commit().await?; @@ -419,7 +419,7 @@ async fn edit_schedule( .await?; if schedule.enabled { - tx = push_scheduled_job(&db, tx, &schedule, None).await?; + tx = push_scheduled_job(&db, tx, &schedule, None, None).await?; } tx.commit().await?; @@ -683,7 +683,7 @@ pub async fn set_enabled( .await?; if payload.enabled { - tx = push_scheduled_job(&db, tx, &schedule, None).await?; + tx = push_scheduled_job(&db, tx, &schedule, None, None).await?; } tx.commit().await?; diff --git a/backend/windmill-api/src/scripts.rs b/backend/windmill-api/src/scripts.rs index c2a03f6bbb..c321f16b1c 100644 --- a/backend/windmill-api/src/scripts.rs +++ b/backend/windmill-api/src/scripts.rs @@ -900,7 +900,7 @@ async fn create_script_internal<'c>( clear_schedule(&mut tx, &schedule.path, &w_id).await?; if schedule.enabled { - tx = push_scheduled_job(&db, tx, &schedule, None).await?; + tx = push_scheduled_job(&db, tx, &schedule, None, None).await?; } } } else { diff --git a/backend/windmill-queue/src/jobs.rs b/backend/windmill-queue/src/jobs.rs index c951f06307..40b4fea108 100644 --- a/backend/windmill-queue/src/jobs.rs +++ b/backend/windmill-queue/src/jobs.rs @@ -1693,7 +1693,7 @@ pub async fn handle_maybe_scheduled_job<'c>( let push_next_job_future = (|| { tokio::time::timeout(std::time::Duration::from_secs(5), async { let mut tx = db.begin().await?; - tx = push_scheduled_job(db, tx, &schedule, None).await?; + tx = push_scheduled_job(db, tx, &schedule, None, Some(job.scheduled_for)).await?; tx.commit().await?; Ok::<(), Error>(()) }) diff --git a/backend/windmill-queue/src/schedule.rs b/backend/windmill-queue/src/schedule.rs index 4fcd22771a..6594837ffa 100644 --- a/backend/windmill-queue/src/schedule.rs +++ b/backend/windmill-queue/src/schedule.rs @@ -9,6 +9,8 @@ use crate::push; use crate::PushIsolationLevel; use anyhow::Context; +use chrono::DateTime; +use chrono::Utc; use sqlx::{PgExecutor, Postgres, Transaction}; use std::collections::HashMap; use std::str::FromStr; @@ -34,6 +36,7 @@ pub async fn push_scheduled_job<'c>( mut tx: Transaction<'c, Postgres>, schedule: &Schedule, authed: Option<&Authed>, + now_cutoff: Option>, ) -> Result> { if !*LICENSE_KEY_VALID.read().await { return Err(error::Error::BadRequest( @@ -50,6 +53,19 @@ pub async fn push_scheduled_job<'c>( let now = now_from_db(&mut *tx).await?; + let now = match now_cutoff { + Some(now_cutoff) if now_cutoff >= now => { + tracing::error!( + "now_cutoff ({:?}) is after now ({:?}) for schedule {}. Using now_cutoff + 1s. This likely means the pg clock was shifted backwards.", + now_cutoff, + now, + &schedule.path + ); + now_cutoff + chrono::Duration::seconds(1) + } + _ => now, + }; + let starting_from = match schedule.paused_until { Some(paused_until) if paused_until > now => paused_until.with_timezone(&tz), paused_until_o => {