starred/windmill
1
0
Fork 0
mirror of https://github.com/windmill-labs/windmill.git synced 2026-08-19 00:02:03 +00:00
Code Issues Packages Projects Releases Wiki Activity

add more debug lines

Browse Source
This commit is contained in:
Ruben Fiszel
2024-06-19 19:43:13 +02:00
parent 88416020b8
commit 72ab75a149
1 changed files with 9 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/windmill-worker/src/worker_flow.rs
+9
View File
@@ -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];