diff --git a/backend/windmill-worker/src/worker.rs b/backend/windmill-worker/src/worker.rs index e91e41f508..f51f827e90 100644 --- a/backend/windmill-worker/src/worker.rs +++ b/backend/windmill-worker/src/worker.rs @@ -483,11 +483,18 @@ async fn handle_job_error( .map(|(_, m)| m) .unwrap_or_else(|_| Map::new()); - if let Some(parent_job_id) = job.parent_job { + if job.is_flow_step || job.job_kind == JobKind::FlowPreview || job.job_kind == JobKind::Flow { + let (flow, job_status_to_update) = if let Some(parent_job_id) = job.parent_job { + (parent_job_id, job.id) + } else { + (job.id, Uuid::nil()) + }; let updated_flow = update_flow_status_after_job_completion( db, client, - &job, + flow, + &job_status_to_update, + &job.workspace_id, false, serde_json::Value::Object(m), metrics.clone(), @@ -499,19 +506,21 @@ async fn handle_job_error( ) .await; if let Err(err) = updated_flow { - if let Ok(mut tx) = db.begin().await { - if let Ok(Some(parent_job)) = - get_queued_job(parent_job_id, &job.workspace_id, &mut tx).await - { - let _ = add_completed_job_error( - db, - client, - &parent_job, - format!("Unexpected error during flow job error handling:\n{err}"), - err, - metrics, - ) - .await; + if let Some(parent_job_id) = job.parent_job { + if let Ok(mut tx) = db.begin().await { + if let Ok(Some(parent_job)) = + get_queued_job(parent_job_id, &job.workspace_id, &mut tx).await + { + let _ = add_completed_job_error( + db, + client, + &parent_job, + format!("Unexpected error during flow job error handling:\n{err}"), + err, + metrics, + ) + .await; + } } } } @@ -632,20 +641,24 @@ async fn handle_queued_job( Ok(r) => { add_completed_job(db, client, &job, true, false, r.clone(), logs).await?; if job.is_flow_step { - update_flow_status_after_job_completion( - db, - client, - &job, - true, - r, - Some(metrics.clone()), - false, - same_worker_tx.clone(), - worker_dir, - worker_config.keep_job_dir, - &worker_config.base_internal_url, - ) - .await?; + if let Some(parent_job) = job.parent_job { + update_flow_status_after_job_completion( + db, + client, + parent_job, + &job.id, + &job.workspace_id, + true, + r, + Some(metrics.clone()), + false, + same_worker_tx.clone(), + worker_dir, + worker_config.keep_job_dir, + &worker_config.base_internal_url, + ) + .await?; + } } } Err(e) => { @@ -677,20 +690,24 @@ async fn handle_queued_job( ) .await?; if job.is_flow_step { - update_flow_status_after_job_completion( - db, - client, - &job, - false, - serde_json::Value::Object(output_map), - Some(metrics), - false, - same_worker_tx, - worker_dir, - worker_config.keep_job_dir, - &worker_config.base_internal_url, - ) - .await?; + if let Some(parent_job) = job.parent_job { + update_flow_status_after_job_completion( + db, + client, + parent_job, + &job.id, + &job.workspace_id, + false, + serde_json::Value::Object(output_map), + Some(metrics), + false, + same_worker_tx, + worker_dir, + worker_config.keep_job_dir, + &worker_config.base_internal_url, + ) + .await?; + } } } }; diff --git a/backend/windmill-worker/src/worker_flow.rs b/backend/windmill-worker/src/worker_flow.rs index eecd787288..d059b7c01c 100644 --- a/backend/windmill-worker/src/worker_flow.rs +++ b/backend/windmill-worker/src/worker_flow.rs @@ -39,7 +39,9 @@ use windmill_queue::{ pub async fn update_flow_status_after_job_completion( db: &DB, client: &windmill_api_client::Client, - job: &QueuedJob, + flow: uuid::Uuid, + job_id_for_status: &Uuid, + w_id: &str, success: bool, result: serde_json::Value, metrics: Option, @@ -49,13 +51,7 @@ pub async fn update_flow_status_after_job_completion( keep_job_dir: bool, base_internal_url: &str, ) -> error::Result<()> { - tracing::debug!("UPDATE FLOW STATUS: {job:?} {success} {result:?}"); - - let w_id = &job.workspace_id; - - let flow = job - .parent_job - .ok_or_else(|| Error::InternalErr(format!("expected parent job")))?; + tracing::debug!("UPDATE FLOW STATUS: {flow:?} {success} {result:?} {w_id}"); let mut tx = db.begin().await?; @@ -131,7 +127,7 @@ pub async fn update_flow_status_after_job_completion( old_status.step + 1, FlowStatusModule::Success { id: module_status.id(), - job: job.id, + job: job_id_for_status.clone(), flow_jobs, branch_chosen, approvers: vec![], @@ -142,7 +138,7 @@ pub async fn update_flow_status_after_job_completion( old_status.step, FlowStatusModule::Failure { id: module_status.id(), - job: job.id, + job: job_id_for_status.clone(), flow_jobs, branch_chosen, }, @@ -332,11 +328,13 @@ pub async fn update_flow_status_after_job_completion( let _ = tokio::fs::remove_dir_all(format!("{worker_dir}/{}", flow_job.id)).await; } - if flow_job.parent_job.is_some() { + if let Some(parent_job) = flow_job.parent_job { return Ok(update_flow_status_after_job_completion( db, client, - &flow_job, + parent_job, + &flow, + w_id, success, result, metrics, @@ -577,11 +575,12 @@ pub async fn handle_flow( let flow = serde_json::from_value::(value)?; if flow.modules.is_empty() { - let fake_job = QueuedJob { parent_job: Some(flow_job.id), ..flow_job.clone() }; update_flow_status_after_job_completion( db, client, - &fake_job, + flow_job.id, + &Uuid::nil(), + flow_job.workspace_id.as_str(), true, serde_json::json!({}), None, diff --git a/frontend/src/lib/components/ScriptBuilder.svelte b/frontend/src/lib/components/ScriptBuilder.svelte index 16ae56c776..c0e2c40d66 100644 --- a/frontend/src/lib/components/ScriptBuilder.svelte +++ b/frontend/src/lib/components/ScriptBuilder.svelte @@ -86,7 +86,7 @@
-
+
-
+ - +
{#if testIsLoading} {:else}