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

fix: early return should consider failure_module result (#9241)

Browse Source
This commit is contained in:
hugocasa
2026-05-20 17:58:05 +02:00
committed by GitHub
parent 289549a048
commit 2db1c0a1fc
8 changed files with 266 additions and 79 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/.sqlx/query-6d992a933bb878733b7afd7a4295b9ad6f5276b60ce20e0378d6148976e02777.json → backend/.sqlx/query-04409657066c624308954958d9dd451452efc25e57769fb94b771d4879150835.json
Generated
+13 -7
View File
@@ -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"
}
backend/tests/worker.rs
+89
View File
@@ -3990,6 +3990,94 @@ async fn test_failure_module(db: Pool<Postgres>) -> anyhow::Result<()> {
Ok(())
}
#[cfg(feature = "deno_core")]
#[sqlx::test(fixtures("base"))]
async fn test_run_wait_result_early_return_with_failure_module(
db: Pool<Postgres>,
) -> 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::<serde_json::Value>(&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<Postgres>) -> anyhow::Result<()> {
@@ -4783,6 +4871,7 @@ async fn test_result_format(db: Pool<Postgres>) -> anyhow::Result<()> {
Uuid::parse_str(ordered_result_job_id).unwrap(),
"test-workspace",
None,
false,
"test-user",
)
.await
backend/windmill-api-jobs/src/execution.rs
+53 -9
View File
@@ -262,6 +262,7 @@ pub async fn run_wait_result_internal(
uuid: Uuid,
w_id: &str,
node_id_for_empty_return: Option<String>,
has_failure_module: bool,
username: &str,
) -> error::Result<(Box<RawValue>, 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<String>,
has_failure_module: bool,
username: &str,
) -> error::Result<Response> {
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<String>,
bool,
Option<sqlx::Transaction<'c, sqlx::Postgres>>,
)> {
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<TriggerMetadata>,
) -> error::Result<Response> {
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<String>,
bool,
Option<sqlx::Transaction<'c, sqlx::Postgres>>,
)> {
#[cfg(feature = "enterprise")]
backend/windmill-api/src/jobs.rs
+46 -22
View File
@@ -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<TriggerMetadata>,
) -> error::Result<(Uuid, Option<String>)> {
) -> error::Result<(Uuid, Option<String>, 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::<Uuid>()
.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<RunDependenciesRequest>,
) -> error::Result<Response> {
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<RunFlowDependenciesRequest>,
) -> error::Result<Response> {
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::<Uuid>()
.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<JobUpdateSSEStream>,
poll_delay_ms: Option<u64>,
early_return: Option<String>,
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<String> = 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<bool>,
flow_stream_job_id: Option<Uuid>,
early_return: Option<&str>,
has_failure_module: bool,
early_return_suppressed: &mut bool,
) -> error::Result<JobUpdate> {
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
backend/windmill-api/src/triggers/http/handler.rs
+2 -1
View File
@@ -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(
backend/windmill-common/src/lib.rs
+2
View File
@@ -1384,6 +1384,7 @@ pub struct FlowVersionInfo {
pub tag: Option<String>,
pub early_return: Option<String>,
pub has_preprocessor: Option<bool>,
pub has_failure_module: Option<bool>,
pub chat_input_enabled: Option<bool>,
pub on_behalf_of_email: Option<String>,
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,
backend/windmill-trigger/src/global_handler.rs
+22 -21
View File
@@ -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,
backend/windmill-trigger/src/trigger_helpers.rs
+39 -19
View File
@@ -524,6 +524,7 @@ pub async fn trigger_runnable_inner<'c>(
Uuid,
Option<i32>,
Option<String>,
bool,
Option<sqlx::Transaction<'c, sqlx::Postgres>>,
)> {
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<axum::response::Response> {
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<RawValue>, 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?,