more error messages

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"
+}

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"
 }

backend/.sqlx/query-d42005ce0f8fbb2f65e8caa3cb06b6888b6c29e40ff575bb3893ea93e48450dc.json

@@ -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"
-}

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 {

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
+					})
+				)
+			}
 		}
 	}