fix: make schedule more resilient in case of pg clock shifts

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?;
         }
     }
 

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?;
 

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 {

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

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<DateTime<Utc>>,
 ) -> Result<Transaction<'c, Postgres>> {
     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 => {