From 72ab75a149e24e08f2e4bcb18ec16fa2e52dbfc3 Mon Sep 17 00:00:00 2001
From: Ruben Fiszel <ruben@rubenfiszel.com>
Date: Wed, 19 Jun 2024 19:43:13 +0200
Subject: [PATCH] add more debug lines

---
 backend/windmill-worker/src/worker_flow.rs | 9 +++++++++
 1 file changed, 9 insertions(+)

diff --git a/backend/windmill-worker/src/worker_flow.rs b/backend/windmill-worker/src/worker_flow.rs
index 4e8d4e089b..5a950433a4 100644
--- a/backend/windmill-worker/src/worker_flow.rs
+++ b/backend/windmill-worker/src/worker_flow.rs
@@ -1986,6 +1986,7 @@ async fn push_next_flow_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>
     let mut tx: QueueTransaction<'_, R> = (rsmq.clone(), db.begin().await?).into();
 
     for i in (0..len).into_iter() {
+        tracing::debug!(id = %flow_job.id, root_id = %job_root, "pushing job {i} of {len}");
         let payload_tag = match &job_payloads {
             ContinuePayload::SingleJob(payload) => payload.clone(),
             ContinuePayload::BranchAllJobs(payloads) => payloads[i].clone(),
@@ -2085,6 +2086,8 @@ async fn push_next_flow_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>
             Err(e) => (None, Some(e)),
         };
 
+        tracing::debug!(id = %flow_job.id, root_id = %job_root, "computed args for job {i} of {len}");
+
         let value_with_parallel = module.get_value_with_parallel()?;
 
         let root_job = if {
@@ -2116,6 +2119,8 @@ async fn push_next_flow_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>
             None
         };
 
+        tracing::debug!(id = %flow_job.id, root_id = %job_root, "computed perms for job {i} of {len}");
+
         let tx2 = PushIsolationLevel::Transaction(tx);
         let (uuid, mut inner_tx) = push(
             &db,
@@ -2154,6 +2159,8 @@ async fn push_next_flow_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>
 
         if value_with_parallel.type_ == "forloopflow" {
             if let Some(p) = value_with_parallel.parallelism {
+                tracing::debug!(id = %flow_job.id, root_id = %job_root, "updating suspend for forloopflow job {uuid}");
+
                 if i as u16 >= p {
                     sqlx::query!(
                         "UPDATE queue
@@ -2165,6 +2172,7 @@ async fn push_next_flow_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>
                     .execute(&mut inner_tx)
                     .await?;
                 }
+                tracing::debug!(id = %flow_job.id, root_id = %job_root, "updated suspend for {uuid}");
             }
         }
 
@@ -2206,6 +2214,7 @@ async fn push_next_flow_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>
             )
             .execute(&mut tx)
             .await?;
+            tracing::debug!(id = %flow_job.id, root_id = %job_root, "updated parallel monitor lock for {uuid}");
         }
     }
     let first_uuid = uuids[0];