diff --git a/backend/windmill-worker/src/worker_flow.rs b/backend/windmill-worker/src/worker_flow.rs index 9ec972faa5..c2fee22e31 100644 --- a/backend/windmill-worker/src/worker_flow.rs +++ b/backend/windmill-worker/src/worker_flow.rs @@ -415,29 +415,29 @@ pub async fn update_flow_status_after_job_completion_internal< tx.commit().await?; if parallelism.is_some() { - let mut tx = db.begin().await?; - // this ensure that the lock is taken in the same order and thus avoid deadlocks let ids = sqlx::query_scalar!( - "SELECT id FROM queue WHERE parent_job = $1 AND suspend > 0 ORDER by suspend FOR UPDATE", + "SELECT id FROM queue WHERE parent_job = $1 AND suspend > 0 ORDER by suspend", flow ) - .fetch_all(&mut *tx) + .fetch_all(db) .await .map_err(|e| { Error::InternalErr(format!("error while locking jobs to decrease parallelism of: {e}")) })?; - sqlx::query!( - "UPDATE queue SET suspend = suspend - 1 WHERE id = ANY($1)", - ids.as_slice() - ) - .execute(&mut *tx) - .await - .map_err(|e| { - Error::InternalErr(format!("error decreasing suspend: {e}")) - })?; - - tx.commit().await?; + for id in ids { + sqlx::query!( + "UPDATE queue SET suspend = suspend - 1 WHERE id = $1", + id + ) + .execute(db) + .await + .map_err(|e| { + Error::InternalErr(format!( + "error decreasing suspend for {id}: {e}" + )) + })?; + } } sqlx::query!(