mirror of
https://github.com/windmill-labs/windmill.git
synced 2026-10-08 00:02:27 +00:00
fix: improve deadlocks for parallel branches with parallelism constraints
This commit is contained in:
@@ -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!(
|
||||
|
||||
Reference in New Issue
Block a user