fix flow push schedule on flow update (#6538)

* fix flow schedule

* fix flow schedule

backend/ee-repo-ref.txt

@@ -1 +1 @@
-ad0f06c836d050b3ddc1937ef3a09f6b49cf5191
+b72d44c2a8a2ad3a1e36c938863a9b02891ac4a2

backend/windmill-api/src/jobs.rs

@@ -3828,15 +3828,8 @@ pub async fn run_flow_by_path_inner(
         edited_by,
         early_return,
         ..
-    } = get_latest_flow_version_info_for_path(
-        Some(userdb_authed),
-        &db,
-        db.clone(),
-        &w_id,
-        &flow_path,
-        true,
-    )
-    .await?;
+    } = get_latest_flow_version_info_for_path(Some(userdb_authed), &db, &w_id, &flow_path, true)
+        .await?;
 
     let tag = run_query.tag.clone().or(tag);
 
@@ -5112,15 +5105,8 @@ pub async fn run_wait_result_flow_by_path_internal(
         on_behalf_of_email,
         edited_by,
         version,
-    } = get_latest_flow_version_info_for_path(
-        Some(userdb_authed),
-        &db,
-        db.clone(),
-        &w_id,
-        &flow_path,
-        true,
-    )
-    .await?;
+    } = get_latest_flow_version_info_for_path(Some(userdb_authed), &db, &w_id, &flow_path, true)
+        .await?;
 
     let tag = run_query.tag.clone().or(tag);
     check_tag_available_for_workspace(&db, &w_id, &tag, &authed).await?;

backend/windmill-api/src/trigger_helpers.rs

@@ -252,15 +252,8 @@ pub async fn get_runnable_format(
             )
         }
         RunnableId::FlowPath(path) => {
-            let FlowVersionInfo { version, .. } = get_latest_flow_version_info_for_path(
-                None,
-                db,
-                db.clone(),
-                workspace_id,
-                &path,
-                true,
-            )
-            .await?;
+            let FlowVersionInfo { version, .. } =
+                get_latest_flow_version_info_for_path(None, &db, workspace_id, &path, true).await?;
 
             let key = (
                 HubOrWorkspaceId::WorkspaceId(workspace_id.to_string()),

backend/windmill-api/src/triggers/trigger_helpers.rs

@@ -252,15 +252,8 @@ pub async fn get_runnable_format(
             )
         }
         RunnableId::FlowPath(path) => {
-            let FlowVersionInfo { version, .. } = get_latest_flow_version_info_for_path(
-                None,
-                db,
-                db.clone(),
-                workspace_id,
-                &path,
-                true,
-            )
-            .await?;
+            let FlowVersionInfo { version, .. } =
+                get_latest_flow_version_info_for_path(None, &db, workspace_id, &path, true).await?;
 
             let key = (
                 HubOrWorkspaceId::WorkspaceId(workspace_id.to_string()),

backend/windmill-common/src/jobs.rs

@@ -566,7 +566,7 @@ pub async fn get_payload_tag_from_prefixed_path(
     } else if path.starts_with("flow/") {
         let path = path.strip_prefix("flow/").unwrap().to_string();
         let FlowVersionInfo { dedicated_worker, tag, version, .. } =
-            get_latest_flow_version_info_for_path(None, db, db.clone(), w_id, &path, true).await?;
+            get_latest_flow_version_info_for_path(None, &db, w_id, &path, true).await?;
         (
             JobPayload::Flow { path, dedicated_worker, apply_preprocessor: false, version },
             tag,

backend/windmill-common/src/lib.rs

@@ -509,6 +509,7 @@ pub fn get_latest_deployed_hash_for_path<'e>(
                             computed_hash.unwrap_or_else(|| PermsCache::compute_hash(authed)),
                             ScriptHash(hash),
                         );
+                    } else {
                     }
                     hash
                 } else {
@@ -628,18 +629,17 @@ impl Into<u64> for CachedFlowPath {
         hasher.finish()
     }
 }
-pub fn get_latest_flow_version_info_for_path<
+pub fn get_latest_flow_version_id_for_path<
     'a,
     'e,
     A: sqlx::Acquire<'e, Database = Postgres> + Send + 'a,
 >(
     db_authed: Option<UserDbWithAuthed<'e, AuthedRef<'e>>>,
     db: A,
-    db2: DB,
     w_id: &'a str,
     path: &'a str,
     use_cache: bool,
-) -> impl Future<Output = error::Result<FlowVersionInfo>> + Send + 'a
+) -> impl Future<Output = error::Result<i64>> + Send + 'a
 where
     'e: 'a,
 {
@@ -697,7 +697,22 @@ where
                 version
             }
         };
