From c22ee04657a4004dea8c81f09d911ba2215bf8ec Mon Sep 17 00:00:00 2001 From: Ruben Fiszel Date: Fri, 31 Jan 2025 00:55:51 +0100 Subject: [PATCH] fix: restart zombie job correctly handle concurrency limits (#5181) * restart concurrency limits * sqlx --- ...5d67db17904859e5f01f1d578af4e1d062a85.json | 34 ----------------- ...9c6da2c3d1e04d49306adaaaaf06e54ee8357.json | 34 +++++++++++++++++ backend/src/monitor.rs | 38 +++++++++++++++++-- 3 files changed, 68 insertions(+), 38 deletions(-) delete mode 100644 backend/.sqlx/query-84576f6add15108ade2ed3a88185d67db17904859e5f01f1d578af4e1d062a85.json create mode 100644 backend/.sqlx/query-9a9b639611459659ae355a43f219c6da2c3d1e04d49306adaaaaf06e54ee8357.json diff --git a/backend/.sqlx/query-84576f6add15108ade2ed3a88185d67db17904859e5f01f1d578af4e1d062a85.json b/backend/.sqlx/query-84576f6add15108ade2ed3a88185d67db17904859e5f01f1d578af4e1d062a85.json deleted file mode 100644 index cdcf0228ca..0000000000 --- a/backend/.sqlx/query-84576f6add15108ade2ed3a88185d67db17904859e5f01f1d578af4e1d062a85.json +++ /dev/null @@ -1,34 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "UPDATE queue SET running = false, started_at = null\n WHERE last_ping < now() - ($1 || ' seconds')::interval\n AND running = true AND job_kind NOT IN ('flow', 'flowpreview', 'flownode', 'singlescriptflow') AND same_worker = false RETURNING id, workspace_id, last_ping", - "describe": { - "columns": [ - { - "ordinal": 0, - "name": "id", - "type_info": "Uuid" - }, - { - "ordinal": 1, - "name": "workspace_id", - "type_info": "Varchar" - }, - { - "ordinal": 2, - "name": "last_ping", - "type_info": "Timestamptz" - } - ], - "parameters": { - "Left": [ - "Text" - ] - }, - "nullable": [ - false, - false, - true - ] - }, - "hash": "84576f6add15108ade2ed3a88185d67db17904859e5f01f1d578af4e1d062a85" -} diff --git a/backend/.sqlx/query-9a9b639611459659ae355a43f219c6da2c3d1e04d49306adaaaaf06e54ee8357.json b/backend/.sqlx/query-9a9b639611459659ae355a43f219c6da2c3d1e04d49306adaaaaf06e54ee8357.json new file mode 100644 index 0000000000..8ae3d80ca6 --- /dev/null +++ b/backend/.sqlx/query-9a9b639611459659ae355a43f219c6da2c3d1e04d49306adaaaaf06e54ee8357.json @@ -0,0 +1,34 @@ +{ + "db_name": "PostgreSQL", + "query": "WITH zombie_jobs AS (\n UPDATE queue SET running = false, started_at = null\n WHERE last_ping < now() - ($1 || ' seconds')::interval\n AND running = true AND job_kind NOT IN ('flow', 'flowpreview', 'flownode', 'singlescriptflow') AND same_worker = false \n RETURNING id, workspace_id, last_ping\n ),\n update_concurrency AS (\n UPDATE concurrency_counter cc\n SET job_uuids = job_uuids - zj.id::text\n FROM zombie_jobs zj\n INNER JOIN concurrency_key ck ON ck.job_id = zj.id\n WHERE cc.concurrency_id = ck.key\n )\n SELECT id, workspace_id, last_ping FROM zombie_jobs", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "id", + "type_info": "Uuid" + }, + { + "ordinal": 1, + "name": "workspace_id", + "type_info": "Varchar" + }, + { + "ordinal": 2, + "name": "last_ping", + "type_info": "Timestamptz" + } + ], + "parameters": { + "Left": [ + "Text" + ] + }, + "nullable": [ + false, + false, + true + ] + }, + "hash": "9a9b639611459659ae355a43f219c6da2c3d1e04d49306adaaaaf06e54ee8357" +} diff --git a/backend/src/monitor.rs b/backend/src/monitor.rs index 587a25d64a..33488c6356 100644 --- a/backend/src/monitor.rs +++ b/backend/src/monitor.rs @@ -1506,9 +1506,20 @@ pub async fn reload_base_url_setting(db: &DB) -> error::Result<()> { async fn handle_zombie_jobs(db: &Pool, base_internal_url: &str, worker_name: &str) { if *RESTART_ZOMBIE_JOBS { let restarted = sqlx::query!( - "UPDATE queue SET running = false, started_at = null - WHERE last_ping < now() - ($1 || ' seconds')::interval - AND running = true AND job_kind NOT IN ('flow', 'flowpreview', 'flownode', 'singlescriptflow') AND same_worker = false RETURNING id, workspace_id, last_ping", + "WITH zombie_jobs AS ( + UPDATE queue SET running = false, started_at = null + WHERE last_ping < now() - ($1 || ' seconds')::interval + AND running = true AND job_kind NOT IN ('flow', 'flowpreview', 'flownode', 'singlescriptflow') AND same_worker = false + RETURNING id, workspace_id, last_ping + ), + update_concurrency AS ( + UPDATE concurrency_counter cc + SET job_uuids = job_uuids - zj.id::text + FROM zombie_jobs zj + INNER JOIN concurrency_key ck ON ck.job_id = zj.id + WHERE cc.concurrency_id = ck.key + ) + SELECT id, workspace_id, last_ping FROM zombie_jobs", *ZOMBIE_JOB_TIMEOUT, ) .fetch_all(db) @@ -1652,12 +1663,31 @@ async fn handle_zombie_flows(db: &DB) -> error::Result<()> { tracing::error!(error_message); report_critical_error(error_message, db.clone(), Some(&flow.workspace_id), None).await; // if the flow hasn't started and is a zombie, we can simply restart it + let mut tx = db.begin().await?; + + let concurrency_key = + sqlx::query_scalar!("SELECT key FROM concurrency_key WHERE job_id = $1", flow.id) + .fetch_optional(&mut *tx) + .await?; + + if let Some(key) = concurrency_key { + sqlx::query!( + "UPDATE concurrency_counter SET job_uuids = job_uuids - $2 WHERE concurrency_id = $1", + key, + flow.id.hyphenated().to_string() + ) + .execute(&mut *tx) + .await?; + } + sqlx::query!( "UPDATE queue SET running = false, started_at = null WHERE id = $1 AND canceled = false", flow.id ) - .execute(db) + .execute(&mut *tx) .await?; + + tx.commit().await?; } else { let id = flow.id.clone(); let last_ping = flow.last_ping.clone();