diff --git a/backend/.sqlx/query-6d992a933bb878733b7afd7a4295b9ad6f5276b60ce20e0378d6148976e02777.json b/backend/.sqlx/query-04409657066c624308954958d9dd451452efc25e57769fb94b771d4879150835.json similarity index 54% rename from backend/.sqlx/query-6d992a933bb878733b7afd7a4295b9ad6f5276b60ce20e0378d6148976e02777.json rename to backend/.sqlx/query-04409657066c624308954958d9dd451452efc25e57769fb94b771d4879150835.json index e32c1e06ed..9c88e54c21 100644 --- a/backend/.sqlx/query-6d992a933bb878733b7afd7a4295b9ad6f5276b60ce20e0378d6148976e02777.json +++ b/backend/.sqlx/query-04409657066c624308954958d9dd451452efc25e57769fb94b771d4879150835.json @@ -1,6 +1,6 @@ { "db_name": "PostgreSQL", - "query": "\n SELECT\n flow_version.id AS version,\n flow_version.value->>'early_return' as early_return,\n flow_version.value->>'preprocessor_module' IS NOT NULL as has_preprocessor,\n (flow_version.value->>'chat_input_enabled')::boolean as chat_input_enabled,\n flow.tag,\n flow.dedicated_worker,\n flow.on_behalf_of_email,\n flow.edited_by,\n flow.labels\n FROM\n flow_version\n INNER JOIN flow\n ON flow.path = flow_version.path AND\n flow.workspace_id = flow_version.workspace_id\n WHERE\n flow_version.workspace_id = $1 AND\n flow_version.path = $2 AND\n flow_version.id = $3\n ", + "query": "\n SELECT\n flow_version.id AS version,\n flow_version.value->>'early_return' as early_return,\n flow_version.value->>'preprocessor_module' IS NOT NULL as has_preprocessor,\n flow_version.value->>'failure_module' IS NOT NULL as has_failure_module,\n (flow_version.value->>'chat_input_enabled')::boolean as chat_input_enabled,\n flow.tag,\n flow.dedicated_worker,\n flow.on_behalf_of_email,\n flow.edited_by,\n flow.labels\n FROM\n flow_version\n INNER JOIN flow\n ON flow.path = flow_version.path AND\n flow.workspace_id = flow_version.workspace_id\n WHERE\n flow_version.workspace_id = $1 AND\n flow_version.path = $2 AND\n flow_version.id = $3\n ", "describe": { "columns": [ { @@ -20,31 +20,36 @@ }, { "ordinal": 3, - "name": "chat_input_enabled", + "name": "has_failure_module", "type_info": "Bool" }, { "ordinal": 4, + "name": "chat_input_enabled", + "type_info": "Bool" + }, + { + "ordinal": 5, "name": "tag", "type_info": "Varchar" }, { - "ordinal": 5, + "ordinal": 6, "name": "dedicated_worker", "type_info": "Bool" }, { - "ordinal": 6, + "ordinal": 7, "name": "on_behalf_of_email", "type_info": "Text" }, { - "ordinal": 7, + "ordinal": 8, "name": "edited_by", "type_info": "Varchar" }, { - "ordinal": 8, + "ordinal": 9, "name": "labels", "type_info": "TextArray" } @@ -61,6 +66,7 @@ null, null, null, + null, true, true, true, @@ -68,5 +74,5 @@ true ] }, - "hash": "6d992a933bb878733b7afd7a4295b9ad6f5276b60ce20e0378d6148976e02777" + "hash": "04409657066c624308954958d9dd451452efc25e57769fb94b771d4879150835" } diff --git a/backend/tests/worker.rs b/backend/tests/worker.rs index 0d436701cc..f21862fd38 100644 --- a/backend/tests/worker.rs +++ b/backend/tests/worker.rs @@ -3990,6 +3990,94 @@ async fn test_failure_module(db: Pool) -> anyhow::Result<()> { Ok(()) } +#[cfg(feature = "deno_core")] +#[sqlx::test(fixtures("base"))] +async fn test_run_wait_result_early_return_with_failure_module( + db: Pool, +) -> anyhow::Result<()> { + initialize_tracing().await; + let server = ApiServer::start(db.clone()).await?; + let port = server.addr.port(); + + let flow: FlowValue = serde_json::from_value(serde_json::json!({ + "modules": [{ + "id": "a", + "value": { + "type": "rawscript", + "language": "deno", + "input_transforms": {}, + "content": "export function main() { throw new Error('boom'); }", + }, + }], + "failure_module": { + "value": { + "type": "rawscript", + "language": "deno", + "input_transforms": {}, + "content": "export function main() { return { recovered: true } }", + }, + }, + })) + .unwrap(); + + let completed = + RunJob::from(JobPayload::RawFlow { value: flow, path: None, restarted_from: None }) + .run_until_complete(&db, false, port) + .await; + + // Sanity: flow result is the failure_module's output. + assert_eq!( + json!({ "recovered": true }), + completed.json_result().unwrap() + ); + + let read_body = |response: axum::response::Response| async move { + let bytes = axum::body::to_bytes(response.into_body(), usize::MAX) + .await + .unwrap(); + serde_json::from_slice::(&bytes).unwrap() + }; + + // has_failure_module=false: legacy behavior — return the early_return node's failure. + let resp_a_only = windmill_api::jobs::run_wait_result( + &db, + completed.id, + "test-workspace", + Some("a".to_string()), + false, + "test-user", + ) + .await + .unwrap(); + let body_a_only = read_body(resp_a_only).await; + let body_a_only_str = body_a_only.to_string(); + assert!( + body_a_only_str.contains("boom"), + "expected node 'a' failure, got {body_a_only_str}", + ); + assert!( + !body_a_only_str.contains("recovered"), + "expected node 'a' failure, not failure_module output; got {body_a_only_str}", + ); + + // has_failure_module=true: skip the early_return node's failure and return the + // failure_module's recovered result instead. + let resp_with_fm = windmill_api::jobs::run_wait_result( + &db, + completed.id, + "test-workspace", + Some("a".to_string()), + true, + "test-user", + ) + .await + .unwrap(); + let body_with_fm = read_body(resp_with_fm).await; + assert_eq!(json!({ "recovered": true }), body_with_fm); + + Ok(()) +} + #[cfg(feature = "python")] #[sqlx::test(fixtures("base"))] async fn test_flow_lock_all(db: Pool) -> anyhow::Result<()> { @@ -4783,6 +4871,7 @@ async fn test_result_format(db: Pool) -> anyhow::Result<()> { Uuid::parse_str(ordered_result_job_id).unwrap(), "test-workspace", None, + false, "test-user", ) .await diff --git a/backend/windmill-api-jobs/src/execution.rs b/backend/windmill-api-jobs/src/execution.rs index e4359e19da..0581b7acd9 100644 --- a/backend/windmill-api-jobs/src/execution.rs +++ b/backend/windmill-api-jobs/src/execution.rs @@ -262,6 +262,7 @@ pub async fn run_wait_result_internal( uuid: Uuid, w_id: &str, node_id_for_empty_return: Option, + has_failure_module: bool, username: &str, ) -> error::Result<(Box, bool)> { let mut result = None; @@ -283,9 +284,15 @@ pub async fn run_wait_result_internal( let fast_poll_duration = *WAIT_RESULT_FAST_POLL_DURATION_SECS as u64 * 1000; let mut accumulated_delay = 0 as u64; + // Once we observe the early_return node failed with a failure_module configured, + // its result is final — no need to re-query it on every poll. + let mut early_return_failed_and_suppressed = false; loop { - if let Some(node_id_for_empty_return) = node_id_for_empty_return.as_ref() { + if let Some(node_id_for_empty_return) = node_id_for_empty_return + .as_ref() + .filter(|_| !early_return_failed_and_suppressed) + { let result_and_success = get_result_and_success_by_id_from_flow( &db, w_id, @@ -296,8 +303,16 @@ pub async fn run_wait_result_internal( .await .ok(); if let Some((r, s)) = result_and_success { - result = Some(r); - success = s; + // When the early_return node failed but the flow has a failure_module, + // the error handler will run and may recover. Skip this result and let + // the loop fall through to the completed flow result below, which is + // the failure_module's output. + if has_failure_module && !s { + early_return_failed_and_suppressed = true; + } else { + result = Some(r); + success = s; + } } } @@ -445,10 +460,18 @@ pub async fn run_wait_result( uuid: Uuid, w_id: &str, node_id_for_empty_return: Option, + has_failure_module: bool, username: &str, ) -> error::Result { - let (result, success) = - run_wait_result_internal(db, uuid, w_id, node_id_for_empty_return, username).await?; + let (result, success) = run_wait_result_internal( + db, + uuid, + w_id, + node_id_for_empty_return, + has_failure_module, + username, + ) + .await?; result_to_response(result, success) } @@ -624,6 +647,7 @@ pub async fn run_flow<'c>( ) -> error::Result<( Uuid, Option, + bool, Option>, )> { let FlowVersionInfo { @@ -631,6 +655,7 @@ pub async fn run_flow<'c>( tag, dedicated_worker, has_preprocessor, + has_failure_module, chat_input_enabled, on_behalf_of_email, edited_by, @@ -728,10 +753,20 @@ pub async fn run_flow<'c>( // If we were given a transaction, return it; otherwise commit it if return_tx { - Ok((uuid, early_return, Some(tx))) + Ok(( + uuid, + early_return, + has_failure_module.unwrap_or(false), + Some(tx), + )) } else { tx.commit().await?; - Ok((uuid, early_return, None)) + Ok(( + uuid, + early_return, + has_failure_module.unwrap_or(false), + None, + )) } } @@ -746,7 +781,7 @@ pub async fn run_flow_and_wait_result( args: PushArgsOwned, trigger: Option, ) -> error::Result { - let (uuid, early_return, _) = run_flow( + let (uuid, early_return, has_failure_module, _) = run_flow( authed, db, None, @@ -760,7 +795,15 @@ pub async fn run_flow_and_wait_result( ) .await?; - run_wait_result(&db, uuid, w_id, early_return, &authed.username).await + run_wait_result( + &db, + uuid, + w_id, + early_return, + has_failure_module, + &authed.username, + ) + .await } // --------------------------------------------------------------------------- @@ -780,6 +823,7 @@ pub async fn push_flow_job_by_path_into_queue<'c>( ) -> error::Result<( Uuid, Option, + bool, Option>, )> { #[cfg(feature = "enterprise")] diff --git a/backend/windmill-api/src/jobs.rs b/backend/windmill-api/src/jobs.rs index cb8d71022f..85f0aedfc3 100644 --- a/backend/windmill-api/src/jobs.rs +++ b/backend/windmill-api/src/jobs.rs @@ -3941,7 +3941,7 @@ async fn batch_rerun_handle_job( None, ) .await; - if let Ok((uuid, _, _)) = result { + if let Ok((uuid, _, _, _)) = result { return Ok(uuid.to_string()); } } @@ -4003,7 +4003,7 @@ pub async fn run_flow_by_path( ) .await?; - let (uuid, _, _) = push_flow_job_by_path_into_queue( + let (uuid, _, _, _) = push_flow_job_by_path_into_queue( authed, db, None, @@ -4037,7 +4037,7 @@ pub async fn run_flow_by_version( ) .await?; - let (uuid, _) = + let (uuid, _, _) = run_flow_by_version_inner(authed, db, user_db, w_id, version, run_query, args, None) .await?; @@ -4053,7 +4053,7 @@ pub async fn run_flow_by_version_inner( run_query: RunJobQuery, args: PushArgsOwned, trigger: Option, -) -> error::Result<(Uuid, Option)> { +) -> error::Result<(Uuid, Option, bool)> { #[cfg(feature = "enterprise")] check_license_key_valid().await?; @@ -4065,7 +4065,7 @@ pub async fn run_flow_by_version_inner( let flow_version_info = get_flow_version_info_from_version(&db, version, &w_id, &flow_path).await?; - let (uuid, early_return, _) = run_flow( + let (uuid, early_return, has_failure_module, _) = run_flow( &authed, &db, None, @@ -4079,7 +4079,7 @@ pub async fn run_flow_by_version_inner( ) .await?; - Ok((uuid, early_return)) + Ok((uuid, early_return, has_failure_module)) } #[cfg(not(feature = "enterprise"))] @@ -4983,7 +4983,7 @@ pub async fn run_wait_result_job_by_path_get( .await?; tx.commit().await?; - let wait_result = run_wait_result(&db, uuid, &w_id, None, &authed.username).await; + let wait_result = run_wait_result(&db, uuid, &w_id, None, false, &authed.username).await; handle_delete_after_completion(&db, uuid, &w_id, delete_after_use, delete_after_secs).await?; return wait_result; } @@ -5127,7 +5127,7 @@ pub async fn run_wait_result_script_by_path_internal( .await?; tx.commit().await?; - let wait_result = run_wait_result(&db, uuid, &w_id, None, &authed.username).await; + let wait_result = run_wait_result(&db, uuid, &w_id, None, false, &authed.username).await; handle_delete_after_completion(&db, uuid, &w_id, delete_after_use, delete_after_secs).await?; return wait_result; } @@ -5252,7 +5252,7 @@ pub async fn run_wait_result_script_by_hash( .await?; tx.commit().await?; - let wait_result = run_wait_result(&db, uuid, &w_id, None, &authed.username).await; + let wait_result = run_wait_result(&db, uuid, &w_id, None, false, &authed.username).await; handle_delete_after_completion(&db, uuid, &w_id, delete_after_use, delete_after_secs).await?; return wait_result; } @@ -5405,7 +5405,7 @@ pub async fn stream_job( }; let poll_delay_ms = run_query.poll_delay_ms; - let (uuid, early_return) = match runnable_id { + let (uuid, early_return, has_failure_module) = match runnable_id { RunnableId::ScriptId(ScriptId::ScriptPath(script_path)) | RunnableId::HubScript(script_path) => { let (uuid, _, _) = push_script_job_by_path_into_queue( @@ -5420,7 +5420,7 @@ pub async fn stream_job( None, ) .await?; - (uuid, None) + (uuid, None, false) } RunnableId::ScriptId(ScriptId::ScriptHash(script_hash)) => { let (uuid, _, _) = run_job_by_hash_inner( @@ -5434,10 +5434,10 @@ pub async fn stream_job( None, ) .await?; - (uuid, None) + (uuid, None, false) } RunnableId::FlowId(FlowId::FlowPath(flow_path)) => { - let (uuid, early_return, _) = push_flow_job_by_path_into_queue( + let (uuid, early_return, has_failure_module, _) = push_flow_job_by_path_into_queue( authed.clone(), db.clone(), None, @@ -5449,10 +5449,10 @@ pub async fn stream_job( None, ) .await?; - (uuid, early_return) + (uuid, early_return, has_failure_module) } RunnableId::FlowId(FlowId::FlowVersion(version)) => { - let (uuid, early_return) = run_flow_by_version_inner( + let (uuid, early_return, has_failure_module) = run_flow_by_version_inner( authed.clone(), db.clone(), user_db, @@ -5463,7 +5463,7 @@ pub async fn stream_job( None, ) .await?; - (uuid, early_return) + (uuid, early_return, has_failure_module) } }; @@ -5495,6 +5495,7 @@ pub async fn stream_job( tx, poll_delay_ms, early_return, + has_failure_module, ); let body = axum::body::Body::from_stream(stream.map(Result::<_, std::convert::Infallible>::Ok)); @@ -5978,7 +5979,7 @@ async fn run_wait_result_preview_script( let uuid = uuid .parse::() .map_err(|_| Error::BadRequest("Invalid UUID".to_string()))?; - let result = run_wait_result(&db, uuid, &w_id, None, &authed.username).await; + let result = run_wait_result(&db, uuid, &w_id, None, false, &authed.username).await; return result; } @@ -6252,7 +6253,7 @@ async fn run_dependencies_job( Json(req): Json, ) -> error::Result { let uuid = push_dependencies_job(&authed, &db, &w_id, req).await?; - run_wait_result(&db, uuid, &w_id, None, &authed.username).await + run_wait_result(&db, uuid, &w_id, None, false, &authed.username).await } async fn run_dependencies_job_async( @@ -6361,7 +6362,7 @@ async fn run_flow_dependencies_job( Json(req): Json, ) -> error::Result { let uuid = push_flow_dependencies_job(&authed, &db, &w_id, req).await?; - run_wait_result(&db, uuid, &w_id, None, &authed.username).await + run_wait_result(&db, uuid, &w_id, None, false, &authed.username).await } async fn run_flow_dependencies_job_async( @@ -6780,7 +6781,7 @@ async fn run_wait_result_preview_flow( let uuid = uuid .parse::() .map_err(|_| Error::BadRequest("Invalid UUID".to_string()))?; - let result = run_wait_result(&db, uuid, &w_id, None, &authed.username).await; + let result = run_wait_result(&db, uuid, &w_id, None, false, &authed.username).await; return result; } @@ -7214,6 +7215,8 @@ async fn get_job_update( is_flow, None, None, + false, + &mut false, ) .await?, )) @@ -7255,6 +7258,7 @@ async fn get_job_update_sse( tx, poll_delay_ms, None, + false, ); let stream = tokio_stream::wrappers::ReceiverStream::new(rx).map(|x| { @@ -7293,12 +7297,16 @@ pub fn start_job_update_sse_stream( tx: tokio::sync::mpsc::Sender, poll_delay_ms: Option, early_return: Option, + has_failure_module: bool, ) -> () { tokio::spawn(async move { let mut log_offset = initial_log_offset; let mut stream_offset = initial_stream_offset; let mut last_update_hash: Option = None; let mut flow_stream_job_id = None; + // Latched once the early_return node's failure is observed alongside a + // failure_module — subsequent polls then skip the redundant per-node lookup. + let mut early_return_suppressed = false; // Send initial update immediately let mut running = running; @@ -7321,6 +7329,8 @@ pub fn start_job_update_sse_stream( is_flow, flow_stream_job_id, early_return.as_deref(), + has_failure_module, + &mut early_return_suppressed, ) .await { @@ -7439,6 +7449,8 @@ pub fn start_job_update_sse_stream( is_flow, flow_stream_job_id, early_return.as_deref(), + has_failure_module, + &mut early_return_suppressed, ) .await { @@ -7570,6 +7582,8 @@ async fn get_job_update_data( is_flow: Option, flow_stream_job_id: Option, early_return: Option<&str>, + has_failure_module: bool, + early_return_suppressed: &mut bool, ) -> error::Result { let tags = if log_view { log_job_view( @@ -7733,11 +7747,21 @@ async fn get_job_update_data( let flow_stream_job_id = flow_stream_job_id.or(new_flow_stream_job_id); - let result = if let Some(early_return) = early_return { + let result = if let Some(early_return) = early_return.filter(|_| !*early_return_suppressed) + { match get_result_and_success_by_id_from_flow(db, w_id, job_id, early_return, None).await { + // When the early_return node failed but the flow has a failure_module, + // the error handler will run and may recover. Keep the completed flow + // result instead (it reflects the failure_module's output). Latch the + // observation so subsequent polls skip this query — the early-return + // node's failure is final once observed. + Ok((_, early_success)) if has_failure_module && !early_success => { + *early_return_suppressed = true; + result + } Ok((early_result, _)) => Some(early_result), - Err(_) => result, + _ => result, } } else { result diff --git a/backend/windmill-api/src/triggers/http/handler.rs b/backend/windmill-api/src/triggers/http/handler.rs index ffb61e31fa..0fd7717582 100644 --- a/backend/windmill-api/src/triggers/http/handler.rs +++ b/backend/windmill-api/src/triggers/http/handler.rs @@ -529,7 +529,7 @@ async fn route_job( match trigger.request_type { RequestType::SyncSse => { // Trigger the job (always async when streaming) - let (uuid, _, early_return, _) = trigger_runnable_inner( + let (uuid, _, early_return, has_failure_module, _) = trigger_runnable_inner( &db, None, Some(user_db.clone()), @@ -578,6 +578,7 @@ async fn route_job( tx, None, early_return, + has_failure_module, ); let body = axum::body::Body::from_stream( diff --git a/backend/windmill-common/src/lib.rs b/backend/windmill-common/src/lib.rs index 973a289c48..3c4244398e 100644 --- a/backend/windmill-common/src/lib.rs +++ b/backend/windmill-common/src/lib.rs @@ -1384,6 +1384,7 @@ pub struct FlowVersionInfo { pub tag: Option, pub early_return: Option, pub has_preprocessor: Option, + pub has_failure_module: Option, pub chat_input_enabled: Option, pub on_behalf_of_email: Option, pub edited_by: String, @@ -1512,6 +1513,7 @@ pub fn get_flow_version_info_from_version< flow_version.id AS version, flow_version.value->>'early_return' as early_return, flow_version.value->>'preprocessor_module' IS NOT NULL as has_preprocessor, + flow_version.value->>'failure_module' IS NOT NULL as has_failure_module, (flow_version.value->>'chat_input_enabled')::boolean as chat_input_enabled, flow.tag, flow.dedicated_worker, diff --git a/backend/windmill-trigger/src/global_handler.rs b/backend/windmill-trigger/src/global_handler.rs index def5a5457d..3fbbcddd0e 100644 --- a/backend/windmill-trigger/src/global_handler.rs +++ b/backend/windmill-trigger/src/global_handler.rs @@ -205,27 +205,28 @@ pub async fn resume_suspended_trigger_jobs( } else { // Job was created after trigger edit - delete and repush with new configuration // Pass the transaction to trigger_runnable_inner so everything is in the same transaction - let (_uuid, _delete_after_use, _early_return, tx_o) = trigger_runnable_inner( - &db, - Some(tx), - Some(user_db.clone()), - authed.clone(), - &w_id, - &trigger.script_path, - trigger.is_flow, - windmill_queue::PushArgsOwned { - extra: None, - args: job.args.map(|a| a.0).unwrap_or_default(), - }, - trigger.retry.as_ref(), - trigger.error_handler_path.as_deref(), - trigger.error_handler_args.as_ref(), - trigger_path.clone(), - None, - trigger_metadata.clone(), - None, - ) - .await?; + let (_uuid, _delete_after_use, _early_return, _has_failure_module, tx_o) = + trigger_runnable_inner( + &db, + Some(tx), + Some(user_db.clone()), + authed.clone(), + &w_id, + &trigger.script_path, + trigger.is_flow, + windmill_queue::PushArgsOwned { + extra: None, + args: job.args.map(|a| a.0).unwrap_or_default(), + }, + trigger.retry.as_ref(), + trigger.error_handler_path.as_deref(), + trigger.error_handler_args.as_ref(), + trigger_path.clone(), + None, + trigger_metadata.clone(), + None, + ) + .await?; tx = match tx_o { Some(tx) => tx, diff --git a/backend/windmill-trigger/src/trigger_helpers.rs b/backend/windmill-trigger/src/trigger_helpers.rs index b89699a169..44e5884420 100644 --- a/backend/windmill-trigger/src/trigger_helpers.rs +++ b/backend/windmill-trigger/src/trigger_helpers.rs @@ -524,6 +524,7 @@ pub async fn trigger_runnable_inner<'c>( Uuid, Option, Option, + bool, Option>, )> { let error_handler_args = error_handler_args.map(|args| { @@ -536,10 +537,10 @@ pub async fn trigger_runnable_inner<'c>( }); let user_db = user_db.unwrap_or_else(|| UserDB::new(db.clone())); - let (uuid, resolved_delete_secs, early_return, tx_out) = if is_flow { + let (uuid, resolved_delete_secs, early_return, has_failure_module, tx_out) = if is_flow { let run_query = RunJobQuery { job_id, suspended_mode, ..Default::default() }; let path = StripPath(runnable_path.to_string()); - let (uuid, early_return, tx_out) = push_flow_job_by_path_into_queue( + let (uuid, early_return, has_failure_module, tx_out) = push_flow_job_by_path_into_queue( authed, db.clone(), tx_o, @@ -551,7 +552,7 @@ pub async fn trigger_runnable_inner<'c>( Some(trigger), ) .await?; - (uuid, None, early_return, tx_out) + (uuid, None, early_return, has_failure_module, tx_out) } else { let (uuid, resolved_delete_secs, tx_out) = trigger_script_internal( db, @@ -570,10 +571,16 @@ pub async fn trigger_runnable_inner<'c>( suspended_mode, ) .await?; - (uuid, resolved_delete_secs, None, tx_out) + (uuid, resolved_delete_secs, None, false, tx_out) }; - Ok((uuid, resolved_delete_secs, early_return, tx_out)) + Ok(( + uuid, + resolved_delete_secs, + early_return, + has_failure_module, + tx_out, + )) } #[allow(dead_code)] @@ -631,7 +638,7 @@ pub async fn trigger_runnable_and_wait_for_result( trigger: TriggerMetadata, ) -> Result { let username = authed.username.clone(); - let (uuid, resolved_delete_secs, early_return, _) = trigger_runnable_inner( + let (uuid, resolved_delete_secs, early_return, has_failure_module, _) = trigger_runnable_inner( db, None, user_db, @@ -649,8 +656,15 @@ pub async fn trigger_runnable_and_wait_for_result( None, ) .await?; - let (result, success) = - run_wait_result_internal(db, uuid, &workspace_id, early_return, &username).await?; + let (result, success) = run_wait_result_internal( + db, + uuid, + &workspace_id, + early_return, + has_failure_module, + &username, + ) + .await?; match resolved_delete_secs { Some(0) => delete_job_metadata_after_use(&db, uuid).await?, @@ -677,7 +691,7 @@ pub async fn trigger_runnable_and_wait_for_raw_result( trigger: TriggerMetadata, ) -> Result<(Box, bool)> { let username = authed.username.clone(); - let (uuid, resolved_delete_secs, early_return, _) = trigger_runnable_inner( + let (uuid, resolved_delete_secs, early_return, has_failure_module, _) = trigger_runnable_inner( db, None, user_db, @@ -696,16 +710,22 @@ pub async fn trigger_runnable_and_wait_for_raw_result( ) .await?; - let (result, success) = - run_wait_result_internal(db, uuid, &workspace_id, early_return, &username) - .await - .with_context(|| { - format!( - "Error fetching job result for {} {}", - if is_flow { "flow" } else { "script" }, - runnable_path - ) - })?; + let (result, success) = run_wait_result_internal( + db, + uuid, + &workspace_id, + early_return, + has_failure_module, + &username, + ) + .await + .with_context(|| { + format!( + "Error fetching job result for {} {}", + if is_flow { "flow" } else { "script" }, + runnable_path + ) + })?; match resolved_delete_secs { Some(0) => delete_job_metadata_after_use(&db, uuid).await?,