+        Ok(version)
+    }
+}
 
+pub fn get_latest_flow_version_info_for_path_from_version<
+    'a,
+    'e,
+    A: sqlx::Acquire<'e, Database = Postgres> + Send + 'a,
+>(
+    db: A,
+    version: i64,
+    w_id: &'a str,
+    path: &'a str,
+) -> impl Future<Output = error::Result<FlowVersionInfo>> + Send + 'a {
+    async move {
+        // as instructed in the docstring of sqlx::Acquire
         let key = (w_id.to_string(), version);
 
         match FLOW_INFO_CACHE.get(&key) {
@@ -707,7 +722,7 @@ where
             }
             _ => {
                 tracing::debug!("Fetching flow version info for {version} ({path})");
-                let mut conn = db2.acquire().await?;
+                let mut conn = db.acquire().await?;
                 let info = sqlx::query_as!(
                     FlowVersionInfo,
                     "SELECT tag, dedicated_worker, flow_version.value->>'early_return' as early_return, flow_version.value->>'preprocessor_module' IS NOT NULL as has_preprocessor, on_behalf_of_email, edited_by, flow_version.id AS version
@@ -732,6 +747,19 @@ where
     }
 }
 
+pub async fn get_latest_flow_version_info_for_path<'e>(
+    db_authed: Option<UserDbWithAuthed<'e, AuthedRef<'e>>>,
+    db: &DB,
+    w_id: &'e str,
+    path: &'e str,
+    use_cache: bool,
+) -> error::Result<FlowVersionInfo> {
+    // as instructed in the docstring of sqlx::Acquire
+    let version =
+        get_latest_flow_version_id_for_path(db_authed, &db.clone(), w_id, path, use_cache).await?;
+    get_latest_flow_version_info_for_path_from_version(db, version, w_id, path).await
+}
+
 async fn get_latest_flow_version_for_path<'e, E: sqlx::PgExecutor<'e>>(
     db: E,
     w_id: &str,

backend/windmill-queue/src/schedule.rs

@@ -15,7 +15,8 @@ use std::str::FromStr;
 use windmill_common::db::Authed;
 use windmill_common::ee_oss::LICENSE_KEY_VALID;
 use windmill_common::flows::Retry;
-use windmill_common::get_latest_flow_version_info_for_path;
+use windmill_common::get_latest_flow_version_id_for_path;
+use windmill_common::get_latest_flow_version_info_for_path_from_version;
 use windmill_common::jobs::check_tag_available_for_workspace_internal;
 use windmill_common::jobs::JobPayload;
 use windmill_common::schedule::schedule_to_user;
@@ -117,17 +118,24 @@ pub async fn push_scheduled_job<'c>(
     }
 
     let (payload, tag, timeout, on_behalf_of_email, created_by) = if schedule.is_flow {
-        let FlowVersionInfo {
-            version, tag, dedicated_worker, on_behalf_of_email, edited_by, ..
-        } = get_latest_flow_version_info_for_path(
+        let version = get_latest_flow_version_id_for_path(
             None,
             &mut *tx,
-            db.clone(),
             &schedule.workspace_id,
             &schedule.script_path,
             false,
         )
         .await?;
+
+        let FlowVersionInfo {
+            version, tag, dedicated_worker, on_behalf_of_email, edited_by, ..
+        } = get_latest_flow_version_info_for_path_from_version(
+            &mut *tx,
+            version,
+            &schedule.workspace_id,
+            &schedule.script_path,
+        )
+        .await?;
         (
             JobPayload::Flow {
                 path: schedule.script_path.clone(),

backend/windmill-worker/src/worker_flow.rs

@@ -4251,7 +4251,7 @@ async fn flow_to_payload(
     db: &DB,
 ) -> Result<JobPayloadWithTag, Error> {
     let FlowVersionInfo { version, on_behalf_of_email, edited_by, tag, .. } =
-        get_latest_flow_version_info_for_path(None, db, db.clone(), w_id, &path, true).await?;
+        get_latest_flow_version_info_for_path(None, &db, w_id, &path, true).await?;
     let on_behalf_of = if let Some(email) = on_behalf_of_email {
         Some(OnBehalfOf { email, permissioned_as: username_to_permissioned_as(&edited_by) })
     } else {