fix: write a push's runnable settings before it opens its transaction (#11897)

Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>

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<Postgres>) {
+    // 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<i32> = 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));
+}

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<Postgres>,
+    version: i64,
+) -> error::Result<Arc<FlowData>> {
+    // 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<Postgres>,
@@ -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)