fix: stop holding a pool connection while waiting for another (#11902)

* fix: stop holding a pool connection while waiting for another Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * fix: count only earlier occurrences as schedule handler history Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * chore: update ee-repo-ref to ac53164fd27fe962a020f3ee944949df1f760603 This commit updates the EE repository reference after PR #904 was merged in windmill-ee-private. Previous ee-repo-ref: 047764f902f10fce0156138c4240ba5d1a2e969a New ee-repo-ref: ac53164fd27fe962a020f3ee944949df1f760603 Automated by sync-ee-ref workflow. --------- Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com> Co-authored-by: windmill-internal-app[bot] <windmill-internal-app[bot]@users.noreply.github.com>
2026-10-22 08:03:38 +00:00 · 2026-10-20 19:11:46 +02:00
parent 613adb5287
commit 2c559b452f
8 changed files with 266 additions and 52 deletions
@@ -1,6 +1,6 @@
 {
  "db_name": "PostgreSQL",
-  "query": "SELECT (status = 'success' OR EXISTS (\n                            SELECT 1 FROM native_retry_attempt nra\n                            JOIN v2_job jc ON jc.id = nra.job_id\n                            JOIN v2_job_completed cc ON cc.id = nra.job_id\n                            WHERE jc.parent_job = j.id AND cc.status = 'success'\n                        )) AS \"success!\"\n                    FROM v2_job j JOIN v2_job_completed USING (id)\n                    WHERE j.workspace_id = $1 AND trigger_kind = 'schedule' AND trigger = $2\n                        AND parent_job IS NULL\n                        AND runnable_path = $3\n                        AND j.id != $4\n                    ORDER BY created_at DESC\n                    LIMIT $5",
+  "query": "SELECT (status = 'success' OR EXISTS (\n                            SELECT 1 FROM native_retry_attempt nra\n                            JOIN v2_job jc ON jc.id = nra.job_id\n                            JOIN v2_job_completed cc ON cc.id = nra.job_id\n                            WHERE jc.parent_job = j.id AND cc.status = 'success'\n                        )) AS \"success!\"\n                    FROM v2_job j JOIN v2_job_completed USING (id)\n                    WHERE j.workspace_id = $1 AND trigger_kind = 'schedule' AND trigger = $2\n                        AND parent_job IS NULL\n                        AND runnable_path = $3\n                        AND j.created_at < (SELECT created_at FROM v2_job WHERE id = $4)\n                    ORDER BY created_at DESC\n                    LIMIT $5",
  "describe": {
    "columns": [
      {
@@ -22,5 +22,5 @@
      null
    ]
  },
-  "hash": "d537182f8c8931427f2cf175bd850ce2581a33c8678212de99edb6f823f38994"
+  "hash": "7c481a2cf03e7f80e7170a5165b5225e4e0828d82f5c6ddca9b059f4a876d137"
 }
@@ -1,6 +1,6 @@
 {
  "db_name": "PostgreSQL",
-  "query": "SELECT (status = 'success' OR EXISTS (\n                        SELECT 1 FROM native_retry_attempt nra\n                        JOIN v2_job jc ON jc.id = nra.job_id\n                        JOIN v2_job_completed cc ON cc.id = nra.job_id\n                        WHERE jc.parent_job = j.id AND cc.status = 'success'\n                    )) AS \"success!\",\n                    result AS \"result: Json<Box<RawValue>>\",\n                    started_at AS \"started_at!\"FROM v2_job j JOIN v2_job_completed USING (id)\n                WHERE j.workspace_id = $1 AND trigger_kind = 'schedule' AND trigger = $2\n                    AND parent_job IS NULL\n                    AND runnable_path = $3\n                    AND j.id != $4\n                ORDER BY created_at DESC\n                LIMIT $5",
+  "query": "SELECT (status = 'success' OR EXISTS (\n                        SELECT 1 FROM native_retry_attempt nra\n                        JOIN v2_job jc ON jc.id = nra.job_id\n                        JOIN v2_job_completed cc ON cc.id = nra.job_id\n                        WHERE jc.parent_job = j.id AND cc.status = 'success'\n                    )) AS \"success!\",\n                    result AS \"result: Json<Box<RawValue>>\",\n                    started_at AS \"started_at!\"FROM v2_job j JOIN v2_job_completed USING (id)\n                WHERE j.workspace_id = $1 AND trigger_kind = 'schedule' AND trigger = $2\n                    AND parent_job IS NULL\n                    AND runnable_path = $3\n                    AND j.created_at < (SELECT created_at FROM v2_job WHERE id = $4)\n                ORDER BY created_at DESC\n                LIMIT $5",
  "describe": {
    "columns": [
      {
@@ -34,5 +34,5 @@
      true
    ]
  },
-  "hash": "5b0847d2b95a128a5b648dd4847af44ed0992ced76286a08a33134140a454391"
+  "hash": "92710770a07d9cc622fd36bde7114fe6f508293f3334e4e17e7d396b52012a7d"
 }
@@ -1 +1 @@
-dd0c7087f9b2453260f3d73b0d589042b4393a1d
+ac53164fd27fe962a020f3ee944949df1f760603
@@ -0,0 +1,200 @@
+//! Paths that hold a pool connection (an open transaction or an acquired connection) must not
+//! wait for a second one from the same pool: a burst the size of the pool would have every
+//! holder wait for another until the acquire timeout. On a one-connection pool any such second
+//! checkout hangs.
+
+use std::time::Duration;
+
+use serde_json::json;
+use sqlx::{postgres::PgPoolOptions, Pool, Postgres};
+use windmill_common::{
+    db::{AuthedRef, UserDB, UserDbWithAuthed},
+    error::Error,
+    jobs::JobPayload,
+    worker::Connection,
+};
+use windmill_test_utils::*;
+
+async fn one_connection_pool(db: &Pool<Postgres>) -> Pool<Postgres> {
+    PgPoolOptions::new()
+        .max_connections(1)
+        .acquire_timeout(Duration::from_secs(5))
+        .connect_with((*db.connect_options()).clone())
+        .await
+        .unwrap()
+}
+
+#[sqlx::test(fixtures("base"))]
+async fn run_by_path_of_a_missing_script_uses_one_connection(db: Pool<Postgres>) {
+    let one = one_connection_pool(&db).await;
+    let groups = vec![];
+    let folders = vec![];
+    let authed = AuthedRef {
+        email: "test2@windmill.dev",
+        username: "test-user-2",
+        is_admin: &false,
+        is_operator: &false,
+        groups: &groups,
+        folders: &folders,
+        scopes: &None,
+        token_prefix: &None,
+    };
+    let lookup = windmill_common::get_latest_deployed_hash_for_path(
+        Some(UserDbWithAuthed { authed: &authed, db: UserDB::new(one.clone()) }),
+        one.clone(),
+        "test-workspace",
+        "f/missing/script",
+    );
+    let res = tokio::time::timeout(Duration::from_secs(15), lookup)
+        .await
+        .unwrap();
+    assert!(matches!(res, Err(Error::NotFound(_))), "{:?}", res.err());
+}
+
+#[sqlx::test(fixtures("base"))]
+async fn for_loop_flow_runs_on_one_connection(db: Pool<Postgres>) -> anyhow::Result<()> {
+    initialize_tracing().await;
+    let server = ApiServer::start(db.clone()).await?;
+    let flow = serde_json::from_value(json!({
+        "modules": [{
+            "id": "loop",
+            "value": {
+                "type": "forloopflow",
+                "iterator": { "type": "static", "value": [1, 2] },
+                "skip_failures": false,
+                "parallel": false,
+                "modules": [{ "id": "inner", "value": { "type": "identity" } }]
+            }
+        }]
+    }))?;
+    let job = RunJob::from(JobPayload::RawFlow { value: flow, path: None, restarted_from: None })
+        .push(&db)
+        .await;
+
+    let one = one_connection_pool(&db).await;
+    let db2 = db.clone();
+    in_test_worker(
+        Connection::Sql(one),
+        async move {
+            while sqlx::query_scalar::<_, bool>(
+                "SELECT EXISTS (SELECT 1 FROM v2_job_completed WHERE id = $1)",
+            )
+            .bind(job)
+            .fetch_one(&db2)
+            .await
+            .unwrap()
+                == false
+            {
+                tokio::time::sleep(Duration::from_millis(100)).await;
+            }
+        },
+        server.addr.port(),
+    )
+    .await;
+
+    let status: String =
+        sqlx::query_scalar("SELECT status::text FROM v2_job_completed WHERE id = $1")
+            .bind(job)
+            .fetch_one(&db)
+            .await?;
+    assert_eq!(status, "success");
+    Ok(())
+}
+
+#[cfg(all(feature = "enterprise", feature = "private"))]
+#[sqlx::test(fixtures("base", "schedule"))]
+async fn schedule_handlers_run_on_one_connection(db: Pool<Postgres>) -> anyhow::Result<()> {
+    use windmill_queue::{add_completed_job, get_mini_completed_job};
+
+    const SCHEDULE: &str = "f/system/handled_schedule";
+    const HANDLER: &str = "script/f/system/schedule_recovery_handler";
+    sqlx::query(
+        "INSERT INTO schedule (workspace_id, path, edited_by, schedule, enabled, script_path, args,
+            is_flow, email, timezone, permissioned_as, on_success, on_recovery)
+         VALUES ('test-workspace', $1, 'test-user', '0 0 0 1 1 *', true, 'f/system/failing_script',
+            '{}', false, 'test@windmill.dev', 'UTC', 'u/test-user', $2, $2)",
+    )
+    .bind(SCHEDULE)
+    .bind(HANDLER)
+    .execute(&db)
+    .await?;
+
+    // The previous occurrence failed, so this successful one is a recovery. The handlers run
+    // once the completion has committed, when the next tick may already have completed: a
+    // later occurrence must not count as history, or its success would hide the recovery.
+    let insert_job =
+        "INSERT INTO v2_job (id, workspace_id, created_by, created_at, permissioned_as,
+            permissioned_as_email, kind, runnable_id, runnable_path, script_lang, tag,
+            visible_to_owner, trigger, trigger_kind)
+         VALUES ($1, 'test-workspace', 'test-user', now() - $2 * interval '1 minute', 'u/test-user',
+            'test@windmill.dev', 'script', 12349, 'f/system/failing_script', 'bun', 'bun', true,
+            $3, 'schedule')";
+    for (minutes_ago, status) in [(1.0, "failure"), (-1.0, "success")] {
+        let id = uuid::Uuid::new_v4();
+        sqlx::query(insert_job)
+            .bind(id)
+            .bind(minutes_ago)
+            .bind(SCHEDULE)
+            .execute(&db)
+            .await?;
+        sqlx::query(
+            "INSERT INTO v2_job_completed (id, workspace_id, duration_ms, result, started_at,
+                completed_at, status)
+             VALUES ($1, 'test-workspace', 1, '{}', now(), now(), $2::job_status)",
+        )
+        .bind(id)
+        .bind(status)
+        .execute(&db)
+        .await?;
+    }
+
+    let running = uuid::Uuid::new_v4();
+    sqlx::query(insert_job)
+        .bind(running)
+        .bind(0.0)
+        .bind(SCHEDULE)
+        .execute(&db)
+        .await?;
+    sqlx::query(
+        "INSERT INTO v2_job_queue (id, workspace_id, scheduled_for, running, started_at, tag)
+         VALUES ($1, 'test-workspace', now(), true, now(), 'bun')",
+    )
+    .bind(running)
+    .execute(&db)
+    .await?;
+    sqlx::query("INSERT INTO v2_job_runtime (id, ping) VALUES ($1, now())")
+        .bind(running)
+        .execute(&db)
+        .await?;
+
+    let one = one_connection_pool(&db).await;
+    let job = get_mini_completed_job(&running, "test-workspace", &db)
+        .await?
+        .unwrap();
+    let result = json!("OK");
+    let complete = add_completed_job(
+        &one,
+        &job,
+        true,
+        false,
+        sqlx::types::Json(&result),
+        None,
+        0,
+        None,
+        false,
+        None,
+        false,
+    );
+    tokio::time::timeout(Duration::from_secs(30), complete).await??;
+
+    // One on_success job and one on_recovery job.
+    let handlers: i64 = sqlx::query_scalar(
+        "SELECT count(*) FROM v2_job WHERE runnable_path = 'f/system/schedule_recovery_handler'
+            AND parent_job = $1",
+    )
+    .bind(running)
+    .fetch_one(&db)
+    .await?;
+    assert_eq!(handlers, 2);
+    Ok(())
+}
@@ -2443,6 +2443,10 @@ pub fn get_latest_deployed_hash_for_path<'e>(
                            ScriptHash(hash),
                        );
                    } else {
+                        // Release the user-db connection first: holding it while taking a
+                        // second one from the same pool starves the pool under a burst of
+                        // requests for paths the caller cannot see.
+                        drop(conn);
                        let mut conn = db2.acquire().await?;
                        let exists = get_latest_script_hash(&mut *conn, script_path, w_id)
                            .await?
@@ -2847,6 +2851,7 @@ where
                            CachedFlowPath(path.to_string()),
                        );
                    } else {
+                        drop(conn);
                        let mut conn = db.acquire().await?;
                        let exists = get_latest_flow_version_for_path(&mut *conn, w_id, path)
                            .await?
@@ -1641,6 +1641,7 @@ async fn commit_completed_job<T: Serialize + Send + Sync + ValidableJson>(
    }

    let mut _skip_downstream_error_handlers = false;
+    let mut schedule_for_handlers = None;
    if completed_job.is_flow_step() {
        if let Some(parent_job) = monitor_parent {
            let r = sqlx::query_scalar!(
@@ -1720,49 +1721,9 @@ async fn commit_completed_job<T: Serialize + Send + Sync + ValidableJson>(
                // per-occurrence failure/recovery status across the whole retry
                // chain, so multi-count/exact handler policies work even though
                // each attempt is its own completed job.
-                #[cfg(all(feature = "enterprise", feature = "private"))]
                if !retry_pending {
-                    if let Err(err) = crate::jobs_ee::apply_schedule_handlers(
-                        db,
-                        &schedule,
-                        &script_path,
-                        &completed_job.workspace_id,
-                        success,
-                        result,
-                        job_id,
-                        // Current occurrence's root: the terminal native-retry
-                        // attempt's parent, else the job itself.
-                        completed_job.parent_job.unwrap_or(job_id),
-                        completed_job.started_at.unwrap_or(chrono::Utc::now()),
-                        completed_job.priority,
-                    )
-                    .warn_after_seconds(10)
-                    .await
-                    {
-                        if !success {
-                            tracing::error!("Could not apply schedule error handler: {}", err);
-                            let base_url = windmill_common::BASE_URL.load();
-                            let w_id: &String = &completed_job.workspace_id;
-                            if !matches!(err, Error::QuotaExceeded(_)) {
-                                report_error_to_workspace_handler_or_critical_side_channel(
-                                        &completed_job,
-                                        db,
-                                        format!(
-                                            "Failed to push schedule error handler job to handle failed job ({base_url}/run/{}?workspace={w_id}): {}",
-                                            completed_job.id,
-                                            err
-                                        ),
-                                    )
-                                    .warn_after_seconds(10)
-                                    .await;
-                            }
-                        } else {
-                            tracing::error!("Could not apply schedule recovery handler: {}", err);
-                        }
-                    };
+                    schedule_for_handlers = Some((schedule, script_path));
                }
-                #[cfg(not(all(feature = "enterprise", feature = "private")))]
-                let _ = retry_pending;
            } else {
                tracing::error!(
                        "Schedule {schedule_path} in {} not found. Impossible to schedule again and apply schedule handlers",
@@ -1824,6 +1785,52 @@ async fn commit_completed_job<T: Serialize + Send + Sync + ValidableJson>(

    tx.commit().warn_after_seconds(10).await?;

+    // The handlers read and push through `db`, so they run after the commit: inside the
+    // transaction they would hold its connection while waiting for more from the same pool.
+    #[cfg(all(feature = "enterprise", feature = "private"))]
+    if let Some((schedule, script_path)) = schedule_for_handlers {
+        if let Err(err) = crate::jobs_ee::apply_schedule_handlers(
+            db,
+            &schedule,
+            script_path,
+            &completed_job.workspace_id,
+            success,
+            result,
+            job_id,
+            // Current occurrence's root: the terminal native-retry
+            // attempt's parent, else the job itself.
+            completed_job.parent_job.unwrap_or(job_id),
+            completed_job.started_at.unwrap_or(chrono::Utc::now()),
+            completed_job.priority,
+        )
+        .warn_after_seconds(10)
+        .await
+        {
+            if !success {
+                tracing::error!("Could not apply schedule error handler: {}", err);
+                let base_url = windmill_common::BASE_URL.load();
+                let w_id: &String = &completed_job.workspace_id;
+                if !matches!(err, Error::QuotaExceeded(_)) {
+                    report_error_to_workspace_handler_or_critical_side_channel(
+                            &completed_job,
+                            db,
+                            format!(
+                                "Failed to push schedule error handler job to handle failed job ({base_url}/run/{}?workspace={w_id}): {}",
+                                completed_job.id,
+                                err
+                            ),
+                        )
+                        .warn_after_seconds(10)
+                        .await;
+                }
+            } else {
+                tracing::error!("Could not apply schedule recovery handler: {}", err);
+            }
+        };
+    }
+    #[cfg(not(all(feature = "enterprise", feature = "private")))]
+    let _ = schedule_for_handlers;
+
    log_completed_job(completed_job, completed.duration_ms, success);
    // tracing::info!("completed job: {:?}", start.elapsed().as_micros());
    Ok((
@@ -339,7 +339,7 @@ mod native_retry {
        //    -> failure: handler child must NOT count as a recovery
        // f: root fails, only a WAC inline child (unmarked) succeeds
        //    -> failure: WAC inline child must NOT count as a recovery
-        // d: the current occurrence (excluded by `j.id != $4`)
+        // d: the current occurrence (excluded with any later one by the created_at bound)
        let (a, b, c, e, f, d) = (
            Uuid::new_v4(),
            Uuid::new_v4(),
@@ -371,7 +371,8 @@ mod native_retry {
                ))
             FROM v2_job j JOIN v2_job_completed USING (id)
             WHERE j.workspace_id = $1 AND trigger_kind = 'schedule' AND trigger = $2
-                AND parent_job IS NULL AND runnable_path = $3 AND j.id != $4
+                AND parent_job IS NULL AND runnable_path = $3
+                AND j.created_at < (SELECT created_at FROM v2_job WHERE id = $4)
             ORDER BY created_at DESC",
        )
        .bind(WS)
@@ -110,8 +110,8 @@ lazy_static::lazy_static! {

 /// Helper function to write itered data to separate table
 /// Returns None if data was written to separate table, Some(itered) if it should be stored in JSONB
-async fn write_itered_to_db(
-    db: &DB,
+async fn write_itered_to_db<'c>(
+    db: impl sqlx::PgExecutor<'c>,
    job_id: Uuid,
    itered: &Vec<Box<RawValue>>,
 ) -> error::Result<Option<Vec<Box<RawValue>>>> {
@@ -4780,7 +4780,8 @@ async fn push_next_flow_job(
        {
            // A step with its own on-behalf-of carries a cached dispatch address, up to one
            // notify poll stale; accepted, see `get_email_from_permissioned_as`.
-            let is_super_admin = windmill_common::auth::is_super_admin_email(db, email).await?;
+            let is_super_admin =
+                windmill_common::auth::is_super_admin_email(&mut *tx, email).await?;
            check_tag_written_as_available_for_push(
                db,
                &flow_job.workspace_id,
@@ -5046,7 +5047,7 @@ async fn push_next_flow_job(

            // Conditionally write itered to separate table if all workers support it
            // Returns None if written to table, Some(itered) if should be in JSONB
-            let itered_for_status = write_itered_to_db(db, flow_job.id, &itered).await?;
+            let itered_for_status = write_itered_to_db(&mut *tx, flow_job.id, &itered).await?;

            FlowStatusModule::InProgress {
                job: uuid,