starred/windmill
1
0
Fork 0
mirror of https://github.com/windmill-labs/windmill.git synced 2026-09-07 16:03:21 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix: restart zombie job correctly handle concurrency limits (#5181)

Browse Source 
* restart concurrency limits

* sqlx
This commit is contained in:
Ruben Fiszel
2025-01-31 00:55:51 +01:00
committed by GitHub
parent 6fa6d0f4ac
commit c22ee04657
3 changed files with 68 additions and 38 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/.sqlx/query-84576f6add15108ade2ed3a88185d67db17904859e5f01f1d578af4e1d062a85.json
Generated
-34
View File
@@ -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"
}
backend/.sqlx/query-9a9b639611459659ae355a43f219c6da2c3d1e04d49306adaaaaf06e54ee8357.json
Generated
+34
View File
@@ -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"
}
backend/src/monitor.rs
+34 -4
View File
@@ -1506,9 +1506,20 @@ pub async fn reload_base_url_setting(db: &DB) -> error::Result<()> {
async fn handle_zombie_jobs(db: &Pool<Postgres>, 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();