From 2c559b452f38cc86fe45bcdb98574245fd2cf187 Mon Sep 17 00:00:00 2001 From: Ruben Fiszel Date: Tue, 20 Oct 2026 19:11:46 +0200 Subject: [PATCH] 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 * fix: count only earlier occurrences as schedule handler history Co-Authored-By: Claude Opus 5.5 * 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 Co-authored-by: windmill-internal-app[bot] --- ...225e4e0828d82f5c6ddca9b059f4a876d137.json} | 4 +- ...4fe6f508293f3334e4e17e7d396b52012a7d.json} | 4 +- backend/ee-repo-ref.txt | 2 +- backend/tests/pool_hold_and_wait.rs | 200 ++++++++++++++++++ backend/windmill-common/src/lib.rs | 5 + backend/windmill-queue/src/jobs.rs | 89 ++++---- .../windmill-queue/tests/native_retry_test.rs | 5 +- backend/windmill-worker/src/worker_flow.rs | 9 +- 8 files changed, 266 insertions(+), 52 deletions(-) rename backend/.sqlx/{query-d537182f8c8931427f2cf175bd850ce2581a33c8678212de99edb6f823f38994.json => query-7c481a2cf03e7f80e7170a5165b5225e4e0828d82f5c6ddca9b059f4a876d137.json} (79%) rename backend/.sqlx/{query-5b0847d2b95a128a5b648dd4847af44ed0992ced76286a08a33134140a454391.json => query-92710770a07d9cc622fd36bde7114fe6f508293f3334e4e17e7d396b52012a7d.json} (84%) create mode 100644 backend/tests/pool_hold_and_wait.rs diff --git a/backend/.sqlx/query-d537182f8c8931427f2cf175bd850ce2581a33c8678212de99edb6f823f38994.json b/backend/.sqlx/query-7c481a2cf03e7f80e7170a5165b5225e4e0828d82f5c6ddca9b059f4a876d137.json similarity index 79% rename from backend/.sqlx/query-d537182f8c8931427f2cf175bd850ce2581a33c8678212de99edb6f823f38994.json rename to backend/.sqlx/query-7c481a2cf03e7f80e7170a5165b5225e4e0828d82f5c6ddca9b059f4a876d137.json index 8c2a7716fd..e637b8c87d 100644 --- a/backend/.sqlx/query-d537182f8c8931427f2cf175bd850ce2581a33c8678212de99edb6f823f38994.json +++ b/backend/.sqlx/query-7c481a2cf03e7f80e7170a5165b5225e4e0828d82f5c6ddca9b059f4a876d137.json @@ -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" } diff --git a/backend/.sqlx/query-5b0847d2b95a128a5b648dd4847af44ed0992ced76286a08a33134140a454391.json b/backend/.sqlx/query-92710770a07d9cc622fd36bde7114fe6f508293f3334e4e17e7d396b52012a7d.json similarity index 84% rename from backend/.sqlx/query-5b0847d2b95a128a5b648dd4847af44ed0992ced76286a08a33134140a454391.json rename to backend/.sqlx/query-92710770a07d9cc622fd36bde7114fe6f508293f3334e4e17e7d396b52012a7d.json index 28c09ca528..514793704f 100644 --- a/backend/.sqlx/query-5b0847d2b95a128a5b648dd4847af44ed0992ced76286a08a33134140a454391.json +++ b/backend/.sqlx/query-92710770a07d9cc622fd36bde7114fe6f508293f3334e4e17e7d396b52012a7d.json @@ -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>\",\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>\",\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" } diff --git a/backend/ee-repo-ref.txt b/backend/ee-repo-ref.txt index 5abb85e984..e61c7a7c19 100644 --- a/backend/ee-repo-ref.txt +++ b/backend/ee-repo-ref.txt @@ -1 +1 @@ -dd0c7087f9b2453260f3d73b0d589042b4393a1d +ac53164fd27fe962a020f3ee944949df1f760603 diff --git a/backend/tests/pool_hold_and_wait.rs b/backend/tests/pool_hold_and_wait.rs new file mode 100644 index 0000000000..44af05ab39 --- /dev/null +++ b/backend/tests/pool_hold_and_wait.rs @@ -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) -> Pool { + 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) { + 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) -> 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) -> 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(()) +} diff --git a/backend/windmill-common/src/lib.rs b/backend/windmill-common/src/lib.rs index 9f1ccf7211..d02ceddd1d 100644 --- a/backend/windmill-common/src/lib.rs +++ b/backend/windmill-common/src/lib.rs @@ -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? diff --git a/backend/windmill-queue/src/jobs.rs b/backend/windmill-queue/src/jobs.rs index 1b35decd97..aea82eb6fd 100644 --- a/backend/windmill-queue/src/jobs.rs +++ b/backend/windmill-queue/src/jobs.rs @@ -1641,6 +1641,7 @@ async fn commit_completed_job( } 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( // 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( 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(( diff --git a/backend/windmill-queue/tests/native_retry_test.rs b/backend/windmill-queue/tests/native_retry_test.rs index 4cf278cf83..949a3cf4f6 100644 --- a/backend/windmill-queue/tests/native_retry_test.rs +++ b/backend/windmill-queue/tests/native_retry_test.rs @@ -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) diff --git a/backend/windmill-worker/src/worker_flow.rs b/backend/windmill-worker/src/worker_flow.rs index c188f8de58..e4be72086e 100644 --- a/backend/windmill-worker/src/worker_flow.rs +++ b/backend/windmill-worker/src/worker_flow.rs @@ -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>, ) -> error::Result>>> { @@ -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,