diff --git a/backend/src/flows.rs b/backend/src/flows.rs index 34fa328a2d..d85b719de1 100644 --- a/backend/src/flows.rs +++ b/backend/src/flows.rs @@ -78,14 +78,15 @@ pub struct FlowValue { #[derive(Deserialize, Serialize, Debug, Clone)] pub struct FlowModule { #[serde(default)] - pub input_transform: HashMap, + #[serde(alias = "input_transform")] + pub input_transforms: HashMap, pub value: FlowModuleValue, pub stop_after_if_expr: Option, pub skip_if_stopped: Option, pub summary: Option, } -#[derive(Deserialize, Serialize, Debug, Clone)] +#[derive(Deserialize, Serialize, Debug, Clone, PartialEq)] #[serde( tag = "type", rename_all(serialize = "lowercase", deserialize = "lowercase") @@ -420,14 +421,14 @@ mod tests { let fv = FlowValue { modules: vec![ FlowModule { - input_transform: hm, + input_transforms: hm, value: FlowModuleValue::Script { path: "test".to_string() }, stop_after_if_expr: None, skip_if_stopped: Some(false), summary: None, }, FlowModule { - input_transform: HashMap::new(), + input_transforms: HashMap::new(), value: FlowModuleValue::RawScript(RawCode { content: "test".to_string(), language: crate::scripts::ScriptLang::Deno, @@ -438,7 +439,7 @@ mod tests { summary: None, }, FlowModule { - input_transform: [( + input_transforms: [( "iterand".to_string(), InputTransform::Static { value: serde_json::json!(vec![1, 2, 3]) }, )] @@ -454,7 +455,7 @@ mod tests { }, ], failure_module: Some(FlowModule { - input_transform: HashMap::new(), + input_transforms: HashMap::new(), value: FlowModuleValue::Flow { path: "test".to_string() }, stop_after_if_expr: Some("previous.isEmpty()".to_string()), skip_if_stopped: None, @@ -464,4 +465,29 @@ mod tests { println!("{}", serde_json::json!(fv).to_string()); Ok(()) } + + #[test] + fn test_back_compat() { + /* renamed input_transform -> input_transforms but should deserialize old name */ + let s = r#" + { + "value": { + "type": "rawscript", + "content": "def main(n): return", + "language": "python3" + }, + "input_transform": { + "n": { + "expr": "flow_input.iter.value", + "type": "javascript" + } + } + } + "#; + let module: FlowModule = serde_json::from_str(s).unwrap(); + assert_eq!( + module.input_transforms["n"], + InputTransform::Javascript { expr: "flow_input.iter.value".to_string() } + ); + } } diff --git a/backend/src/parser_ts.rs b/backend/src/parser_ts.rs index ae47dafb6e..ad02303db8 100644 --- a/backend/src/parser_ts.rs +++ b/backend/src/parser_ts.rs @@ -11,7 +11,7 @@ use crate::{ parser::{Arg, MainArgSignature, ObjectProperty, Typ}, }; -use swc_common::{sync::Lrc, FileName, SourceMap}; +use swc_common::{sync::Lrc, FileName, SourceMap, SourceMapper, Spanned}; use swc_ecma_ast::{ AssignPat, BindingIdent, Decl, ExportDecl, Expr, FnDecl, Ident, ModuleDecl, ModuleItem, Pat, Str, TsArrayType, TsEntityName, TsKeywordType, TsKeywordTypeKind, TsLit, TsLitType, @@ -76,7 +76,9 @@ pub fn parse_deno_signature(code: &str) -> error::Result { let (name, typ, _nullable) = left.as_ident().map(binding_ident_to_arg).ok_or_else(|| { error::Error::ExecutionErr(format!( - "Arg {left:?} has unexpected syntax" + "parameter syntax unsupported: `{}`", + cm.span_to_snippet(left.span()) + .unwrap_or_else(|_| cm.span_to_string(left.span())) )) })?; Ok(Arg { @@ -92,7 +94,9 @@ pub fn parse_deno_signature(code: &str) -> error::Result { }) } _ => Err(error::Error::ExecutionErr(format!( - "Arg {x:?} has unexpected syntax" + "parameter syntax unsupported: `{}`", + cm.span_to_snippet(x.span()) + .unwrap_or_else(|_| cm.span_to_string(x.span())) ))), }) .collect::, error::Error>>()?, diff --git a/backend/src/worker.rs b/backend/src/worker.rs index acab6cbd5a..f28d731684 100644 --- a/backend/src/worker.rs +++ b/backend/src/worker.rs @@ -1252,7 +1252,7 @@ mod tests { use std::sync::Once; static ONCE: Once = Once::new(); - ONCE.call_once(|| crate::tracing_init::initialize_tracing()); + ONCE.call_once(crate::tracing_init::initialize_tracing); } /// it's important this is unique between tests as there is one prometheus registry and @@ -1285,7 +1285,7 @@ mod tests { content: numbers.to_string(), path: None, }), - input_transform: Default::default(), + input_transforms: Default::default(), stop_after_if_expr: Default::default(), skip_if_stopped: Default::default(), summary: Default::default(), @@ -1301,7 +1301,7 @@ mod tests { content: doubles.to_string(), path: None, }), - input_transform: [( + input_transforms: [( "n".to_string(), InputTransform::Javascript { expr: "previous_result.iter.value".to_string(), @@ -1316,7 +1316,7 @@ mod tests { } .into(), }, - input_transform: Default::default(), + input_transforms: Default::default(), stop_after_if_expr: Default::default(), skip_if_stopped: Default::default(), summary: Default::default(), @@ -1515,7 +1515,7 @@ def main(): "type": "rawscript", "language": "python3", "content": "def main(n): return n", - }, + } , } ], } @@ -1544,6 +1544,87 @@ def main(): assert_eq!(result, serde_json::json!(9)); } + #[sqlx::test(fixtures("base"))] + async fn test_failure_module(db: DB) { + initialize_tracing().await; + + let flow: FlowValue = serde_json::from_value(serde_json::json!({ + "modules": [{ + "input_transform": { + "l": { "type": "javascript", "expr": "[]", }, + "n": { "type": "javascript", "expr": "flow_input.n", }, + }, + "value": { + "type": "rawscript", + "language": "deno", + "content": "export function main(n, l) { if (n == 0) throw l; return { l: [...l, 0] } }", + }, + }, { + "input_transform": { + "l": { "type": "javascript", "expr": "previous_result.l", }, + "n": { "type": "javascript", "expr": "flow_input.n", }, + }, + "value": { + "type": "rawscript", + "language": "deno", + "content": "export function main(n, l) { if (n == 1) throw l; return { l: [...l, 1] } }", + }, + }, { + "input_transform": { + "l": { "type": "javascript", "expr": "previous_result.l", }, + "n": { "type": "javascript", "expr": "flow_input.n", }, + }, + "value": { + "type": "rawscript", + "language": "deno", + "content": "export function main(n, l) { if (n == 2) throw l; return { l: [...l, 2] } }", + }, + }], + "failure_module": { + "input_transform": { "error": { "type": "javascript", "expr": "previous_result", } }, + "value": { + "type": "rawscript", + "language": "deno", + "content": "export function main(error) { return { 'from failure module': error } }", + } + }, + })) + .unwrap(); + + let result = RunJob::from(JobPayload::RawFlow { value: flow.clone(), path: None }) + .arg("n", json!(0)) + .wait_until_complete(&db) + .await; + assert!(result["from failure module"]["error"] + .as_str() + .unwrap() + .contains("Uncaught (in promise) []")); + + let result = RunJob::from(JobPayload::RawFlow { value: flow.clone(), path: None }) + .arg("n", json!(1)) + .wait_until_complete(&db) + .await; + assert!(result["from failure module"]["error"] + .as_str() + .unwrap() + .contains("Uncaught (in promise) [ 0 ]")); + + let result = RunJob::from(JobPayload::RawFlow { value: flow.clone(), path: None }) + .arg("n", json!(2)) + .wait_until_complete(&db) + .await; + assert!(result["from failure module"]["error"] + .as_str() + .unwrap() + .contains("Uncaught (in promise) [ 0, 1 ]")); + + let result = RunJob::from(JobPayload::RawFlow { value: flow.clone(), path: None }) + .arg("n", json!(3)) + .wait_until_complete(&db) + .await; + assert_eq!(json!({ "l": [0, 1, 2] }), result); + } + #[sqlx::test(fixtures("base"))] async fn test_iteration(db: DB) { initialize_tracing().await; diff --git a/backend/src/worker_flow.rs b/backend/src/worker_flow.rs index d3fd596663..9f65b42869 100644 --- a/backend/src/worker_flow.rs +++ b/backend/src/worker_flow.rs @@ -98,47 +98,60 @@ pub async fn update_flow_status_after_job_completion( .await? .unwrap_or(false); - let (step_counter, new_status) = match &old_status.modules[old_status.step as usize] { - module_status @ FlowStatusModule::InProgress { - iterator: Some(Iterator { index, itered, .. }), - .. - } if *index + 1 < itered.len() && (success || skip_loop_failures) => { - (old_status.step, module_status.clone()) + let module = usize::try_from(old_status.step) + .ok() + .and_then(|i| old_status.modules.get(i)) + .unwrap_or(&old_status.failure_module); + + let (step_counter, new_status) = match module { + FlowStatusModule::InProgress { iterator: Some(Iterator { index, itered, .. }), .. } + if *index + 1 < itered.len() && (success || skip_loop_failures) => + { + (old_status.step, module.clone()) } - module_status => { - let forloop_jobs = match module_status { - FlowStatusModule::InProgress { forloop_jobs: Some(jobs), .. } => Some(jobs.clone()), + _ => { + let forloop_jobs = match module { + FlowStatusModule::InProgress { forloop_jobs, .. } => forloop_jobs.clone(), _ => None, }; - let new_status = if success || (forloop_jobs.is_some() && skip_loop_failures) { - FlowStatusModule::Success { job: job.id, forloop_jobs } + if success || (forloop_jobs.is_some() && skip_loop_failures) { + ( + old_status.step + 1, + FlowStatusModule::Success { job: job.id, forloop_jobs }, + ) } else { - FlowStatusModule::Failure { job: job.id, forloop_jobs } - }; - (old_status.step + 1, new_status) + ( + old_status.step, + FlowStatusModule::Failure { job: job.id, forloop_jobs }, + ) + } } }; - let last_step = step_counter as usize == old_status.modules.len(); + let is_last_step = usize::try_from(step_counter) + .map(|i| !(..old_status.modules.len()).contains(&i)) + .unwrap_or(true); tracing::debug!( - "old status: {:#?}\n{:#?}\n{last_step}", + "old status: {:#?}\n{:#?}\n{is_last_step}", old_status, new_status ); - let prev_step = old_status.step; let (stop_early_expr, skip_if_stop_early) = - sqlx::query_as::<_, (Option, Option)>(&format!( - "UPDATE queue - SET - flow_status = jsonb_set(jsonb_set(flow_status, '{{modules, {prev_step}}}', $1), \ - '{{\"step\"}}', $2) - WHERE id = $3 - RETURNING - (raw_flow->'modules'->{prev_step}->>'stop_after_if_expr'), - (raw_flow->'modules'->{prev_step}->>'skip_if_stopped')::bool", - )) + sqlx::query_as::<_, (Option, Option)>( + " + UPDATE queue + SET flow_status = JSONB_SET( + JSONB_SET(flow_status, ARRAY['modules', $1::TEXT], $2), + ARRAY['step'], $3) + WHERE id = $4 + RETURNING + (raw_flow->'modules'->$1->>'stop_after_if_expr'), + (raw_flow->'modules'->$1->>'skip_if_stopped')::bool + ", + ) + .bind(old_status.step) .bind(serde_json::json!(new_status)) .bind(serde_json::json!(step_counter)) .bind(flow) @@ -181,9 +194,18 @@ pub async fn update_flow_status_after_job_completion( } _ => result.clone(), }; - let done = if !(success || skip_loop_failures) || last_step || stop_early { - tx.commit().await?; + let should_continue_flow = match success { + _ if stop_early => false, + true => !is_last_step, + false if skip_loop_failures => !is_last_step, + false if has_failure_module(flow, &mut tx).await? => true, + false => false, + }; + + tx.commit().await?; + + let done = if !should_continue_flow { let logs = if stop_early { "Flow job stopped early".to_string() } else { @@ -201,8 +223,6 @@ pub async fn update_flow_status_after_job_completion( .await?; true } else { - tx.commit().await?; - match handle_flow(&flow_job, db, result.clone()).await { Err(err) => { let _ = add_completed_job_error( @@ -261,6 +281,23 @@ async fn skip_loop_failures<'c>( .map_err(|e| Error::InternalErr(format!("error during retrieval of skip_loop_failures: {e}"))) } +async fn has_failure_module<'c>( + flow: Uuid, + tx: &mut sqlx::Transaction<'c, sqlx::Postgres>, +) -> Result { + sqlx::query_scalar( + " + SELECT raw_flow->'failure_module' != 'null'::jsonb + FROM queue + WHERE id = $1 + ", + ) + .bind(flow) + .fetch_one(tx) + .await + .map_err(|e| Error::InternalErr(format!("error during retrieval of has_failure_module: {e}"))) +} + async fn compute_stop_early(expr: String, result: serde_json::Value) -> error::Result { match eval_timeout(expr, [("result".to_string(), result)].into(), None, vec![]).await? { serde_json::Value::Bool(true) => Ok(true), @@ -313,20 +350,20 @@ pub async fn get_step_of_flow_status(db: &DB, id: Uuid) -> error::Result { async fn transform_input( flow_args: &Option, last_result: serde_json::Value, - input_transform: &HashMap, + input_transforms: &HashMap, workspace: &str, token: &str, steps: Vec, ) -> anyhow::Result> { let mut mapped = serde_json::Map::new(); - for (key, val) in input_transform.into_iter() { + for (key, val) in input_transforms.into_iter() { if let InputTransform::Static { value } = val { mapped.insert(key.to_string(), value.to_owned()); } } - for (key, val) in input_transform.into_iter() { + for (key, val) in input_transforms.into_iter() { match val { InputTransform::Static { value: _ } => (), InputTransform::Javascript { expr } => { @@ -400,15 +437,17 @@ async fn push_next_flow_job( serde_json::from_value::(flow_job.flow_status.clone().unwrap_or_default()) .with_context(|| format!("parse flow status {}", flow_job.id))?; - let i = usize::try_from(status.step) + /* `mut` because reassigned on FlowStatusModule::Failure when failure_module is Some */ + + let mut i = usize::try_from(status.step) .with_context(|| format!("invalid module index {}", status.step))?; - let module: &FlowModule = flow + let mut module: &FlowModule = flow .modules .get(i) .with_context(|| format!("no module at index {}", status.step))?; - let status_module: FlowStatusModule = status + let mut status_module: FlowStatusModule = status .modules .get(i) .cloned() @@ -420,7 +459,23 @@ async fn push_next_flow_job( status_module ); - /* Don't evaluate `module.input_transform` after iteration has begun. Instead, args are + if matches!(&status_module, FlowStatusModule::Success { .. }) { + anyhow::bail!("no job for {status_module:?}") + } else if matches!(&status_module, FlowStatusModule::Failure { .. }) { + /* To run to the failure module, call push_next_flow_job with the current step on + * FlowStatusModule::Failure. This must update the step index to the end so that no + * subsequent steps are run after the failure module. */ + i = flow.modules.len(); + module = flow + .failure_module + .as_ref() + /* If this fails, it's a update_flow_status_in_progress shouldn't have called + * handle_flow to get here. */ + .context("missing failure module")?; + status_module = status.failure_module.clone(); + }; + + /* Don't evaluate `module.input_transforms` after iteration has begun. Instead, args are * carried through the Iterator by the InProgress variant. */ #[rustfmt::skip] @@ -452,7 +507,7 @@ async fn push_next_flow_job( transform_input( &flow_job.args, last_result.clone(), - &module.input_transform, + &module.input_transforms, &flow_job.workspace_id, &token, steps, @@ -494,7 +549,7 @@ async fn push_next_flow_job( UPDATE queue SET flow_status = JSONB_SET( JSONB_SET(flow_status, ARRAY['modules', $1::TEXT], $2), - '{ step }', $3) + ARRAY['step'], $3) WHERE id = $4 RETURNING * "#, @@ -551,7 +606,7 @@ async fn push_next_flow_job( .with_context(|| format!("could not iterate index {index} of {itered:?}"))?; args.extend(iterator_args); - args.insert("iter".to_string(), json!({ "index": 0, "value": next })); + args.insert("iter".to_string(), json!({ "index": index, "value": next })); Some(NextLoopStatus { index, itered, forloop_jobs }) } @@ -631,12 +686,17 @@ async fn push_next_flow_job( }; sqlx::query( - "UPDATE queue - SET flow_status = jsonb_set(flow_status, ARRAY['modules', $1::TEXT], $2) - WHERE id = $3", + " + UPDATE queue + SET flow_status = JSONB_SET( + JSONB_SET(flow_status, ARRAY['modules', $1::TEXT], $2), + ARRAY['step'], $3) + WHERE id = $4 + ", ) .bind(status.step) .bind(json!(new_status)) + .bind(json!(i)) .bind(flow_job.id) .execute(&mut tx) .await?;