diff --git a/backend/.sqlx/query-1110504158d8a644bd8ccbace5415ffb96ad7b5d1cdedb147f4adbbd210f312b.json b/backend/.sqlx/query-1110504158d8a644bd8ccbace5415ffb96ad7b5d1cdedb147f4adbbd210f312b.json new file mode 100644 index 0000000000..34b26aaa7f --- /dev/null +++ b/backend/.sqlx/query-1110504158d8a644bd8ccbace5415ffb96ad7b5d1cdedb147f4adbbd210f312b.json @@ -0,0 +1,14 @@ +{ + "db_name": "PostgreSQL", + "query": "UPDATE queue SET suspend = suspend - 1 WHERE id = ANY($1)", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "UuidArray" + ] + }, + "nullable": [] + }, + "hash": "1110504158d8a644bd8ccbace5415ffb96ad7b5d1cdedb147f4adbbd210f312b" +} diff --git a/backend/.sqlx/query-ca2bdfa3c310b05348e20464df8403fde43dec24a8b8823383d0ac8f2116babf.json b/backend/.sqlx/query-a9da4176b6c2487bcb878dc11810b47ec58d43c6025892c57c107cd8ee6cb238.json similarity index 74% rename from backend/.sqlx/query-ca2bdfa3c310b05348e20464df8403fde43dec24a8b8823383d0ac8f2116babf.json rename to backend/.sqlx/query-a9da4176b6c2487bcb878dc11810b47ec58d43c6025892c57c107cd8ee6cb238.json index 6157317987..73a103e4fe 100644 --- a/backend/.sqlx/query-ca2bdfa3c310b05348e20464df8403fde43dec24a8b8823383d0ac8f2116babf.json +++ b/backend/.sqlx/query-a9da4176b6c2487bcb878dc11810b47ec58d43c6025892c57c107cd8ee6cb238.json @@ -1,6 +1,6 @@ { "db_name": "PostgreSQL", - "query": "SELECT id FROM queue WHERE parent_job = $1 AND suspend > 0 ORDER by id FOR UPDATE", + "query": "SELECT id FROM queue WHERE parent_job = $1 AND suspend > 0 ORDER by suspend FOR UPDATE", "describe": { "columns": [ { @@ -18,5 +18,5 @@ false ] }, - "hash": "ca2bdfa3c310b05348e20464df8403fde43dec24a8b8823383d0ac8f2116babf" + "hash": "a9da4176b6c2487bcb878dc11810b47ec58d43c6025892c57c107cd8ee6cb238" } diff --git a/backend/.sqlx/query-d42005ce0f8fbb2f65e8caa3cb06b6888b6c29e40ff575bb3893ea93e48450dc.json b/backend/.sqlx/query-d42005ce0f8fbb2f65e8caa3cb06b6888b6c29e40ff575bb3893ea93e48450dc.json deleted file mode 100644 index f2c558bd01..0000000000 --- a/backend/.sqlx/query-d42005ce0f8fbb2f65e8caa3cb06b6888b6c29e40ff575bb3893ea93e48450dc.json +++ /dev/null @@ -1,14 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "UPDATE queue SET suspend = suspend - 1 WHERE parent_job = $1 AND suspend > 0", - "describe": { - "columns": [], - "parameters": { - "Left": [ - "Uuid" - ] - }, - "nullable": [] - }, - "hash": "d42005ce0f8fbb2f65e8caa3cb06b6888b6c29e40ff575bb3893ea93e48450dc" -} diff --git a/backend/windmill-worker/src/worker_flow.rs b/backend/windmill-worker/src/worker_flow.rs index 44430fa7c5..9ec972faa5 100644 --- a/backend/windmill-worker/src/worker_flow.rs +++ b/backend/windmill-worker/src/worker_flow.rs @@ -418,9 +418,9 @@ pub async fn update_flow_status_after_job_completion_internal< let mut tx = db.begin().await?; // this ensure that the lock is taken in the same order and thus avoid deadlocks - let _ = sqlx::query!( - "SELECT id FROM queue WHERE parent_job = $1 AND suspend > 0 ORDER by id FOR UPDATE", - job_id_for_status + let ids = sqlx::query_scalar!( + "SELECT id FROM queue WHERE parent_job = $1 AND suspend > 0 ORDER by suspend FOR UPDATE", + flow ) .fetch_all(&mut *tx) .await @@ -428,14 +428,15 @@ pub async fn update_flow_status_after_job_completion_internal< Error::InternalErr(format!("error while locking jobs to decrease parallelism of: {e}")) })?; sqlx::query!( - "UPDATE queue SET suspend = suspend - 1 WHERE parent_job = $1 AND suspend > 0", - flow + "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?; } @@ -446,13 +447,18 @@ pub async fn update_flow_status_after_job_completion_internal< flow ) .execute(db) - .await?; + .await + .map_err(|e| { + Error::InternalErr(format!("error while setting last ping to null: {e}")) + })?; let r = sqlx::query_scalar!( "DELETE FROM parallel_monitor_lock WHERE parent_flow_id = $1 and job_id = $2 RETURNING last_ping", flow, job_id_for_status - ).fetch_optional(db).await?; + ).fetch_optional(db).await.map_err(|e| { + Error::InternalErr(format!("error while removing parallel_monitor_lock: {e}")) + })?; if r.is_some() { tracing::info!( "parallel flow has removed lock on its parent, last ping was {:?}", @@ -741,7 +747,10 @@ pub async fn update_flow_status_after_job_completion_internal< &_cleanup_module.flow_jobs_to_clean, ) .execute(db) - .await?; + .await + .map_err(|e| { + Error::InternalErr(format!("error while cleaning up completed_job: {e}")) + })?; } } if flow_job.canceled { diff --git a/frontend/src/lib/components/FlowStatusViewerInner.svelte b/frontend/src/lib/components/FlowStatusViewerInner.svelte index 934b9097b5..cb7f47c649 100644 --- a/frontend/src/lib/components/FlowStatusViewerInner.svelte +++ b/frontend/src/lib/components/FlowStatusViewerInner.svelte @@ -118,25 +118,6 @@ ) } - if (flowJobIds) { - let common = { - iteration_from: Math.max(flowJobIds.flowJobs.length - 20, 0), - iteration_total: flowJobIds?.length - } - let modId = flowJobIds?.moduleId ?? '' - $localDurationStatuses[modId] = { - ...($localDurationStatuses[modId] ?? { byJob: {} }), - ...common - } - let prefixed = modId - globalDurationStatuses.forEach((x) => - x.update((x) => { - x[prefixed] = { ...(x[prefixed] ?? { byJob: {} }), ...common } - return x - }) - ) - } - function updateForloop(len: number) { forloop_selected = flowJobIds?.flowJobs[len - 1] ?? '' lastSize = len @@ -236,6 +217,26 @@ timeout && clearTimeout(timeout) innerModules = [] await loadJobInProgress() + + if (flowJobIds) { + console.log('flowJobIds', flowJobIds) + let common = { + iteration_from: Math.max(flowJobIds.flowJobs.length - 20, 0), + iteration_total: flowJobIds?.length + } + let modId = flowJobIds?.moduleId ?? '' + $localDurationStatuses[modId] = { + ...($localDurationStatuses[modId] ?? { byJob: {} }), + ...common + } + let prefixed = modId + globalDurationStatuses.forEach((x) => + x.update((x) => { + x[prefixed] = { ...(x[prefixed] ?? { byJob: {} }), ...common } + return x + }) + ) + } } }