diff --git a/backend/ee-repo-ref.txt b/backend/ee-repo-ref.txt index 40b748d22c..f068310e92 100644 --- a/backend/ee-repo-ref.txt +++ b/backend/ee-repo-ref.txt @@ -1 +1 @@ -ad0f06c836d050b3ddc1937ef3a09f6b49cf5191 \ No newline at end of file +b72d44c2a8a2ad3a1e36c938863a9b02891ac4a2 \ No newline at end of file diff --git a/backend/windmill-api/src/jobs.rs b/backend/windmill-api/src/jobs.rs index 4b9f64ea3c..48386582d4 100644 --- a/backend/windmill-api/src/jobs.rs +++ b/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?; diff --git a/backend/windmill-api/src/trigger_helpers.rs b/backend/windmill-api/src/trigger_helpers.rs index 73ba0250bd..14f9e42005 100644 --- a/backend/windmill-api/src/trigger_helpers.rs +++ b/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()), diff --git a/backend/windmill-api/src/triggers/trigger_helpers.rs b/backend/windmill-api/src/triggers/trigger_helpers.rs index a0bb07c7ad..8901c24fcc 100644 --- a/backend/windmill-api/src/triggers/trigger_helpers.rs +++ b/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()), diff --git a/backend/windmill-common/src/jobs.rs b/backend/windmill-common/src/jobs.rs index 1418df1165..d6b58896bc 100644 --- a/backend/windmill-common/src/jobs.rs +++ b/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, diff --git a/backend/windmill-common/src/lib.rs b/backend/windmill-common/src/lib.rs index 4980e81700..0cea7ec316 100644 --- a/backend/windmill-common/src/lib.rs +++ b/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 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>>, db: A, - db2: DB, w_id: &'a str, path: &'a str, use_cache: bool, -) -> impl Future> + Send + 'a +) -> impl Future> + 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> + 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>>, + db: &DB, + w_id: &'e str, + path: &'e str, + use_cache: bool, +) -> error::Result { + // 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, diff --git a/backend/windmill-queue/src/schedule.rs b/backend/windmill-queue/src/schedule.rs index d2fbe693e7..4fcd22771a 100644 --- a/backend/windmill-queue/src/schedule.rs +++ b/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(), diff --git a/backend/windmill-worker/src/worker_flow.rs b/backend/windmill-worker/src/worker_flow.rs index 9d75526282..2c9518255d 100644 --- a/backend/windmill-worker/src/worker_flow.rs +++ b/backend/windmill-worker/src/worker_flow.rs @@ -4251,7 +4251,7 @@ async fn flow_to_payload( db: &DB, ) -> Result { 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 {