From e6c8ef18a11f079083138a991543f7e273d4e511 Mon Sep 17 00:00:00 2001 From: Ruben Fiszel Date: Tue, 20 Oct 2026 18:31:38 +0200 Subject: [PATCH] fix: write a push's runnable settings before it opens its transaction (#11897) Co-authored-by: Claude Opus 5.5 --- backend/tests/push_pool_connections.rs | 63 ++++++++++++++++++++++ backend/windmill-queue/src/jobs.rs | 72 ++++++++++++++++---------- 2 files changed, 109 insertions(+), 26 deletions(-) create mode 100644 backend/tests/push_pool_connections.rs diff --git a/backend/tests/push_pool_connections.rs b/backend/tests/push_pool_connections.rs new file mode 100644 index 0000000000..006340cf8a --- /dev/null +++ b/backend/tests/push_pool_connections.rs @@ -0,0 +1,63 @@ +//! `push` must not need a second pool connection while it holds its transaction: a burst of +//! pushes the size of the pool would each hold one and wait for another until the acquire +//! timeout. On a one-connection pool any such second checkout hangs. + +use std::time::Duration; + +use sqlx::{postgres::PgPoolOptions, Pool, Postgres}; +use windmill_common::jobs::JobPayload; +use windmill_test_utils::RunJob; + +#[sqlx::test(fixtures("base"))] +async fn concurrency_limited_flow_pushes_on_one_connection(db: Pool) { + // The flow and settings caches outlive the test database (they are on disk), so only a + // version and a limit no run has used yet make `push` read the flow and write its settings. + let limit = (uuid::Uuid::new_v4().as_u128() % 1_000_000_000) as i32 + 1; + let version = limit as i64; + let value = format!(r#"{{"modules":[],"concurrent_limit":{limit}}}"#); + sqlx::query( + "WITH f AS ( + INSERT INTO flow (workspace_id, path, summary, description, value, edited_by, schema, extra_perms, versions) + VALUES ('test-workspace', 'f/t/limited', '', '', $1::jsonb, 'test-user', '{}', '{}', ARRAY[$2]) + ) + INSERT INTO flow_version (id, workspace_id, path, value, schema, created_by) + VALUES ($2, 'test-workspace', 'f/t/limited', $1::jsonb, '{}', 'test-user')", + ) + .bind(&value) + .bind(version) + .execute(&db) + .await + .unwrap(); + + let one = PgPoolOptions::new() + .max_connections(1) + .acquire_timeout(Duration::from_secs(5)) + .connect_with((*db.connect_options()).clone()) + .await + .unwrap(); + + let push = RunJob::from(JobPayload::Flow { + path: "f/t/limited".to_string(), + dedicated_worker: None, + apply_preprocessor: false, + version, + labels: None, + job_token_scopes: None, + }) + .push(&one); + let job = tokio::time::timeout(Duration::from_secs(15), push) + .await + .unwrap(); + + let stored: Option = sqlx::query_scalar( + "SELECT cs.concurrent_limit FROM v2_job_queue q + JOIN runnable_settings rs ON rs.hash = q.runnable_settings_handle + JOIN concurrency_settings cs ON cs.hash = rs.concurrency_settings + WHERE q.id = $1", + ) + .bind(job) + .fetch_one(&db) + .await + .unwrap(); + assert_eq!(stored, Some(limit)); +} diff --git a/backend/windmill-queue/src/jobs.rs b/backend/windmill-queue/src/jobs.rs index eafb344870..1b35decd97 100644 --- a/backend/windmill-queue/src/jobs.rs +++ b/backend/windmill-queue/src/jobs.rs @@ -6200,6 +6200,38 @@ pub async fn push<'c, 'd>( .await } +/// Reads a flow version through the pool unless the caller handed `push` its own transaction, +/// which may hold that version uncommitted. Opening a transaction just for this read would hold +/// its connection while `push` still writes the runnable settings through the pool. +async fn fetch_pushed_flow_version( + tx: &mut PushIsolationLevel<'_>, + db: &Pool, + version: i64, +) -> error::Result> { + // Do not use the lite version unless all workers are updated. + let lite = + !(*DISABLE_FLOW_SCRIPT || (!MIN_VERSION_IS_AT_LEAST_1_432.met().await && !*CLOUD_HOSTED)); + // Fallback to the original version if the lite version is not found. + // This also prevent a race condition where the flow is run just after deploy and + // the lite version is still being created. + macro_rules! fetch { + ($e:expr) => { + if lite { + match cache::flow::fetch_version_lite($e, version).await { + Ok(data) => Ok(data), + Err(_) => cache::flow::fetch_version($e, version).await, + } + } else { + cache::flow::fetch_version($e, version).await + } + }; + } + match tx { + PushIsolationLevel::Transaction(ntx) => fetch!(&mut **ntx), + PushIsolationLevel::Isolated(..) | PushIsolationLevel::IsolatedRoot(_) => fetch!(db), + } +} + // #[instrument(level = "trace", skip_all)] async fn push_inner<'c, 'd>( db: &Pool, @@ -7078,22 +7110,7 @@ async fn push_inner<'c, 'd>( labels, job_token_scopes, } => { - let mut ntx = tx.into_tx().await?; - // Do not use the lite version unless all workers are updated. - let data = if *DISABLE_FLOW_SCRIPT - || (!MIN_VERSION_IS_AT_LEAST_1_432.met().await && !*CLOUD_HOSTED) - { - cache::flow::fetch_version(&mut *ntx, version).await - } else { - // Fallback to the original version if the lite version is not found. - // This also prevent a race condition where the flow is run just after deploy and - // the lite version is still being created. - match cache::flow::fetch_version_lite(&mut *ntx, version).await { - Ok(data) => Ok(data), - Err(_) => cache::flow::fetch_version(&mut *ntx, version).await, - } - }?; - tx = PushIsolationLevel::Transaction(ntx); + let data = fetch_pushed_flow_version(&mut tx, db, version).await?; let mut value = data.value().clone(); refuse_step_scopes_on_outdated_workers(&value).await?; @@ -7463,6 +7480,19 @@ async fn push_inner<'c, 'd>( crate::tags::apply_dev_workspace_suffix(tag, workspace_id, db).await? }; + // Before `into_tx`: these write through the pool on a cache miss, and a burst of pushes each + // holding its transaction would then wait on each other for a second connection. They stay + // off `tx` because they are cached as written, which a rollback would undo. + let runnable_settings_handle = windmill_common::runnable_settings::insert_rs( + RunnableSettings { + debouncing_settings: debouncing_settings.insert_cached(db).await?, + concurrency_settings: concurrency_settings.insert_cached(db).await?, + retry_settings: retry_settings.insert_cached(db).await?, + }, + db, + ) + .await?; + let mut tx = tx.into_tx().await?; let job_id: Uuid = if let Some(job_id) = job_id { @@ -7642,16 +7672,6 @@ async fn push_inner<'c, 'd>( (job_kind, scheduled_for_o) }; - let runnable_settings_handle = windmill_common::runnable_settings::insert_rs( - RunnableSettings { - debouncing_settings: debouncing_settings.insert_cached(db).await?, - concurrency_settings: concurrency_settings.insert_cached(db).await?, - retry_settings: retry_settings.insert_cached(db).await?, - }, - db, - ) - .await?; - let (guarded_concurrent_limit, guarded_concurrency_time_window_s) = if windmill_common::runnable_settings::min_version_supports_runnable_settings_v0().await { (None, None)