diff --git a/backend/windmill-worker/src/worker.rs b/backend/windmill-worker/src/worker.rs index b7a98b8cd1..3ff04cbbba 100644 --- a/backend/windmill-worker/src/worker.rs +++ b/backend/windmill-worker/src/worker.rs @@ -627,7 +627,12 @@ pub async fn drop_cache() { const OUTSTANDING_WAIT_TIME_THRESHOLD_MS: i64 = 1000; -async fn insert_wait_time(job_id: Uuid, root_job_id: Option, db: &Pool, wait_time: i64) -> sqlx::error::Result<()> { +async fn insert_wait_time( + job_id: Uuid, + root_job_id: Option, + db: &Pool, + wait_time: i64, +) -> sqlx::error::Result<()> { sqlx::query!( "INSERT INTO outstanding_wait_time(job_id, self_wait_time_ms) VALUES ($1, $2) ON CONFLICT (job_id) DO UPDATE SET self_wait_time_ms = EXCLUDED.self_wait_time_ms", @@ -682,7 +687,6 @@ fn add_outstanding_wait_time( }.in_current_span()); } - #[tracing::instrument(name = "worker", level = "info", skip_all, fields(worker = %worker_name))] pub async fn run_worker( db: &Pool, @@ -4022,6 +4026,26 @@ async fn lock_modules( } e.value = FlowModuleValue::BranchAll { branches: nbranches, parallel }.into() } + FlowModuleValue::WhileloopFlow { modules, skip_failures } => { + e.value = FlowModuleValue::WhileloopFlow { + modules: lock_modules( + modules, + job, + mem_peak, + canceled_by, + job_dir, + db, + worker_name, + worker_dir, + job_path, + base_internal_url, + token, + ) + .await?, + skip_failures, + } + .into() + } FlowModuleValue::BranchOne { branches, default } => { let mut nbranches = vec![]; for mut b in branches {