fix: memory optimization for flow with big args (#4019)

backend/Cargo.lock

@@ -4873,6 +4873,12 @@ dependencies = [
  "malachite-nz",
 ]
 
+[[package]]
+name = "mappable-rc"
+version = "0.1.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "204651f31b0a6a7b2128d2b92c372cd94607b210c3a6b6e542c57a8cfd4db996"
+
 [[package]]
 name = "match_cfg"
 version = "0.1.0"
@@ -10468,6 +10474,7 @@ dependencies = [
  "itertools 0.13.0",
  "jsonwebtoken",
  "lazy_static",
+ "mappable-rc",
  "mysql_async",
  "native-tls",
  "nix",

backend/Cargo.toml

@@ -214,6 +214,7 @@ convert_case = "0.6.0"
 getrandom = "0.2"
 tokio-postgres = {version = "^0.7", features = ["array-impls", "with-serde_json-1", "with-chrono-0_4", "with-uuid-1", "with-bit-vec-0_6"]}
 bit-vec = "=0.6.3"
+mappable-rc = "^0"
 mysql_async = { version = "*", default-features = false, features = ["minimal", "default", "native-tls-tls"]}
 postgres-native-tls = "^0"
 native-tls = "^0"

backend/ee-repo-ref.txt

@@ -1 +1 @@
-c7ef7778d68c86977a5d803fe98c418ef6a485c7
+ffc6e57d99f19adda60fa4baa8623413c44bb4dd

backend/tests/worker.rs

@@ -899,7 +899,7 @@ impl RunJob {
             tx,
             "test-workspace",
             payload,
-            hm_args.into(),
+            windmill_queue::PushArgs::from(&hm_args),
             /* user */ "test-user",
             /* email  */ "test@windmill.dev",
             /* permissioned_as */ "u/test-user".to_string(),

backend/windmill-api/src/apps.rs

@@ -46,7 +46,7 @@ use windmill_common::{
 };
 
 use windmill_git_sync::{handle_deployment_metadata, DeployedObject};
-use windmill_queue::{push, PushArgs, PushIsolationLevel, QueueTransaction};
+use windmill_queue::{push, PushArgs, PushArgsOwned, PushIsolationLevel, QueueTransaction};
 
 pub fn workspaced_service() -> Router {
     Router::new()
@@ -653,7 +653,7 @@ async fn create_app(
         tx,
         &w_id,
         JobPayload::AppDependencies { path: app.path.clone(), version: v_id },
-        PushArgs { args, extra: HashMap::new() },
+        PushArgs { args: &args, extra: None },
         &authed.username,
         &authed.email,
         windmill_common::users::username_to_permissioned_as(&authed.username),
@@ -944,7 +944,7 @@ async fn update_app(
         tx,
         &w_id,
         JobPayload::AppDependencies { path: npath.clone(), version: v_id },
-        PushArgs { args, extra: HashMap::new() },
+        PushArgs { args: &args, extra: None },
         &authed.username,
         &authed.email,
         windmill_common::users::username_to_permissioned_as(&authed.username),
@@ -1157,7 +1157,7 @@ async fn execute_component(
         tx,
         &w_id,
         job_payload,
-        args,
+        PushArgs { args: &args.args, extra: args.extra },
         &username,
         &email,
         permissioned_as,
@@ -1239,12 +1239,12 @@ async fn build_args(
     policy: Policy,
     component: &str,
     path: String,
-    args: HashMap<String, Box<RawValue>>,
+    mut args: HashMap<String, Box<RawValue>>,
     authed: Option<&ApiAuthed>,
     user_db: &UserDB,
     db: &DB,
     w_id: &str,
-) -> Result<(PushArgs, Option<Uuid>)> {
+) -> Result<(PushArgsOwned, Option<Uuid>)> {
     let mut job_id: Option<Uuid> = None;
     let key = format!("{}:{}", component, &path);
     let (static_inputs, one_of_inputs, allow_user_resources) = match policy {
@@ -1294,7 +1294,6 @@ async fn build_args(
         )))?,
     };
 
-    let mut args = args.clone();
     let mut safe_args = HashMap::<String, Box<RawValue>>::new();
 
     // tracing::error!("{:?}", allow_user_resources);
@@ -1413,5 +1412,8 @@ async fn build_args(
     for (k, v) in static_inputs {
         extra.insert(k.to_string(), v.to_owned());
     }
-    Ok((PushArgs { extra, args: safe_args }, job_id))
+    Ok((
+        PushArgsOwned { extra: Some(extra), args: safe_args },
+        job_id,
+    ))
 }

backend/windmill-api/src/capture.rs

@@ -18,7 +18,7 @@ use windmill_common::{
     error::{JsonResult, Result},
     utils::{not_found_if_none, StripPath},
 };
-use windmill_queue::PushArgs;
+use windmill_queue::{PushArgs, PushArgsOwned};
 
 use crate::db::{ApiAuthed, DB};
 
@@ -86,7 +86,7 @@ pub async fn new_payload(
 pub async fn update_payload(
     Extension(db): Extension<DB>,
     Path((w_id, path)): Path<(String, StripPath)>,
-    args: PushArgs,
+    args: PushArgsOwned,
 ) -> Result<StatusCode> {
     let mut tx = db.begin().await?;
 
@@ -99,7 +99,7 @@ pub async fn update_payload(
         ",
         &w_id,
         &path.to_path(),
-        Json(args) as Json<PushArgs>,
+        Json(PushArgs { args: &args.args, extra: args.extra }) as Json<PushArgs>,
     )
     .execute(&mut *tx)
     .await?;

backend/windmill-api/src/flows.rs

@@ -425,7 +425,7 @@ async fn create_flow(
             dedicated_worker: nf.dedicated_worker,
             version: version,
         },
-        args.into(),
+        windmill_queue::PushArgs { args: &args, extra: None },
         &authed.username,
         &authed.email,
         windmill_common::users::username_to_permissioned_as(&authed.username),
@@ -820,7 +820,7 @@ async fn update_flow(
             dedicated_worker: nf.dedicated_worker,
             version: version,
         },
-        windmill_queue::PushArgs { args, extra: HashMap::new() },
+        windmill_queue::PushArgs { args: &args, extra: None },
         &authed.username,
         &authed.email,
         windmill_common::users::username_to_permissioned_as(&authed.username),

backend/windmill-api/src/jobs.rs

@@ -77,7 +77,7 @@ use windmill_common::{METRICS_DEBUG_ENABLED, METRICS_ENABLED};
 use windmill_common::{get_latest_deployed_hash_for_path, BASE_URL};
 use windmill_queue::{
     cancel_job, get_queued_job, get_result_by_id_from_running_flow, job_is_complete, push,
-    DecodeQueries, PushArgs, PushIsolationLevel, QueueTransaction,
+    DecodeQueries, PushArgs, PushArgsOwned, PushIsolationLevel, QueueTransaction,
 };
 
 #[cfg(feature = "prometheus")]
@@ -588,14 +588,20 @@ async fn get_flow_job_debug_info(
                 "This endpoint is only for root flow jobs".to_string(),
             ));
         }
+        let leaf_jobs: HashMap<String, JobResult> = job
+            .leaf_jobs
+            .clone()
+            .and_then(|x| serde_json::from_value(x).ok())
+            .unwrap_or_else(HashMap::new);
         let mut jobs = HashMap::new();
-        jobs.insert("root_job".to_string(), Job::QueuedJob(job.clone()));
+        let id = job.id.clone();
+        jobs.insert("root_job".to_string(), Job::QueuedJob(job));
 
         let mut job_ids = vec![];
         let jobs_with_root = sqlx::query_scalar!(
             "SELECT id FROM queue WHERE workspace_id = $1 and root_job = $2",
             &w_id,
-            &job.id,
+            &id,
         )
         .fetch_all(&db)
         .await?;
@@ -604,10 +610,6 @@ async fn get_flow_job_debug_info(
             job_ids.push(job);
         }
 
-        let leaf_jobs: HashMap<String, JobResult> = job
-            .leaf_jobs
-            .and_then(|x| serde_json::from_value(x).ok())
-            .unwrap_or_else(HashMap::new);
         for job in leaf_jobs.iter() {
             match job.1 {
                 JobResult::ListJob(jobs) => job_ids.extend(jobs.to_owned()),
@@ -920,7 +922,7 @@ async fn get_args(
     Extension(db): Extension<DB>,
     Path((w_id, id)): Path<(String, Uuid)>,
 ) -> error::JsonResult<Box<serde_json::value::RawValue>> {
-    let record = sqlx::query(
+    let record = sqlx::query_as::<_, RawArgs>(
         "SELECT created_by, args
         FROM completed_job 
         WHERE completed_job.id = $1 AND completed_job.workspace_id = $2",
@@ -931,8 +933,6 @@ async fn get_args(
     .await?;
 
     if let Some(record) = record {
-        let record = RawArgs::from_row(&record)
-            .map_err(|e| Error::InternalErr(format!("error parsing args: {e:#}")))?;
         if opt_authed.is_none() && record.created_by != "anonymous" {
             return Err(Error::BadRequest(
                 "As a non logged in user, you can only see jobs ran by anonymous users".to_string(),
@@ -940,7 +940,7 @@ async fn get_args(
         }
         Ok(Json(record.args.map(|x| x.0).unwrap_or_default()))
     } else {
-        let record = sqlx::query(
+        let record = sqlx::query_as::<_, RawArgs>(
             "SELECT created_by, args
             FROM queue
             WHERE queue.id = $1 AND queue.workspace_id = $2",
@@ -950,8 +950,6 @@ async fn get_args(
         .fetch_optional(&db)
         .await?;
         let record = not_found_if_none(record, "Job Args", id.to_string())?;
-        let record = RawArgs::from_row(&record)
-            .map_err(|e| Error::InternalErr(format!("error parsing args: {e:#}")))?;
         if opt_authed.is_none() && record.created_by != "anonymous" {
             return Err(Error::BadRequest(
                 "As a non logged in user, you can only see jobs ran by anonymous users".to_string(),
@@ -2636,7 +2634,7 @@ pub async fn run_flow_by_path(
     Extension(rsmq): Extension<Option<rsmq_async::MultiplexedRsmq>>,
     Path((w_id, flow_path)): Path<(String, StripPath)>,
     Query(run_query): Query<RunJobQuery>,
-    args: PushArgs,
+    args: PushArgsOwned,
 ) -> error::Result<(StatusCode, String)> {
     #[cfg(feature = "enterprise")]
     check_license_key_valid().await?;
@@ -2669,7 +2667,7 @@ pub async fn run_flow_by_path(
         tx,
         &w_id,
         JobPayload::Flow { path: flow_path.to_string(), dedicated_worker },
-        args,
+        PushArgs { args: &args.args, extra: args.extra },
         authed.display_username(),
         &authed.email,
         username_to_permissioned_as(&authed.username),
@@ -2744,10 +2742,12 @@ pub async fn restart_flow(
         .with_context(|| "No flow path set for completed flow job")?;
     check_scopes(&authed, || format!("run:flow/{flow_path}"))?;
 
+    let ehm = HashMap::new();
     let push_args = completed_job
         .args
-        .map(|json| PushArgs { args: json.0, extra: HashMap::new() })
-        .unwrap_or_else(PushArgs::empty);
+        .as_ref()
+        .map(|json| PushArgs { args: &json.0, extra: None })
+        .unwrap_or_else(|| PushArgs::from(&ehm));
 
     let scheduled_for = run_query.get_scheduled_for(&db).await?;
 
@@ -2793,7 +2793,7 @@ pub async fn run_script_by_path(
     Extension(rsmq): Extension<Option<rsmq_async::MultiplexedRsmq>>,
     Path((w_id, script_path)): Path<(String, StripPath)>,
     Query(run_query): Query<RunJobQuery>,
-    args: PushArgs,
+    args: PushArgsOwned,
 ) -> error::Result<(StatusCode, String)> {
     #[cfg(feature = "enterprise")]
     check_license_key_valid().await?;
@@ -2818,7 +2818,7 @@ pub async fn run_script_by_path(
         tx,
         &w_id,
         job_payload,
-        args,
+        PushArgs { args: &args.args, extra: args.extra },
         authed.display_username(),
         &authed.email,
         username_to_permissioned_as(&authed.username),
@@ -2886,7 +2886,7 @@ pub async fn run_workflow_as_code(
     let mut extra = HashMap::new();
     extra.insert(ENTRYPOINT_OVERRIDE.to_string(), to_raw_value(&entrypoint));
 
-    let args = PushArgs { args: task.args.unwrap_or_else(HashMap::new), extra };
+    let args = PushArgs { args: &task.args.unwrap_or_else(HashMap::new), extra: Some(extra) };
     let scheduled_for = run_query.get_scheduled_for(&db).await?;
 
     let tag = run_query.tag.clone().or(tag).or(Some(job.tag));
@@ -2898,7 +2898,7 @@ pub async fn run_workflow_as_code(
         tx,
         &w_id,
         job_payload,
-        args,
+        PushArgs { args: &args.args, extra: args.extra },
         authed.display_username(),
         &authed.email,
         username_to_permissioned_as(&authed.username),
@@ -3021,15 +3021,14 @@ async fn run_wait_result(
         }
 
         if result.is_none() {
-            let row = sqlx::query(
+            let row = sqlx::query_as::<_, RawResult>(
                 "SELECT null as created_by, result, language, flow_status FROM completed_job WHERE id = $1 AND workspace_id = $2",
             )
             .bind(uuid)
             .bind(&w_id)
             .fetch_optional(db)
             .await?;
-            if let Some(row) = row {
-                let raw_result = RawResult::from_row(&row)?;
+            if let Some(raw_result) = row {
                 result = match format_result(
                     raw_result.language.as_ref(),
                     raw_result.flow_status.map(|x| x.0),
@@ -3203,7 +3202,7 @@ pub async fn run_wait_result_job_by_path_get(
     });
 
     let inner_args: HashMap<String, Box<RawValue>> = HashMap::new();
-    let args = PushArgs { extra: payload_args, args: inner_args };
+    let args = PushArgs { extra: Some(payload_args), args: &inner_args };
 
     check_queue_too_long(&db, QUEUE_LIMIT_WAIT_RESULT.or(run_query.queue_limit)).await?;
     let script_path = script_path.to_path();
@@ -3224,7 +3223,7 @@ pub async fn run_wait_result_job_by_path_get(
         tx,
         &w_id,
         job_payload,
-        args,
+        PushArgs { args: &args.args, extra: args.extra },
         authed.display_username(),
         &authed.email,
         username_to_permissioned_as(&authed.username),
@@ -3285,7 +3284,7 @@ pub async fn run_wait_result_flow_by_path_get(
         payload_args.insert(k.to_string(), v.clone());
     });
 
-    let args = PushArgs { extra: payload_args, args: HashMap::new() };
+    let args = PushArgsOwned { extra: Some(payload_args), args: HashMap::new() };
 
     run_wait_result_flow_by_path_internal(
         db, run_query, flow_path, authed, rsmq, user_db, args, w_id,
@@ -3300,7 +3299,7 @@ pub async fn run_wait_result_script_by_path(
     Extension(db): Extension<DB>,
     Path((w_id, script_path)): Path<(String, StripPath)>,
     Query(run_query): Query<RunJobQuery>,
-    args: PushArgs,
+    args: PushArgsOwned,
 ) -> error::Result<Response> {
     #[cfg(feature = "enterprise")]
     check_license_key_valid().await?;
@@ -3326,7 +3325,7 @@ async fn run_wait_result_script_by_path_internal(
     rsmq: Option<rsmq_async::MultiplexedRsmq>,
     user_db: UserDB,
     w_id: String,
-    args: PushArgs,
+    args: PushArgsOwned,
 ) -> error::Result<Response> {
     check_queue_too_long(&db, QUEUE_LIMIT_WAIT_RESULT.or(run_query.queue_limit)).await?;
     let script_path = script_path.to_path();
@@ -3347,7 +3346,7 @@ async fn run_wait_result_script_by_path_internal(
         tx,
         &w_id,
         job_payload,
-        args,
+        PushArgs { args: &args.args, extra: args.extra },
         authed.display_username(),
         &authed.email,
         username_to_permissioned_as(&authed.username),
@@ -3383,7 +3382,7 @@ pub async fn run_wait_result_script_by_hash(
     Extension(db): Extension<DB>,
     Path((w_id, script_hash)): Path<(String, ScriptHash)>,
     Query(run_query): Query<RunJobQuery>,
-    args: PushArgs,
+    args: PushArgsOwned,
 ) -> error::Result<Response> {
     #[cfg(feature = "enterprise")]
     check_license_key_valid().await?;
@@ -3431,7 +3430,7 @@ pub async fn run_wait_result_script_by_hash(
             dedicated_worker,
             priority,
         },
-        args,
+        PushArgs { args: &args.args, extra: args.extra },
         authed.display_username(),
         &authed.email,
         username_to_permissioned_as(&authed.username),
@@ -3467,7 +3466,7 @@ pub async fn run_wait_result_flow_by_path(
     Extension(db): Extension<DB>,
     Path((w_id, flow_path)): Path<(String, StripPath)>,
     Query(run_query): Query<RunJobQuery>,
-    args: PushArgs,
+    args: PushArgsOwned,
 ) -> error::Result<Response> {
     #[cfg(feature = "enterprise")]
     check_license_key_valid().await?;
@@ -3485,7 +3484,7 @@ async fn run_wait_result_flow_by_path_internal(
     authed: ApiAuthed,
     rsmq: Option<rsmq_async::MultiplexedRsmq>,
     user_db: UserDB,
-    args: PushArgs,
+    args: PushArgsOwned,
     w_id: String,
 ) -> error::Result<Response> {
     check_queue_too_long(&db, run_query.queue_limit).await?;
@@ -3509,11 +3508,11 @@ async fn run_wait_result_flow_by_path_internal(
     .fetch_optional(&mut tx)
     .await?
     .map(|x| (x.tag, x.dedicated_worker, x.early_return))
-    .ok_or_else(
-        || Error::NotFound(
-            format!("flow not found at path {flow_path} in workspace {w_id}")
-        )
-    )?;
+    .ok_or_else(|| {
+        Error::NotFound(format!(
+            "flow not found at path {flow_path} in workspace {w_id}"
+        ))
+    })?;
 
     let tag = run_query.tag.clone().or(tag);
     check_tag_available_for_workspace(&w_id, &tag).await?;
@@ -3525,7 +3524,7 @@ async fn run_wait_result_flow_by_path_internal(
         tx,
         &w_id,
         JobPayload::Flow { path: flow_path.to_string(), dedicated_worker },
-        args,
+        PushArgs { args: &args.args, extra: args.extra },
         authed.display_username(),
         &authed.email,
         username_to_permissioned_as(&authed.username),
@@ -3593,7 +3592,7 @@ async fn run_preview_script(
                 dedicated_worker: preview.dedicated_worker,
             }),
         },
-        preview.args.unwrap_or_default().into(),
+        PushArgs::from(&preview.args.unwrap_or_default()),
         authed.display_username(),
         &authed.email,
         username_to_permissioned_as(&authed.username),
@@ -3675,7 +3674,7 @@ async fn run_bundle_preview_script(
                         custom_concurrency_key: None,
                     }),
                 },
-                args.into(),
+                PushArgs::from(&args),
                 authed.display_username(),
                 &authed.email,
                 username_to_permissioned_as(&authed.username),
@@ -3789,16 +3788,17 @@ async fn run_dependencies_job(
     }
     let raw_script = req.raw_scripts[0].clone();
     let script_path = raw_script.script_path;
+    let ehm = HashMap::new();
     let (args, raw_code) = if let Some(deps) = req.raw_deps {
         let mut hm = HashMap::new();
         hm.insert(
             "raw_deps".to_string(),
             JsonRawValue::from_string("true".to_string()).unwrap(),
         );
-        (PushArgs { extra: hm, args: HashMap::new() }, deps)
+        (PushArgs { extra: Some(hm), args: &ehm }, deps)
     } else {
         (
-            PushArgs::empty(),
+            PushArgs::from(&ehm),
             raw_script.raw_code.unwrap_or_else(|| "".to_string()),
         )
     };
@@ -3869,7 +3869,10 @@ async fn run_flow_dependencies_job(
         PushIsolationLevel::IsolatedRoot(db.clone(), rsmq),
         &w_id,
         JobPayload::RawFlowDependencies { path: req.path, flow_value: req.flow_value },
-        HashMap::from([("skip_flow_update".to_string(), to_raw_value(&true))]).into(),
+        PushArgs::from(&HashMap::from([(
+            "skip_flow_update".to_string(),
+            to_raw_value(&true),
+        )])),
         authed.display_username(),
         &authed.email,
         username_to_permissioned_as(&authed.username),
@@ -3971,12 +3974,13 @@ async fn add_batch_jobs(
                 }
             };
             for _ in 0..n {
+                let ehm = HashMap::new();
                 let (uuid, ntx) = push(
                     &db,
                     tx,
                     &w_id,
                     payload.clone(),
-                    PushArgs::empty(),
+                    PushArgs::from(&ehm),
                     authed.display_username(),
                     &authed.email,
                     username_to_permissioned_as(&authed.username),
@@ -4106,7 +4110,7 @@ async fn run_preview_flow_job(
             path: raw_flow.path,
             restarted_from: raw_flow.restarted_from,
         },
-        raw_flow.args.unwrap_or_default().into(),
+        PushArgs::from(&raw_flow.args.unwrap_or_default()),
         authed.display_username(),
         &authed.email,
         username_to_permissioned_as(&authed.username),
@@ -4139,7 +4143,7 @@ pub async fn run_job_by_hash(
     Extension(rsmq): Extension<Option<rsmq_async::MultiplexedRsmq>>,
     Path((w_id, script_hash)): Path<(String, ScriptHash)>,
     Query(run_query): Query<RunJobQuery>,
-    args: PushArgs,
+    args: PushArgsOwned,
 ) -> error::Result<(StatusCode, String)> {
     #[cfg(feature = "enterprise")]
     check_license_key_valid().await?;
@@ -4185,7 +4189,7 @@ pub async fn run_job_by_hash(
             dedicated_worker,
             priority,
         },
-        args,
+        PushArgs { args: &args.args, extra: args.extra },
         authed.display_username(),
         &authed.email,
         username_to_permissioned_as(&authed.username),
@@ -4593,7 +4597,7 @@ async fn get_completed_job<'a>(
     Extension(db): Extension<DB>,
     Path((w_id, id)): Path<(String, Uuid)>,
 ) -> error::Result<Response> {
-    let job_o = sqlx::query("SELECT id, workspace_id, parent_job, created_by, created_at, duration_ms, success, script_hash, script_path, 
+    let job_o = sqlx::query_as::<_, CompletedJob>("SELECT id, workspace_id, parent_job, created_by, created_at, duration_ms, success, script_hash, script_path, 
     CASE WHEN args is null or pg_column_size(args) < 90000 THEN args ELSE '\"WINDMILL_TOO_BIG\"'::jsonb END as args, CASE WHEN result is null or pg_column_size(result) < 90000 THEN result ELSE '\"WINDMILL_TOO_BIG\"'::jsonb END as result, logs, deleted, raw_code, canceled, canceled_by, canceled_reason, job_kind, env_id,
     schedule_path, permissioned_as, flow_status, raw_flow, is_flow_step, language, started_at, is_skipped,
     raw_lock, email, visible_to_owner, mem_peak, tag, priority, result->'wm_labels' as labels FROM completed_job WHERE id = $1 AND workspace_id = $2")
@@ -4602,9 +4606,7 @@ async fn get_completed_job<'a>(
         .fetch_optional(&db)
         .await?;
 
-    let job = not_found_if_none(job_o, "Completed Job", id.to_string())?;
-
-    let cj = CompletedJob::from_row(&job)?;
+    let cj = not_found_if_none(job_o, "Completed Job", id.to_string())?;
 
     if opt_authed.is_none() && cj.created_by != "anonymous" {
         return Err(Error::BadRequest(
@@ -4649,7 +4651,7 @@ async fn get_completed_job_result(
     Query(JsonPath { json_path, suspended_job, approver, resume_id, secret }): Query<JsonPath>,
 ) -> error::Result<Response> {
     let result_o = if let Some(json_path) = json_path {
-        sqlx::query(
+        sqlx::query_as::<_, RawResult>(
             "SELECT result #> $3 as result, flow_status, language, created_by FROM completed_job WHERE id = $1 AND workspace_id = $2",
         )
         .bind(id)
@@ -4663,16 +4665,14 @@ async fn get_completed_job_result(
         .fetch_optional(&db)
         .await?
     } else {
-        sqlx::query("SELECT result, flow_status, language, created_by FROM completed_job WHERE id = $1 AND workspace_id = $2")
+        sqlx::query_as::<_, RawResult>("SELECT result, flow_status, language, created_by FROM completed_job WHERE id = $1 AND workspace_id = $2")
             .bind(id)
             .bind(&w_id)
             .fetch_optional(&db)
             .await?
     };
 
-    let result = not_found_if_none(result_o, "Completed Job", id.to_string())?;
-
-    let raw_result = RawResult::from_row(&result)?;
+    let raw_result = not_found_if_none(result_o, "Completed Job", id.to_string())?;
 
     if opt_authed.is_none() && raw_result.created_by.unwrap_or_default() != "anonymous" {
         match (suspended_job, resume_id, approver, secret) {
@@ -4740,7 +4740,7 @@ async fn get_completed_job_result_maybe(
     Path((w_id, id)): Path<(String, Uuid)>,
     Query(GetCompletedJobQuery { get_started }): Query<GetCompletedJobQuery>,
 ) -> error::Result<Response> {
-    let result_o = sqlx::query(
+    let result_o = sqlx::query_as::<_, RawResultWithSuccess>(
         "SELECT result, success, language, flow_status, created_by FROM completed_job WHERE id = $1 AND workspace_id = $2",
     )
     .bind(id)
@@ -4748,8 +4748,7 @@ async fn get_completed_job_result_maybe(
     .fetch_optional(&db)
     .await?;
 
-    if let Some(result) = result_o {
-        let res = RawResultWithSuccess::from_row(&result)?;
+    if let Some(res) = result_o {
         let result = format_result(
             res.language.as_ref(),
             res.flow_status.map(|x| x.0),
@@ -4804,7 +4803,7 @@ async fn delete_completed_job<'a>(
     let mut tx = user_db.begin(&authed).await?;
 
     require_admin(authed.is_admin, &authed.username)?;
-    let job_o = sqlx::query(
+    let job_o = sqlx::query_as::<_, CompletedJob>(
         "UPDATE completed_job SET args = null, logs = '', result = null, deleted = true WHERE id = $1 AND workspace_id = $2 \
          RETURNING *",
     )
@@ -4817,7 +4816,7 @@ async fn delete_completed_job<'a>(
         .execute(&mut *tx)
         .await?;
 
-    let job = not_found_if_none(job_o, "Completed Job", id.to_string())?;
+    let cj = not_found_if_none(job_o, "Completed Job", id.to_string())?;
 
     audit_log(
         &mut *tx,
@@ -4831,7 +4830,6 @@ async fn delete_completed_job<'a>(
     .await?;
 
     tx.commit().await?;
-    let cj = CompletedJob::from_row(&job)?;
 
     let cj = format_completed_job_result(cj);
 

backend/windmill-api/src/scripts.rs

@@ -745,7 +745,7 @@ async fn create_script_internal<'c>(
                 path: ns.path,
                 dedicated_worker: ns.dedicated_worker,
             },
-            args.into(),
+            windmill_queue::PushArgs::from(&args),
             &authed.username,
             &authed.email,
             permissioned_as,

backend/windmill-common/src/jobs.rs

@@ -113,14 +113,6 @@ pub struct QueuedJob {
 }
 
 impl QueuedJob {
-    pub fn get_args(&self) -> HashMap<String, Box<RawValue>> {
-        if let Some(args) = self.args.as_ref() {
-            args.0.clone()
-        } else {
-            HashMap::new()
-        }
-    }
-
     pub fn script_path(&self) -> &str {
         self.script_path
             .as_ref()
@@ -287,8 +279,8 @@ impl CompletedJob {
 }
 
 #[derive(sqlx::FromRow)]
-pub struct BranchResults<'a> {
-    pub result: &'a RawValue,
+pub struct BranchResults {
+    pub result: sqlx::types::Json<Box<RawValue>>,
     pub id: Uuid,
 }
 

backend/windmill-queue/src/jobs.rs

@@ -31,7 +31,7 @@ use reqwest::{
     Client, StatusCode,
 };
 use rsmq_async::RsmqConnection;
-use serde::{Deserialize, Serialize};
+use serde::{ser::SerializeMap, Deserialize, Serialize};
 use serde_json::{json, value::RawValue};
 use sqlx::{types::Json, FromRow, Pool, Postgres, Transaction};
 #[cfg(feature = "benchmark")]
@@ -137,7 +137,7 @@ pub struct CanceledBy {
 pub async fn cancel_single_job<'c>(
     username: &str,
     reason: Option<String>,
-    job_running: &QueuedJob,
+    job_running: Arc<QueuedJob>,
     w_id: &str,
     mut tx: Transaction<'c, Postgres>,
     db: &Pool<Postgres>,
@@ -146,10 +146,10 @@ pub async fn cancel_single_job<'c>(
 ) -> error::Result<(Transaction<'c, Postgres>, Option<Uuid>)> {
     if force_cancel || (job_running.parent_job.is_none() && !job_running.running) {
         let username = username.to_string();
-        let job_running = job_running.clone();
         let w_id = w_id.to_string();
         let db = db.clone();
         let rsmq = rsmq.clone();
+        let job_running = job_running.clone();
         tokio::task::spawn(async move {
             let reason: String = reason
                 .clone()
@@ -223,8 +223,8 @@ pub async fn cancel_job<'c>(
             "You are not logged in and this job was not created by an anonymous user like you so you cannot cancel it".to_string(),
         ));
     }
-    let mut job = job.unwrap();
 
+    let mut job = job.unwrap();
     if force_cancel {
         // if force canceling a flow step, make sure we force cancel from the highest parent
         loop {
@@ -240,6 +240,8 @@ pub async fn cancel_job<'c>(
         }
     }
 
+    let job = Arc::new(job);
+
     // get all children
     let mut jobs = vec![job.id];
     let mut jobs_to_cancel = vec![];
@@ -260,7 +262,7 @@ pub async fn cancel_job<'c>(
     let (ntx, _) = cancel_single_job(
         username,
         reason.clone(),
-        &job,
+        job.clone(),
         w_id,
         tx,
         db,
@@ -278,7 +280,7 @@ pub async fn cancel_job<'c>(
             let (ntx, _) = cancel_single_job(
                 username,
                 reason.clone(),
-                &job,
+                Arc::new(job),
                 w_id,
                 tx,
                 db,
@@ -417,6 +419,12 @@ impl ValidableJson for Box<RawValue> {
     }
 }
 
+impl ValidableJson for Arc<Box<RawValue>> {
+    fn is_valid_json(&self) -> bool {
+        !self.get().is_empty()
+    }
+}
+
 impl ValidableJson for serde_json::Value {
     fn is_valid_json(&self) -> bool {
         true
@@ -984,6 +992,7 @@ pub async fn add_completed_job<
                     None
                 };
 
+                let ehm = HashMap::new();
                 let (_uuid, tx) = push(
                     db,
                     tx,
@@ -1005,8 +1014,8 @@ pub async fn add_completed_job<
                     queued_job
                         .args
                         .as_ref()
-                        .map(|x| PushArgs { args: x.0.clone(), extra: HashMap::new() })
-                        .unwrap_or_else(PushArgs::empty),
+                        .map(|x| PushArgs::from(&x.0))
+                        .unwrap_or_else(|| PushArgs::from(&ehm)),
                     &queued_job.created_by,
                     &queued_job.email,
                     queued_job.permissioned_as.clone(),
@@ -1357,7 +1366,7 @@ async fn apply_schedule_handlers<
     } else {
         #[cfg(feature = "enterprise")]
         if let Some(on_recovery_path) = schedule.on_recovery.clone() {
-            let mut tx: QueueTransaction<'_, R> = (rsmq.clone(), db.begin().await?).into();
+            let tx: QueueTransaction<'_, R> = (rsmq.clone(), db.begin().await?).into();
             let times = schedule.on_recovery_times.unwrap_or(1).max(1);
             let past_jobs = sqlx::query_as::<_, CompletedJobSubset>(
                 "SELECT success, result, started_at FROM completed_job WHERE workspace_id = $1 AND schedule_path = $2 AND script_path = $3 AND id != $4 ORDER BY created_at DESC LIMIT $5",
@@ -1382,7 +1391,7 @@ async fn apply_schedule_handlers<
             let failed_job = past_jobs[past_jobs.len() - 1].clone();
 
             if !failed_job.success {
-                tx = handle_recovered_schedule(
+                handle_recovered_schedule(
                     db,
                     tx,
                     job_id,
@@ -1398,9 +1407,9 @@ async fn apply_schedule_handlers<
                     schedule.on_recovery_extra_args.clone(),
                 )
                 .await?;
+            } else {
+                tx.commit().await?;
             }
-
-            tx.commit().await?;
         }
     }
 
@@ -1480,7 +1489,7 @@ pub async fn push_error_handler<
         tx,
         handler_w_id,
         payload,
-        PushArgs { extra, args: result },
+        PushArgs { extra: Some(extra), args: &result },
         if is_global_error_handler {
             "global"
         } else if is_schedule_error_handler {
@@ -1553,7 +1562,7 @@ async fn handle_recovered_schedule<
     successful_times: i32,
     successful_job_started_at: DateTime<Utc>,
     extra_args: Option<Json<Box<RawValue>>>,
-) -> windmill_common::error::Result<QueueTransaction<'c, R>> {
+) -> windmill_common::error::Result<()> {
     let (payload, tag) = get_payload_tag_from_prefixed_path(on_recovery_path, db, w_id).await?;
 
     let mut extra = HashMap::new();
@@ -1606,7 +1615,7 @@ async fn handle_recovered_schedule<
         tx,
         w_id,
         payload,
-        PushArgs { extra: extra, args: args },
+        PushArgs { extra: Some(extra), args: &args },
         SCHEDULE_RECOVERY_HANDLER_USERNAME,
         SCHEDULE_RECOVERY_HANDLER_USER_EMAIL,
         ERROR_HANDLER_USER_GROUP.to_string(),
@@ -1631,7 +1640,8 @@ async fn handle_recovered_schedule<
         uuid,
         schedule_path
     );
-    return Ok(tx);
+    tx.commit().await?;
+    Ok(())
 }
 
 pub async fn pull<R: rsmq_async::RsmqConnection + Send + Clone>(
@@ -1930,7 +1940,7 @@ async fn pull_single_job_and_mark_as_running_no_concurrency_limit<
                     .map_err(|_| anyhow::anyhow!("Failed to parsed Redis message"))?,
             );
 
-            let m2r = sqlx::query(
+            let m2r = sqlx::query_as::<_, QueuedJob>(
                 "UPDATE queue
             SET running = true
             , started_at = coalesce(started_at, now())
@@ -1948,11 +1958,6 @@ async fn pull_single_job_and_mark_as_running_no_concurrency_limit<
             .bind(uuid)
             .fetch_optional(db)
             .await?;
-            let m2 = if let Some(row) = m2r {
-                Some(QueuedJob::from_row(&row)?)
-            } else {
-                None
-            };
 
             rsmq.delete_message(&tag.unwrap(), &msg.id)
                 .await
@@ -1961,7 +1966,7 @@ async fn pull_single_job_and_mark_as_running_no_concurrency_limit<
             #[cfg(feature = "benchmark")]
             println!("rsmq 2: {:?}", instant.elapsed());
 
-            m2
+            m2r
         } else {
             None
         }
@@ -1981,7 +1986,7 @@ async fn pull_single_job_and_mark_as_running_no_concurrency_limit<
         let priority_tags_sorted = config.priority_tags_sorted.clone();
         drop(config);
         let r = if suspend_first {
-            sqlx::query("UPDATE queue
+            sqlx::query_as::<_, QueuedJob>("UPDATE queue
             SET running = true
               , started_at = coalesce(started_at, now())
               , last_ping = now()
@@ -2007,18 +2012,13 @@ async fn pull_single_job_and_mark_as_running_no_concurrency_limit<
         } else {
             None
         };
-        let r = if let Some(row) = r {
-            Some(QueuedJob::from_row(&row)?)
-        } else {
-            None
-        };
         if r.is_none() {
             // #[cfg(feature = "benchmark")]
             // let instant = Instant::now();
             let mut highest_priority_job: Option<QueuedJob> = None;
 
             for priority_tags in priority_tags_sorted {
-                let r = sqlx::query(
+                let r = sqlx::query_as::<_, QueuedJob>(
                     "UPDATE queue
                     SET running = true
                     , started_at = coalesce(started_at, now())
@@ -2044,12 +2044,11 @@ async fn pull_single_job_and_mark_as_running_no_concurrency_limit<
                 .fetch_optional(db)
                 .await?;
 
-                if let Some(pulled_row) = r {
-                    let pulled_job = QueuedJob::from_row(&pulled_row)?;
-                    highest_priority_job = Some(pulled_job.clone());
+                if let Some(pulled_job) = r {
+                    let id = pulled_job.id;
+                    highest_priority_job = Some(pulled_job);
                     tracing::debug!(
-                        "Pulling for job {} with tags {:?} with priority {}",
-                        pulled_job.id,
+                        "Pulling for job {id} with tags {:?} with priority {}",
                         priority_tags.tags,
                         priority_tags.priority
                     );
@@ -2103,18 +2102,13 @@ async fn legacy_concurrency_key(db: &Pool<Postgres>, queued_job: &QueuedJob) ->
     .flatten()
     .flatten();
 
-    r.map(|x| {
-        interpolate_args(
-            x,
-            &queued_job
-                .args
-                .clone()
-                .map(|x| x.0)
-                .unwrap_or_default()
-                .into(),
-            &queued_job.workspace_id,
-        )
-    })
+    let ehm = HashMap::new();
+    let push_args = queued_job
+        .args
+        .as_ref()
+        .map(|x| PushArgs::from(&x.0))
+        .unwrap_or_else(|| PushArgs::from(&ehm));
+    r.map(|x| interpolate_args(x, &push_args, &queued_job.workspace_id))
 }
 
 async fn concurrency_key(
@@ -2138,7 +2132,7 @@ fn interpolate_args(x: String, args: &PushArgs, workspace_id: &str) -> String {
             let arg_value = args
                 .args
                 .get(arg_name)
-                .or(args.extra.get(arg_name))
+                .or(args.extra.as_ref().and_then(|x| x.get(arg_name)))
                 .map(|x| x.get())
                 .unwrap_or_default()
                 .trim_matches('"');
@@ -2427,7 +2421,7 @@ async fn extract_result_from_job_result(
 ) -> error::Result<Box<RawValue>> {
     match job_result {
         JobResult::ListJob(job_ids) => {
-            let rows = sqlx::query(
+            let rows = sqlx::query_as::<_, ResultWithId>(
                 "SELECT id, result FROM completed_job WHERE id = ANY($1) AND workspace_id = $2",
             )
             .bind(job_ids.as_slice())
@@ -2435,11 +2429,7 @@ async fn extract_result_from_job_result(
             .fetch_all(db)
             .await?
             .into_iter()
-            .filter_map(|x| {
-                ResultWithId::from_row(&x)
-                    .ok()
-                    .and_then(|x| x.result.map(|y| (x.id, y)))
-            })
+            .filter_map(|x| x.result.map(|y| (x.id, y)))
             .collect::<HashMap<Uuid, Json<Box<RawValue>>>>();
             let result = job_ids
                 .into_iter()
@@ -2451,7 +2441,7 @@ async fn extract_result_from_job_result(
                 .collect::<Vec<_>>();
             Ok(to_raw_value(&result))
         }
-        JobResult::SingleJob(x) => Ok(sqlx::query(
+        JobResult::SingleJob(x) => Ok(sqlx::query_as::<_, ResultR>(
             "SELECT result #> $3 as result FROM completed_job WHERE id = $1 AND workspace_id = $2",
         )
         .bind(x)
@@ -2463,11 +2453,7 @@ async fn extract_result_from_job_result(
         )
         .fetch_optional(db)
         .await?
-        .map(|r| {
-            ResultR::from_row(&r)
-                .ok()
-                .and_then(|x| x.result.map(|x| x.0))
-        })
+        .map(|r| r.result.map(|x| x.0))
         .flatten()
         .unwrap_or_else(|| to_raw_value(&serde_json::Value::Null))),
     }
@@ -2523,35 +2509,27 @@ pub async fn get_queued_job_tx<'c>(
     w_id: &str,
     tx: &mut Transaction<'c, Postgres>,
 ) -> error::Result<Option<QueuedJob>> {
-    let r = sqlx::query(
+    sqlx::query_as::<_, QueuedJob>(
         "SELECT *
             FROM queue WHERE id = $1 AND workspace_id = $2",
     )
     .bind(id)
     .bind(w_id)
     .fetch_optional(&mut **tx)
-    .await?;
-    if let Some(row) = r {
-        Ok(Some(QueuedJob::from_row(&row)?.to_owned()))
-    } else {
-        Ok(None)
-    }
+    .await
+    .map_err(Into::into)
 }
 
 pub async fn get_queued_job(id: &Uuid, w_id: &str, db: &DB) -> error::Result<Option<QueuedJob>> {
-    let r = sqlx::query(
+    sqlx::query_as::<_, QueuedJob>(
         "SELECT *
             FROM queue WHERE id = $1 AND workspace_id = $2",
     )
     .bind(id)
     .bind(w_id)
     .fetch_optional(db)
-    .await?;
-    if let Some(row) = r {
-        Ok(Some(QueuedJob::from_row(&row)?.to_owned()))
-    } else {
-        Ok(None)
-    }
+    .await
+    .map_err(Into::into)
 }
 
 pub enum PushIsolationLevel<'c, R: rsmq_async::RsmqConnection + Send + 'c> {
@@ -2586,14 +2564,48 @@ macro_rules! fetch_scalar_isolated {
 
 use sqlx::types::JsonRawValue;
 
-#[derive(Serialize, Debug)]
-pub struct PushArgs {
-    #[serde(flatten)]
-    pub extra: HashMap<String, Box<RawValue>>,
-    #[serde(flatten)]
+#[derive(Debug)]
+pub struct PushArgsOwned {
+    pub extra: Option<HashMap<String, Box<RawValue>>>,
     pub args: HashMap<String, Box<RawValue>>,
 }
 
+#[derive(Debug)]
+pub struct PushArgs<'c> {
+    pub extra: Option<HashMap<String, Box<RawValue>>>,
+    pub args: &'c HashMap<String, Box<RawValue>>,
+}
+
+impl<'c> From<&'c HashMap<String, Box<RawValue>>> for PushArgs<'c> {
+    fn from(args: &'c HashMap<String, Box<RawValue>>) -> Self {
+        PushArgs { extra: None, args }
+    }
+}
+
+impl<'c> Serialize for PushArgs<'c> {
+    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
+    where
+        S: serde::Serializer,
+    {
+        let mut map = serializer.serialize_map(Some(
+            self.args.len() + self.extra.as_ref().map(|x| x.len()).unwrap_or_default(),
+        ))?;
+        let mut in_extra = vec![];
+        if let Some(extra) = &self.extra {
+            for (k, v) in extra {
+                map.serialize_entry(k, v)?;
+                in_extra.push(k);
+            }
+        }
+        for (k, v) in self.args {
+            if !in_extra.contains(&k) {
+                map.serialize_entry(k, v)?;
+            }
+        }
+        map.end()
+    }
+}
+
 #[derive(Deserialize)]
 pub struct DecodeQuery {
     pub include_query: Option<String>,
@@ -2648,11 +2660,11 @@ impl DecodeQueries {
     }
 }
 
-impl PushArgs {
-    pub fn insert<K: Into<String>, V: Into<Box<RawValue>>>(&mut self, k: K, v: V) {
-        self.extra.insert(k.into(), v.into());
-    }
-}
+// impl<'c> PushArgs<'c> {
+//     pub fn insert<K: Into<String>, V: Into<Box<RawValue>>>(&mut self, k: K, v: V) {
+//         self.extra.insert(k.into(), v.into());
+//     }
+// }
 
 #[derive(Deserialize)]
 pub struct RequestQuery {
@@ -2687,7 +2699,7 @@ fn restructure_cloudevents_metadata(
     }
 }
 
-impl PushArgs {
+impl PushArgsOwned {
     async fn from_json(
         mut extra: HashMap<String, Box<RawValue>>,
         use_raw: bool,
@@ -2705,12 +2717,12 @@ impl PushArgs {
                 .unwrap_or_else(|| to_raw_value(&serde_json::Value::Null));
             let mut hm = HashMap::new();
             hm.insert("body".to_string(), args);
-            Ok(PushArgs { extra, args: hm })
+            Ok(PushArgsOwned { extra: Some(extra), args: hm })
         } else {
             let hm = serde_json::from_str::<Option<HashMap<String, Box<JsonRawValue>>>>(&str)
                 .map_err(|e| Error::BadRequest(format!("invalid json: {}", e)).into_response())?
                 .unwrap_or_else(HashMap::new);
-            Ok(PushArgs { extra, args: hm })
+            Ok(PushArgsOwned { extra: Some(extra), args: hm })
         }
     }
 
@@ -2727,7 +2739,7 @@ impl PushArgs {
             Error::BadRequest(format!("invalid cloudevents+json: {}", e)).into_response()
         })?;
         let hm = restructure_cloudevents_metadata(hm).map_err(|e| e.into_response())?;
-        Ok(PushArgs { extra, args: hm })
+        Ok(PushArgsOwned { extra: Some(extra), args: hm })
     }
 }
 
@@ -2771,10 +2783,10 @@ mod tests {
         "#;
         let extra = HashMap::new();
 
-        let a1 = PushArgs::from_ce_json(extra.clone(), false, r1.to_string())
+        let a1 = PushArgsOwned::from_ce_json(extra.clone(), false, r1.to_string())
             .await
             .expect("Failed to parse the cloudevent");
-        let a2 = PushArgs::from_ce_json(extra.clone(), false, r2.to_string())
+        let a2 = PushArgsOwned::from_ce_json(extra.clone(), false, r2.to_string())
             .await
             .expect("Failed to parse the cloudevent");
 
@@ -2792,7 +2804,7 @@ mod tests {
 }
 
 #[axum::async_trait]
-impl<S> FromRequest<S, axum::body::Body> for PushArgs
+impl<S> FromRequest<S, axum::body::Body> for PushArgsOwned
 where
     S: Send + Sync,
 {
@@ -2824,7 +2836,7 @@ where
             let str = String::from_utf8(bytes.to_vec())
                 .map_err(|e| Error::BadRequest(format!("invalid utf8: {}", e)).into_response())?;
 
-            PushArgs::from_json(extra, use_raw, str).await
+            PushArgsOwned::from_json(extra, use_raw, str).await
         } else if content_type
             .unwrap()
             .starts_with("application/cloudevents+json")
@@ -2835,7 +2847,7 @@ where
             let str = String::from_utf8(bytes.to_vec())
                 .map_err(|e| Error::BadRequest(format!("invalid utf8: {}", e)).into_response())?;
 
-            PushArgs::from_ce_json(extra, use_raw, str).await
+            PushArgsOwned::from_ce_json(extra, use_raw, str).await
         } else if content_type
             .unwrap()
             .starts_with("application/cloudevents-batch+json")
@@ -2868,7 +2880,7 @@ where
                 .map(|(k, v)| (k, to_raw_value(&v)))
                 .collect::<HashMap<_, _>>();
 
-            return Ok(PushArgs { extra, args: payload });
+            return Ok(PushArgsOwned { extra: Some(extra), args: payload });
         } else {
             Err(StatusCode::UNSUPPORTED_MEDIA_TYPE.into_response())
         }
@@ -2906,9 +2918,9 @@ pub fn build_extra(
     args
 }
 
-impl PushArgs {
+impl PushArgsOwned {
     pub fn empty() -> Self {
-        PushArgs { extra: HashMap::new(), args: HashMap::new() }
+        PushArgsOwned { extra: None, args: HashMap::new() }
     }
 }
 
@@ -2916,12 +2928,6 @@ pub fn empty_result() -> Box<RawValue> {
     return JsonRawValue::from_string("{}".to_string()).unwrap();
 }
 
-impl From<HashMap<String, Box<JsonRawValue>>> for PushArgs {
-    fn from(value: HashMap<String, Box<JsonRawValue>>) -> Self {
-        PushArgs { extra: HashMap::new(), args: value }
-    }
-}
-
 // impl<T> From<PushArgsInner<T>> for PushArgs<T> {
 //     fn from(value: PushArgsInner<T>) -> Self {
 //         PushArgs::Unwrapped(value)
@@ -2938,12 +2944,12 @@ struct FlowRawValue {
 }
 
 // #[instrument(level = "trace", skip_all)]
-pub async fn push<'c, R: rsmq_async::RsmqConnection + Send + 'c>(
+pub async fn push<'c, 'd, R: rsmq_async::RsmqConnection + Send + 'c>(
     _db: &Pool<Postgres>,
     mut tx: PushIsolationLevel<'c, R>,
     workspace_id: &str,
     job_payload: JobPayload,
-    args: PushArgs,
+    args: PushArgs<'d>,
     user: &str,
     mut email: &str,
     mut permissioned_as: String,
@@ -3731,6 +3737,7 @@ pub async fn push<'c, R: rsmq_async::RsmqConnection + Send + 'c>(
         .map_err(|e| Error::InternalErr(format!("Could not insert concurrency_key={concurrency_key} for job_id={job_id} script_path={script_path:?} workspace_id={workspace_id}: {e:#}")))?;
     }
 
+    tracing::debug!("Pushing job {job_id} with tag {tag}, schedule_path {schedule_path:?}, script_path: {script_path:?}, email {email}, workspace_id {workspace_id}");
     let uuid = sqlx::query_scalar!(
         "INSERT INTO queue
             (workspace_id, id, running, parent_job, created_by, permissioned_as, scheduled_for, 
@@ -3775,6 +3782,7 @@ pub async fn push<'c, R: rsmq_async::RsmqConnection + Send + 'c>(
     .await
     .map_err(|e| Error::InternalErr(format!("Could not insert into queue {job_id} with tag {tag}, schedule_path {schedule_path:?}, script_path: {script_path:?}, email {email}, workspace_id {workspace_id}: {e:#}")))?;
 
+    tracing::debug!("Pushed {job_id}");
     // TODO: technically the job isn't queued yet, as the transaction can be rolled back. Should be solved when moving these metrics to the queue abstraction.
     #[cfg(feature = "prometheus")]
     if METRICS_ENABLED.load(std::sync::atomic::Ordering::Relaxed) {

backend/windmill-queue/src/schedule.rs

@@ -208,7 +208,7 @@ pub async fn push_scheduled_job<'c, R: rsmq_async::RsmqConnection + Send + 'c>(
         tx,
         &schedule.workspace_id,
         payload,
-        crate::PushArgs { args, extra: HashMap::new() },
+        crate::PushArgs { args: &args, extra: None },
         &schedule_to_user(&schedule.path),
         &schedule.email,
         username_to_permissioned_as(&schedule.edited_by),
@@ -228,6 +228,7 @@ pub async fn push_scheduled_job<'c, R: rsmq_async::RsmqConnection + Send + 'c>(
         authed,
     )
     .await?;
+
     Ok(tx) // TODO: Bubble up pushed UUID from here
 }
 

backend/windmill-worker/Cargo.toml

@@ -49,6 +49,7 @@ chrono.workspace = true
 dotenv.workspace = true
 rand.workspace = true # TODO: Remove. only used by token creation hack.
 const_format.workspace = true
+mappable-rc.workspace = true
 git-version.workspace = true
 dyn-iter.workspace = true
 once_cell.workspace = true

backend/windmill-worker/src/bun_executor.rs

@@ -894,8 +894,6 @@ pub async fn get_common_bun_proc_envs(base_internal_url: &str) -> HashMap<String
 
 #[cfg(feature = "enterprise")]
 use crate::{dedicated_worker::handle_dedicated_process, JobCompletedSender};
-#[cfg(feature = "enterprise")]
-use std::sync::Arc;
 
 #[cfg(feature = "enterprise")]
 pub async fn start_worker(
@@ -911,7 +909,7 @@ pub async fn start_worker(
     script_path: &str,
     token: &str,
     job_completed_tx: JobCompletedSender,
-    jobs_rx: Receiver<Arc<QueuedJob>>,
+    jobs_rx: Receiver<std::sync::Arc<QueuedJob>>,
     killpill_rx: tokio::sync::broadcast::Receiver<()>,
 ) -> Result<()> {
     let mut logs = "".to_string();

backend/windmill-worker/src/dedicated_worker.rs

@@ -70,13 +70,14 @@ pub async fn handle_dedicated_process(
     mut killpill_rx: tokio::sync::broadcast::Receiver<()>,
     job_completed_tx: JobCompletedSender,
     token: &str,
-    mut jobs_rx: Receiver<Arc<QueuedJob>>,
+    mut jobs_rx: Receiver<std::sync::Arc<QueuedJob>>,
     worker_name: &str,
     db: &DB,
     script_path: &str,
     mode: &str,
 ) -> std::result::Result<(), error::Error> {
     //do not cache local dependencies
+
     let mut child = {
         let mut cmd = Command::new(command_path);
         cmd.current_dir(job_dir)
@@ -130,7 +131,7 @@ pub async fn handle_dedicated_process(
         }
     });
 
-    let mut jobs = VecDeque::with_capacity(MAX_BUFFERED_DEDICATED_JOBS);
+    let mut jobs: VecDeque<Arc<QueuedJob>> = VecDeque::with_capacity(MAX_BUFFERED_DEDICATED_JOBS);
     // let mut i = 0;
     // let mut j = 0;
     let mut alive = true;
@@ -173,6 +174,7 @@ pub async fn handle_dedicated_process(
                         tracing::info!("job completed on dedicated worker {script_path}: {}", job.id);
                         match serde_json::from_str::<Box<serde_json::value::RawValue>>(&line.replace("wm_res[success]:", "").replace("wm_res[error]:", "")) {
                             Ok(result) => {
+                                let result = Arc::new(result);
                                 append_logs(&job.id, &job.workspace_id,  logs.clone(), db).await;
                                 if line.starts_with("wm_res[success]:") {
                                     job_completed_tx.send(JobCompleted { job , result, mem_peak: 0, canceled_by: None, success: true, cached_res_path: None, token: token.to_string() }).await.unwrap()
@@ -182,7 +184,7 @@ pub async fn handle_dedicated_process(
                             },
                             Err(e) => {
                                 tracing::error!("Could not deserialize job result `{line}`: {e:?}");
-                                job_completed_tx.send(JobCompleted { job , result: to_raw_value(&serde_json::json!({"error": format!("Could not deserialize job result `{line}`: {e:?}")})),  mem_peak: 0, canceled_by: None, success: false, cached_res_path: None, token: token.to_string() }).await.unwrap();
+                                job_completed_tx.send(JobCompleted { job , result: Arc::new(to_raw_value(&serde_json::json!({"error": format!("Could not deserialize job result `{line}`: {e:?}")}))),  mem_peak: 0, canceled_by: None, success: false, cached_res_path: None, token: token.to_string() }).await.unwrap();
                             },
                         };
                         logs = init_log.clone();
@@ -222,7 +224,11 @@ pub async fn handle_dedicated_process(
     Ok(())
 }
 
-type DedicatedWorker = (String, Sender<Arc<QueuedJob>>, Option<JoinHandle<()>>);
+type DedicatedWorker = (
+    String,
+    Sender<std::sync::Arc<QueuedJob>>,
+    Option<JoinHandle<()>>,
+);
 
 // spawn one dedicated worker per compatible steps of the flow, associating the node id to the dedicated worker channel send
 #[async_recursion]
@@ -240,7 +246,8 @@ async fn spawn_dedicated_workers_for_flow(
     job_completed_tx: &JobCompletedSender,
 ) -> Vec<DedicatedWorker> {
     let mut workers = vec![];
-    let mut script_path_to_worker: HashMap<String, Sender<Arc<QueuedJob>>> = HashMap::new();
+    let mut script_path_to_worker: HashMap<String, Sender<std::sync::Arc<QueuedJob>>> =
+        HashMap::new();
     for module in modules.iter() {
         let value = module.get_value();
         if let Ok(value) = value {
@@ -389,7 +396,7 @@ pub async fn create_dedicated_worker_map(
     worker_name: &str,
     job_completed_tx: &JobCompletedSender,
 ) -> (
-    HashMap<String, Sender<Arc<QueuedJob>>>,
+    HashMap<String, Sender<std::sync::Arc<QueuedJob>>>,
     bool,
     Vec<JoinHandle<()>>,
 ) {
@@ -519,7 +526,7 @@ async fn spawn_dedicated_worker(
     #[cfg(feature = "enterprise")]
     {
         let (dedicated_worker_tx, dedicated_worker_rx) =
-            tokio::sync::mpsc::channel::<Arc<QueuedJob>>(MAX_BUFFERED_DEDICATED_JOBS);
+            tokio::sync::mpsc::channel::<std::sync::Arc<QueuedJob>>(MAX_BUFFERED_DEDICATED_JOBS);
         let killpill_rx = killpill_rx.resubscribe();
         let db = db.clone();
         let base_internal_url = base_internal_url.to_string();

backend/windmill-worker/src/deno_executor.rs

@@ -394,8 +394,7 @@ async fn build_import_map(
 
 #[cfg(feature = "enterprise")]
 use crate::{dedicated_worker::handle_dedicated_process, JobCompletedSender};
-#[cfg(feature = "enterprise")]
-use std::sync::Arc;
+
 #[cfg(feature = "enterprise")]
 use tokio::sync::mpsc::Receiver;
 
@@ -410,7 +409,7 @@ pub async fn start_worker(
     script_path: &str,
     token: &str,
     job_completed_tx: JobCompletedSender,
-    jobs_rx: Receiver<Arc<QueuedJob>>,
+    jobs_rx: Receiver<std::sync::Arc<QueuedJob>>,
     killpill_rx: tokio::sync::broadcast::Receiver<()>,
     db: &sqlx::Pool<sqlx::Postgres>,
 ) -> Result<()> {

backend/windmill-worker/src/js_eval.rs

@@ -138,7 +138,7 @@ pub struct OptAuthedClient(Option<AuthedClient>);
 pub async fn eval_timeout(
     expr: String,
     transform_context: HashMap<String, Arc<Box<RawValue>>>,
-    flow_input: Option<Arc<HashMap<String, Box<RawValue>>>>,
+    flow_input: Option<mappable_rc::Marc<HashMap<String, Box<RawValue>>>>,
     authed_client: Option<&AuthedClient>,
     by_id: Option<IdContext>,
 ) -> anyhow::Result<Box<RawValue>> {
@@ -563,7 +563,7 @@ async fn op_resource(
 
 pub struct TransformContext {
     pub envs: HashMap<String, Arc<Box<RawValue>>>,
-    pub flow_input: Option<Arc<HashMap<String, Box<RawValue>>>>,
+    pub flow_input: Option<mappable_rc::Marc<HashMap<String, Box<RawValue>>>>,
 }
 
 #[op2]
@@ -575,7 +575,7 @@ fn op_get_context(op_state: Rc<RefCell<OpState>>, #[string] id: &str) -> String
         client
             .flow_input
             .as_ref()
-            .and_then(|x| serde_json::to_string(&x).ok())
+            .and_then(|x| serde_json::to_string(x.as_ref()).ok())
             .unwrap_or_else(|| "null".to_string())
     } else {
         client

backend/windmill-worker/src/python_executor.rs

@@ -1007,9 +1007,6 @@ pub async fn handle_python_reqs(
     Ok(req_paths)
 }
 
-#[cfg(feature = "enterprise")]
-use std::sync::Arc;
-
 #[cfg(feature = "enterprise")]
 use crate::JobCompletedSender;
 #[cfg(feature = "enterprise")]
@@ -1032,7 +1029,7 @@ pub async fn start_worker(
     script_path: &str,
     token: &str,
     job_completed_tx: JobCompletedSender,
-    jobs_rx: Receiver<Arc<QueuedJob>>,
+    jobs_rx: Receiver<std::sync::Arc<QueuedJob>>,
     killpill_rx: tokio::sync::broadcast::Receiver<()>,
 ) -> error::Result<()> {
     let mut mem_peak: i32 = 0;

backend/windmill-worker/src/worker.rs

@@ -1344,7 +1344,7 @@ pub async fn run_worker<R: rsmq_async::RsmqConnection + Send + Sync + Clone + 's
                         &Uuid::nil(),
                         &w_id,
                         success,
-                        &result,
+                        Arc::new(result),
                         true,
                         same_worker_tx2.clone(),
                         &worker_dir,
@@ -1676,7 +1676,7 @@ pub async fn run_worker<R: rsmq_async::RsmqConnection + Send + Sync + Clone + 's
                         .send(JobCompleted {
                             job: Arc::new(job),
                             success: true,
-                            result: empty_result(),
+                            result: Arc::new(empty_result()),
                             mem_peak: 0,
                             cached_res_path: None,
                             token: "".to_string(),
@@ -1936,6 +1936,7 @@ async fn queue_init_bash_maybe<'c, R: rsmq_async::RsmqConnection + Send + 'c>(
 ) -> error::Result<()> {
     if let Some(content) = WORKER_CONFIG.read().await.init_bash.clone() {
         let tx = PushIsolationLevel::IsolatedRoot(db.clone(), rsmq);
+        let ehm = HashMap::new();
         let (uuid, inner_tx) = push(
             &db,
             tx,
@@ -1952,7 +1953,7 @@ async fn queue_init_bash_maybe<'c, R: rsmq_async::RsmqConnection + Send + 'c>(
                 cache_ttl: None,
                 dedicated_worker: None,
             }),
-            PushArgs::empty(),
+            PushArgs::from(&ehm),
             worker_name,
             "worker@windmill.dev",
             SUPERADMIN_SECRET_EMAIL.to_string(),
@@ -2017,6 +2018,10 @@ pub async fn process_completed_job<R: rsmq_async::RsmqConnection + Send + Sync +
             save_in_cache(db, client, &job, cached_path.to_string(), &result).await;
         }
 
+        let is_flow_step = job.is_flow_step;
+        let parent_job = job.parent_job.clone();
+        let job_id = job.id.clone();
+        let workspace_id = job.workspace_id.clone();
         #[cfg(feature = "prometheus")]
         let timer = _worker_save_completed_job_duration
             .as_ref()
@@ -2033,25 +2038,26 @@ pub async fn process_completed_job<R: rsmq_async::RsmqConnection + Send + Sync +
             false,
         )
         .await?;
+        drop(job);
 
         #[cfg(feature = "prometheus")]
         timer.map(|x| x.stop_and_record());
 
-        if job.is_flow_step {
-            if let Some(parent_job) = job.parent_job {
+        if is_flow_step {
+            if let Some(parent_job) = parent_job {
                 #[cfg(feature = "prometheus")]
                 let timer = _worker_flow_transition_duration
                     .as_ref()
                     .map(|x| x.start_timer());
-                tracing::info!(parent_flow = %parent_job, subflow = %job.id, "updating flow status (2)");
+                tracing::info!(parent_flow = %parent_job, subflow = %job_id, "updating flow status (2)");
                 update_flow_status_after_job_completion(
                     db,
                     client,
                     parent_job,
-                    &job.id,
-                    &job.workspace_id,
+                    &job_id,
+                    &workspace_id,
                     true,
-                    &result,
+                    result,
                     false,
                     same_worker_tx.clone(),
                     &worker_dir,
@@ -2089,7 +2095,7 @@ pub async fn process_completed_job<R: rsmq_async::RsmqConnection + Send + Sync +
                     &job.id,
                     &job.workspace_id,
                     false,
-                    &serde_json::value::to_raw_value(&result).unwrap(),
+                    Arc::new(serde_json::value::to_raw_value(&result).unwrap()),
                     false,
                     same_worker_tx,
                     &worker_dir,
@@ -2205,7 +2211,7 @@ pub async fn handle_job_error<R: rsmq_async::RsmqConnection + Send + Sync + Clon
             &job_status_to_update,
             &job.workspace_id,
             false,
-            &serde_json::value::to_raw_value(&wrapped_error).unwrap(),
+            Arc::new(serde_json::value::to_raw_value(&wrapped_error).unwrap()),
             unrecoverable,
             same_worker_tx,
             worker_dir,
@@ -2291,7 +2297,7 @@ pub enum SendResult {
 #[derive(Debug, Clone)]
 pub struct JobCompleted {
     pub job: Arc<QueuedJob>,
-    pub result: Box<RawValue>,
+    pub result: Arc<Box<RawValue>>,
     pub mem_peak: i32,
     pub success: bool,
     pub cached_res_path: Option<String>,
@@ -2458,7 +2464,7 @@ async fn handle_queued_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>(
             job_completed_tx
                 .send(JobCompleted {
                     job: job,
-                    result: cached_resource_value,
+                    result: Arc::new(cached_resource_value),
                     mem_peak: 0,
                     canceled_by: None,
                     success: true,
@@ -2475,7 +2481,7 @@ async fn handle_queued_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>(
         #[cfg(feature = "prometheus")]
         let timer = _worker_flow_initial_transition_duration.map(|x| x.start_timer());
         handle_flow(
-            &job,
+            job,
             db,
             &client.get_authed().await,
             None,
@@ -2615,7 +2621,7 @@ async fn handle_queued_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>(
 
         process_result(
             job,
-            result,
+            result.map(|x| Arc::new(x)),
             job_dir,
             job_completed_tx,
             mem_peak,
@@ -2632,7 +2638,7 @@ async fn handle_queued_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>(
 
 async fn process_result(
     job: Arc<QueuedJob>,
-    result: error::Result<Box<RawValue>>,
+    result: error::Result<Arc<Box<RawValue>>>,
     job_dir: &str,
     job_completed_tx: JobCompletedSender,
     mem_peak: i32,
@@ -2707,7 +2713,7 @@ async fn process_result(
             job_completed_tx
                 .send(JobCompleted {
                     job: job,
-                    result: to_raw_value(&error_value),
+                    result: Arc::new(to_raw_value(&error_value)),
                     mem_peak,
                     canceled_by,
                     success: false,

backend/windmill-worker/src/worker_flow.rs

@@ -17,6 +17,7 @@ use crate::common::{hash_args, save_in_cache};
 use crate::js_eval::{eval_timeout, IdContext};
 use crate::{AuthedClient, PreviousResult, SameWorkerPayload, SendResult, JOB_TOKEN, KEEP_JOB_DIR};
 use anyhow::Context;
+use mappable_rc::Marc;
 use serde::{Deserialize, Serialize};
 use serde_json::value::RawValue;
 use serde_json::{json, Value};
@@ -47,7 +48,7 @@ use windmill_common::{
 use windmill_queue::schedule::get_schedule_opt;
 use windmill_queue::{
     add_completed_job, add_completed_job_error, append_logs, get_queued_job,
-    handle_maybe_scheduled_job, CanceledBy, PushIsolationLevel, WrappedError,
+    handle_maybe_scheduled_job, CanceledBy, PushArgs, PushIsolationLevel, WrappedError,
 };
 
 type DB = sqlx::Pool<sqlx::Postgres>;
@@ -56,7 +57,6 @@ use windmill_queue::{canceled_job_to_result, get_queued_job_tx, push, QueueTrans
 
 // #[instrument(level = "trace", skip_all)]
 pub async fn update_flow_status_after_job_completion<
-    'a,
     R: rsmq_async::RsmqConnection + Send + Sync + Clone,
 >(
     db: &DB,
@@ -65,7 +65,7 @@ pub async fn update_flow_status_after_job_completion<
     job_id_for_status: &Uuid,
     w_id: &str,
     success: bool,
-    result: &'a RawValue,
+    result: Arc<Box<RawValue>>,
     unrecoverable: bool,
     same_worker_tx: Sender<SameWorkerPayload>,
     worker_dir: &str,
@@ -105,7 +105,7 @@ pub async fn update_flow_status_after_job_completion<
             &nrec.job_id_for_status,
             w_id,
             nrec.success,
-            nrec.result.as_ref(),
+            nrec.result,
             false,
             same_worker_tx.clone(),
             worker_dir,
@@ -127,7 +127,9 @@ pub async fn update_flow_status_after_job_completion<
                     &nrec.job_id_for_status,
                     w_id,
                     false,
-                    &to_raw_value(&Json(&WrappedError { error: json!(e.to_string()) })),
+                    Arc::new(to_raw_value(&Json(&WrappedError {
+                        error: json!(e.to_string()),
+                    }))),
                     true,
                     same_worker_tx.clone(),
                     worker_dir,
@@ -147,7 +149,7 @@ pub struct RecUpdateFlowStatusAfterJobCompletion {
     flow: uuid::Uuid,
     job_id_for_status: Uuid,
     success: bool,
-    result: Box<RawValue>,
+    result: Arc<Box<RawValue>>,
     stop_early_override: Option<bool>,
     skip_error_handler: bool,
 }
@@ -166,7 +168,6 @@ pub struct RowFlowStatus {
 }
 // #[instrument(level = "trace", skip_all)]
 pub async fn update_flow_status_after_job_completion_internal<
-    'a,
     R: rsmq_async::RsmqConnection + Send + Sync + Clone,
 >(
     db: &DB,
@@ -175,7 +176,7 @@ pub async fn update_flow_status_after_job_completion_internal<
     job_id_for_status: &Uuid,
     w_id: &str,
     mut success: bool,
-    result: &'a RawValue,
+    result: Arc<Box<RawValue>>,
     unrecoverable: bool,
     same_worker_tx: Sender<SameWorkerPayload>,
     worker_dir: &str,
@@ -258,7 +259,7 @@ pub async fn update_flow_status_after_job_completion_internal<
         } else if is_failure_step {
             (false, false, false)
         } else {
-            let row = sqlx::query(
+            let r = sqlx::query_as::<_, SkipIfStopped>(
             "SELECT 
                     raw_flow->'modules'->$1::int->'stop_after_if'->>'expr' as stop_early_expr,
                     (raw_flow->'modules'->$1::int->'stop_after_if'->>'skip_if_stopped')::bool as skip_if_stopped,
@@ -272,19 +273,18 @@ pub async fn update_flow_status_after_job_completion_internal<
             .fetch_one(db)
             .await
             .map_err(|e| Error::InternalErr(format!("retrieval of stop_early_expr from state: {e:#}")))?;
-            let r = SkipIfStopped::from_row(&row)?;
 
             let stop_early = success
                 && if let Some(expr) = r.stop_early_expr.clone() {
                     compute_bool_from_expr(
                         expr,
-                        Arc::new(
+                        Marc::new(
                             r.args
                                 .map(|x| x.0)
                                 .unwrap_or_else(|| serde_json::from_str("{}").unwrap())
                                 .to_owned(),
                         ),
-                        Arc::new(result.to_owned()),
+                        result.clone(),
                         None,
                         Some(client),
                         None,
@@ -654,9 +654,9 @@ pub async fn update_flow_status_after_job_completion_internal<
         let nresult = match &new_status {
             Some(FlowStatusModule::Success { flow_jobs: Some(jobs), .. })
             | Some(FlowStatusModule::Failure { flow_jobs: Some(jobs), .. }) => {
-                retrieve_flow_jobs_results(db, w_id, jobs).await?
+                Arc::new(retrieve_flow_jobs_results(db, w_id, jobs).await?)
             }
-            _ => result.to_owned(),
+            _ => result.clone(),
         };
 
         if old_status.retry.fail_count > 0
@@ -730,6 +730,8 @@ pub async fn update_flow_status_after_job_completion_internal<
         )
     };
 
+    let flow_job = Arc::new(flow_job);
+
     let done = if !should_continue_flow {
         {
             let logs = if flow_job.canceled {
@@ -837,10 +839,10 @@ pub async fn update_flow_status_after_job_completion_internal<
     } else {
         tracing::debug!(id = %flow_job.id,  "start handle flow");
         match handle_flow(
-            &flow_job,
+            flow_job.clone(),
             db,
             client,
-            Some(nresult.to_owned()),
+            Some(nresult.clone()),
             same_worker_tx.clone(),
             worker_dir,
             rsmq.clone(),
@@ -886,7 +888,7 @@ pub async fn update_flow_status_after_job_completion_internal<
                 flow: parent_job,
                 job_id_for_status: flow,
                 success: success && !is_failure_step,
-                result: nresult,
+                result: nresult.clone(),
                 stop_early_override: if stop_early {
                     Some(skip_if_stop_early)
                 } else {
@@ -906,7 +908,7 @@ async fn retrieve_flow_jobs_results(
     w_id: &str,
     job_uuids: &Vec<Uuid>,
 ) -> error::Result<Box<RawValue>> {
-    let results = sqlx::query(
+    let results = sqlx::query_as::<_, BranchResults>(
         "SELECT result, id
         FROM completed_job
         WHERE id = ANY($1) AND workspace_id = $2",
@@ -916,10 +918,7 @@ async fn retrieve_flow_jobs_results(
     .fetch_all(db)
     .await?
     .into_iter()
-    .map(|r| {
-        let br = BranchResults::from_row(&r).unwrap();
-        (br.id, br.result.to_owned())
-    })
+    .map(|br| (br.id, br.result))
     .collect::<HashMap<_, _>>();
 
     let results = job_uuids
@@ -1068,7 +1067,7 @@ fn next_retry(retry: &Retry, status: &RetryStatus) -> Option<(u16, Duration)> {
 
 async fn compute_bool_from_expr(
     expr: String,
-    flow_args: Arc<HashMap<String, Box<RawValue>>>,
+    flow_args: Marc<HashMap<String, Box<RawValue>>>,
     result: Arc<Box<RawValue>>,
     by_id: Option<IdContext>,
     client: Option<&AuthedClient>,
@@ -1163,7 +1162,7 @@ pub async fn get_step_of_flow_status(db: &DB, id: Uuid) -> error::Result<Step> {
 /// resumes should be in order of timestamp ascending, so that more recent are at the end
 #[instrument(level = "trace", skip_all)]
 async fn transform_input(
-    flow_args: Arc<HashMap<String, Box<RawValue>>>,
+    flow_args: Marc<HashMap<String, Box<RawValue>>>,
     last_result: Arc<Box<RawValue>>,
     input_transforms: &HashMap<String, InputTransform>,
     resumes: Arc<Box<RawValue>>,
@@ -1216,10 +1215,10 @@ async fn transform_input(
 
 #[instrument(level = "trace", skip_all)]
 pub async fn handle_flow<R: rsmq_async::RsmqConnection + Send + Sync + Clone>(
-    flow_job: &QueuedJob,
+    flow_job: Arc<QueuedJob>,
     db: &sqlx::Pool<sqlx::Postgres>,
     client: &AuthedClient,
-    last_result: Option<Box<RawValue>>,
+    last_result: Option<Arc<Box<RawValue>>>,
     same_worker_tx: Sender<SameWorkerPayload>,
     worker_dir: &str,
     rsmq: Option<R>,
@@ -1251,7 +1250,7 @@ pub async fn handle_flow<R: rsmq_async::RsmqConnection + Send + Sync + Clone>(
             if let Err(err) = handle_maybe_scheduled_job(
                 rsmq.clone(),
                 db,
-                flow_job,
+                &flow_job,
                 &schedule,
                 flow_job.script_path.as_ref().unwrap(),
                 &flow_job.workspace_id,
@@ -1278,7 +1277,7 @@ pub async fn handle_flow<R: rsmq_async::RsmqConnection + Send + Sync + Clone>(
         flow,
         db,
         client,
-        last_result.to_owned(),
+        last_result,
         same_worker_tx,
         worker_dir,
         rsmq,
@@ -1328,15 +1327,19 @@ fn potentially_crash_for_testing() {
     }
 }
 
+// static
+lazy_static::lazy_static! {
+    pub static ref EHM: HashMap<String, Box<RawValue>> = HashMap::new();
+}
 // #[async_recursion]
 // #[instrument(level = "trace", skip_all)]
 async fn push_next_flow_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>(
-    flow_job: &QueuedJob,
+    flow_job: Arc<QueuedJob>,
     mut status: FlowStatus,
     flow: FlowValue,
     db: &sqlx::Pool<sqlx::Postgres>,
     client: &AuthedClient,
-    last_job_result: Option<Box<RawValue>>,
+    last_job_result: Option<Arc<Box<RawValue>>>,
     same_worker_tx: Sender<SameWorkerPayload>,
     worker_dir: &str,
     rsmq: Option<R>,
@@ -1357,7 +1360,14 @@ async fn push_next_flow_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>
         .cloned()
         .unwrap_or_else(|| status.failure_module.module_status.clone());
 
-    let flow_job_args = flow_job.get_args();
+    let fj: mappable_rc::Marc<QueuedJob> = flow_job.clone().into();
+    let arc_flow_job_args: Marc<HashMap<String, Box<RawValue>>> = Marc::map(fj, |x| {
+        if let Some(args) = &x.args {
+            &args.0
+        } else {
+            &EHM
+        }
+    });
 
     // if this is an empty module of if the module has already been completed, successfully, update the parent flow
     if flow.modules.is_empty() || matches!(status_module, FlowStatusModule::Success { .. }) {
@@ -1366,7 +1376,7 @@ async fn push_next_flow_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>
                 flow: flow_job.id,
                 success: true,
                 result: if flow.modules.is_empty() {
-                    to_raw_value(&flow_job_args)
+                    to_raw_value(arc_flow_job_args.as_ref())
                 } else {
                     // it has to be an empty for loop event
                     serde_json::from_str("[]").unwrap()
@@ -1386,8 +1396,6 @@ async fn push_next_flow_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>
         return Ok(());
     }
 
-    let arc_flow_job_args = Arc::new(flow_job_args.clone());
-
     if i == 0 {
         if !flow_job.is_flow_step && flow_job.schedule_path.is_some() {
             let no_flow_overlap = sqlx::query_scalar!(
@@ -1470,7 +1478,7 @@ async fn push_next_flow_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>
     // Compute and initialize last_job_result
     let arc_last_job_result = if status_module.is_failure() {
         // if job is being retried, pass the result of its previous failure
-        Arc::new(last_job_result.unwrap_or(to_raw_value(&json!("{}"))))
+        last_job_result.unwrap_or_else(|| Arc::new(to_raw_value(&json!("{}"))))
     } else if i == 0 {
         // if it's the first job executed in the flow, pass the flow args
         Arc::new(to_raw_value(&flow_job.args))
@@ -1479,7 +1487,7 @@ async fn push_next_flow_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>
         // having last_job_result empty can happen either when the job was suspended and is being restarted, or if it's a
         // flow restart from a specific step
         if last_job_result.is_some() {
-            Arc::new(last_job_result.unwrap())
+            last_job_result.unwrap()
         } else {
             match get_previous_job_result(db, flow_job.workspace_id.as_str(), &status).await? {
                 None => Arc::new(to_raw_value(&json!("{}"))),
@@ -1514,25 +1522,19 @@ async fn push_next_flow_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>
             .await
             .context("lock flow in queue")?;
 
-            let resumes = sqlx::query(
+            let resumes = sqlx::query_as::<_, ResumeRow>(
                 "SELECT value, approver, resume_id, approved FROM resume_job WHERE job = $1 ORDER BY created_at ASC",
             )
             .bind(last)
             .fetch_all(&mut *tx)
             .await?
             .into_iter()
-            .map(|x| ResumeRow::from_row(&x))
             .collect::<Vec<_>>();
 
-            resume_messages.extend(
-                resumes
-                    .iter()
-                    .map(|r| to_raw_value(&r.as_ref().map(|x| x.value.clone()).ok())),
-            );
+            resume_messages.extend(resumes.iter().map(|r| to_raw_value(&r.value)));
             approvers.extend(resumes.iter().map(|r| {
-                r.as_ref()
-                    .ok()
-                    .and_then(|x| x.approver.clone())
+                r.approver
+                    .clone()
                     .as_deref()
                     .unwrap_or_else(|| "anonymous")
                     .to_string()
@@ -1599,9 +1601,7 @@ async fn push_next_flow_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>
                 .await?;
             }
 
-            let is_disapproved = resumes
-                .iter()
-                .find(|x| x.as_ref().is_ok_and(|x| !x.approved));
+            let is_disapproved = resumes.iter().find(|x| !x.approved);
             if is_disapproved.is_none() && resume_messages.len() >= required_events as usize {
                 sqlx::query(
                     "UPDATE queue
@@ -1612,11 +1612,9 @@ async fn push_next_flow_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>
                 .bind(json!(resumes
                     .into_iter()
                     .map(|r| Approval {
-                        resume_id: r.as_ref().map(|x| x.resume_id).unwrap_or_default() as u16,
+                        resume_id: r.resume_id as u16,
                         approver: r
-                            .as_ref()
-                            .ok()
-                            .and_then(|x| x.approver.clone())
+                            .approver.clone()
                             .unwrap_or_else(|| "unknown".to_string())
                     })
                     .collect::<Vec<_>>()))
@@ -1676,10 +1674,7 @@ async fn push_next_flow_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>
 
                 let (logs, error_name) = if let Some(disapprover) = is_disapproved {
                     (
-                        format!(
-                            "Disapproved by {:?}",
-                            disapprover.as_ref().unwrap().approver
-                        ),
+                        format!("Disapproved by {:?}", disapprover.approver),
                         "SuspendedDisapproved",
                     )
                 } else {
@@ -1875,80 +1870,83 @@ async fn push_next_flow_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>
 
     drop(resume_messages);
 
-    let args: windmill_common::error::Result<_> = if module.mock.is_some()
-        && module.mock.as_ref().unwrap().enabled
-    {
-        let mut hm = HashMap::new();
-        hm.insert(
-            "previous_result".to_string(),
-            to_raw_value(
-                &module
-                    .mock
-                    .as_ref()
-                    .unwrap()
-                    .return_value
-                    .clone()
-                    .unwrap_or_else(|| serde_json::from_str("null").unwrap()),
-            ),
-        );
-        Ok(hm)
-    } else if let Some(id) = get_args_from_id {
-        let row = sqlx::query("SELECT args FROM completed_job WHERE id = $1 AND workspace_id = $2")
+    let args: windmill_common::error::Result<_> =
+        if module.mock.is_some() && module.mock.as_ref().unwrap().enabled {
+            let mut hm = HashMap::new();
+            hm.insert(
+                "previous_result".to_string(),
+                to_raw_value(
+                    &module
+                        .mock
+                        .as_ref()
+                        .unwrap()
+                        .return_value
+                        .clone()
+                        .unwrap_or_else(|| serde_json::from_str("null").unwrap()),
+                ),
+            );
+            Ok(Marc::new(hm))
+        } else if let Some(id) = get_args_from_id {
+            let row = sqlx::query_as::<_, RawArgs>(
+                "SELECT args FROM completed_job WHERE id = $1 AND workspace_id = $2",
+            )
             .bind(id)
             .bind(&flow_job.workspace_id)
             .fetch_optional(db)
             .await?;
-        if let Some(row) = row {
-            RawArgs::from_row(&row)
-                .map(|x| x.args.map(|x| x.0).unwrap_or_else(HashMap::new))
-                .map_err(|e| error::Error::InternalErr(format!("Impossible to build args: {e:#}")))
+            if let Some(raw_args) = row {
+                Ok(Marc::new(
+                    raw_args.args.map(|x| x.0).unwrap_or_else(HashMap::new),
+                ))
+            } else {
+                Ok(Marc::new(HashMap::new()))
+            }
         } else {
-            Ok(HashMap::new())
-        }
-    } else {
-        match &module.get_value() {
-            Ok(
-                FlowModuleValue::Script { input_transforms, .. }
-                | FlowModuleValue::RawScript { input_transforms, .. }
-                | FlowModuleValue::Flow { input_transforms, .. },
-            ) => {
-                let ctx = get_transform_context(&flow_job, &previous_id, &status).await?;
-                transform_context = Some(ctx);
-                let by_id = transform_context.as_ref().unwrap();
-                transform_input(
-                    arc_flow_job_args.clone(),
-                    arc_last_job_result.clone(),
-                    input_transforms,
-                    resumes.clone(),
-                    resume.clone(),
-                    approvers.clone(),
-                    by_id,
-                    client,
+            match &module.get_value() {
+                Ok(
+                    FlowModuleValue::Script { input_transforms, .. }
+                    | FlowModuleValue::RawScript { input_transforms, .. }
+                    | FlowModuleValue::Flow { input_transforms, .. },
+                ) => {
+                    let ctx = get_transform_context(&flow_job, &previous_id, &status).await?;
+                    transform_context = Some(ctx);
+                    let by_id = transform_context.as_ref().unwrap();
+                    transform_input(
+                        arc_flow_job_args.clone(),
+                        arc_last_job_result.clone(),
+                        input_transforms,
+                        resumes.clone(),
+                        resume.clone(),
+                        approvers.clone(),
+                        by_id,
+                        client,
+                    )
+                    .await
+                    .map(Marc::new)
+                }
+                Ok(FlowModuleValue::Identity) => serde_json::from_str(
+                    &serde_json::to_string(&PreviousResult {
+                        previous_result: Some(&arc_last_job_result),
+                    })
+                    .unwrap(),
                 )
-                .await
-            }
-            Ok(FlowModuleValue::Identity) => serde_json::from_str(
-                &serde_json::to_string(&PreviousResult {
-                    previous_result: Some(&arc_last_job_result),
-                })
-                .unwrap(),
-            )
-            .map_err(|e| error::Error::InternalErr(format!("identity: {e:#}"))),
+                .map(Marc::new)
+                .map_err(|e| error::Error::InternalErr(format!("identity: {e:#}"))),
 
-            Ok(_) => Ok(flow_job_args),
-            Err(e) => {
-                return Err(error::Error::InternalErr(format!(
-                    "module was not convertible to acceptable value {e:?}"
-                )))
+                Ok(_) => Ok(arc_flow_job_args.clone()),
+                Err(e) => {
+                    return Err(error::Error::InternalErr(format!(
+                        "module was not convertible to acceptable value {e:?}"
+                    )))
+                }
             }
-        }
-    };
+        };
     tracing::debug!(id = %flow_job.id, root_id = %job_root, "flow job args computed");
 
     let next_flow_transform = compute_next_flow_transform(
         arc_flow_job_args.clone(),
         arc_last_job_result.clone(),
-        flow_job,
+        &flow_job,
         &flow,
         transform_context,
         db,
@@ -2007,7 +2005,7 @@ async fn push_next_flow_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>
     };
 
     let mut tx: QueueTransaction<'_, R> = (rsmq.clone(), db.begin().await?).into();
-
+    let nargs = args.as_ref();
     for i in (0..len).into_iter() {
         if i % 100 == 0 && i != 0 {
             tracing::info!(id = %flow_job.id, root_id = %job_root, "pushed (non-commited yet) first {i} subflows of {len}");
@@ -2039,19 +2037,20 @@ async fn push_next_flow_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>
             None
         };
 
-        let transform_inp;
+        let marc;
+        let me;
         let args = match &next_status {
             NextStatus::AllFlowJobs {
                 branchall: Some(BranchAllStatus { .. }),
                 iterator: None,
                 ..
-            } => args.as_ref().map(|args| args.clone()),
+            } => nargs,
             NextStatus::NextLoopIteration {
                 next: ForloopNextIteration { new_args, .. },
                 simple_input_transforms,
             } => {
-                let mut args = if let Ok(args) = args.as_ref() {
-                    args.clone()
+                let mut args = if let Ok(args) = nargs {
+                    args.as_ref().clone()
                 } else {
                     HashMap::new()
                 };
@@ -2061,8 +2060,8 @@ async fn push_next_flow_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>
                 if let Some(input_transforms) = simple_input_transforms {
                     //previous id is none because we do not want to use previous id if we are in a for loop
                     let ctx = get_transform_context(&flow_job, "", &status).await?;
-                    transform_inp = transform_input(
-                        Arc::new(args),
+                    let ti = transform_input(
+                        Marc::new(args),
                         arc_last_job_result.clone(),
                         input_transforms,
                         resumes.clone(),
@@ -2071,10 +2070,23 @@ async fn push_next_flow_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>
                         &ctx,
                         client,
                     )
-                    .await;
-                    transform_inp.as_ref().map(|args| args.clone())
+                    .await
+                    .map_err(|e| {
+                        Error::ExecutionErr(
+                            format!("could not transform input using an expr: {e}",),
+                        )
+                    })
+                    .map(Marc::new);
+                    if let Ok(ti) = ti {
+                        marc = ti;
+                        Ok(&marc)
+                    } else {
+                        me = ti.unwrap_err();
+                        Err(&me)
+                    }
                 } else {
-                    Ok(args)
+                    marc = Marc::new(args);
+                    Ok(&marc)
                 }
             }
             NextStatus::AllFlowJobs {
@@ -2084,7 +2096,7 @@ async fn push_next_flow_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>
             } => {
                 if let Ok(args) = args.as_ref() {
                     let mut hm = HashMap::new();
-                    for (k, v) in args {
+                    for (k, v) in args.iter() {
                         hm.insert(k.to_string(), v.to_owned());
                     }
                     insert_iter_arg(
@@ -2094,8 +2106,8 @@ async fn push_next_flow_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>
                     );
                     if let Some(input_transforms) = simple_input_transforms {
                         let ctx = get_transform_context(&flow_job, &previous_id, &status).await?;
-                        transform_inp = transform_input(
-                            Arc::new(hm),
+                        let ti = transform_input(
+                            Marc::new(hm),
                             arc_last_job_result.clone(),
                             input_transforms,
                             resumes.clone(),
@@ -2104,20 +2116,45 @@ async fn push_next_flow_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>
                             &ctx,
                             client,
                         )
-                        .await;
-                        transform_inp.as_ref().map(|args| args.clone())
+                        .await
+                        .map_err(|e| {
+                            Error::ExecutionErr(format!(
+                                "could not transform input using an expr: {e}"
+                            ))
+                        })
+                        .map(Marc::new);
+                        if let Ok(ti) = ti {
+                            marc = ti;
+                            Ok(&marc)
+                        } else {
+                            me = ti.unwrap_err();
+                            Err(&me)
+                        }
                     } else {
-                        Ok(hm)
+                        marc = Marc::new(hm);
+                        Ok(&marc)
                     }
                 } else {
-                    args.as_ref().map(|args| args.clone())
+                    nargs
                 }
             }
-            _ => args.as_ref().map(|args| args.clone()),
+            _ => nargs,
         };
-        let (ok, err) = match args {
-            Ok(v) => (Some(v), None),
-            Err(e) => (None, Some(e)),
+
+        let push_args;
+        let err;
+        let ov;
+
+        match args {
+            Ok(v) => {
+                ov = v;
+                push_args = PushArgs::from(ov.as_ref());
+                err = None;
+            }
+            Err(e) => {
+                push_args = PushArgs::from(&*EHM);
+                err = Some(e);
+            }
         };
 
         tracing::debug!(id = %flow_job.id, root_id = %job_root, "computed args for job {i} of {len}");
@@ -2161,10 +2198,7 @@ async fn push_next_flow_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>
             tx2,
             &flow_job.workspace_id,
             payload_tag.payload,
-            windmill_queue::PushArgs {
-                args: ok.unwrap_or_else(|| serde_json::from_str("{}").unwrap()),
-                extra: HashMap::new(),
-            },
+            push_args,
             &flow_job.created_by,
             &flow_job.email,
             flow_job.permissioned_as.to_owned(),
@@ -2514,7 +2548,7 @@ fn insert_iter_arg(
 }
 
 async fn compute_next_flow_transform(
-    arc_flow_job_args: Arc<HashMap<String, Box<RawValue>>>,
+    arc_flow_job_args: Marc<HashMap<String, Box<RawValue>>>,
     arc_last_job_result: Arc<Box<RawValue>>,
     flow_job: &QueuedJob,
     flow: &FlowValue,
@@ -3041,7 +3075,7 @@ async fn next_forloop_status(
     resumes: Arc<Box<RawValue>>,
     resume: Arc<Box<RawValue>>,
     approvers: Arc<Box<RawValue>>,
-    arc_flow_job_args: Arc<HashMap<String, Box<RawValue>>>,
+    arc_flow_job_args: Marc<HashMap<String, Box<RawValue>>>,
     client: &AuthedClient,
     parallel: &bool,
 ) -> Result<ForLoopStatus, Error> {

backend/windmill-worker/src/worker_lockfiles.rs

@@ -451,7 +451,7 @@ async fn trigger_dependents_to_recompute_dependencies<
             tx,
             &w_id,
             job_payload,
-            windmill_queue::PushArgs { args, extra: HashMap::new() },
+            windmill_queue::PushArgs { args: &args, extra: None },
             &created_by,
             email,
             permissioned_as.to_string(),

frontend/src/lib/components/ArgInput.svelte

@@ -714,7 +714,11 @@
 						multiple={false}
 					/>
 					{#if value?.length}
-						<div class="text-2xs text-tertiary mt-1">File length: {value.length} base64 chars</div>
+						<div class="text-2xs text-tertiary mt-1"
+							>File length: {value.length} base64 chars ({(value.length / 1024 / 1024).toFixed(
+								2
+							)}MB)</div
+						>
 					{/if}
 				</div>
 			{:else if inputCat == 'resource-string'}

frontend/src/lib/components/FlowStatusViewerInner.svelte

@@ -586,7 +586,7 @@
 					$localDurationStatuses[flowJobIds?.moduleId ?? '']?.iteration_from ?? 0
 				)}
 				<h3 class="text-md leading-6 font-bold text-tertiary border-b mb-4">
-					Embedded flows: ({flowJobIds?.flowJobs.length} items)
+					Subflows: ({flowJobIds?.flowJobs.length} items)
 				</h3>
 				{#if (flowJobIds?.flowJobs.length ?? 0) > 20 && lenToAdd > 0}
 					{@const allToAdd = (flowJobIds?.length ?? 0) - (slicedListJobIds.length ?? 0)}

frontend/src/lib/components/JobArgs.svelte

@@ -53,7 +53,7 @@ ${Object.entries(args)
 </script>
 
 {#if id && workspace && args && typeof args === 'object' && deepEqual( Object.keys(args), ['reason'] ) && args['reason'] == 'WINDMILL_TOO_BIG'}
-	The args are too big in size to be able to fetch s3. Please <a
+	The args are too big in size to be able to fetch alongside job. Please <a
 		href="/api/w/{workspace}/jobs_u/get_args/{id}"
 		target="_blank">download the JSON file to view them</a
 	>.

frontend/src/lib/components/LightweightArgInput.svelte

@@ -480,7 +480,10 @@
 							multiple={false}
 						/>
 						{#if value?.length}
-							<div class="text-2xs text-tertiary mt-1">File length: {value.length} base64 chars</div
+							<div class="text-2xs text-tertiary mt-1"
+								>File length: {value.length} base64 chars ({(value.length / 1024 / 1024).toFixed(
+									2
+								)}MB)</div
 							>
 						{/if}
 					</div>

frontend/src/lib/components/flows/content/FlowModuleEarlyStop.svelte

@@ -8,12 +8,23 @@
 	import { getContext } from 'svelte'
 	import { NEVER_TESTED_THIS_FAR } from '../models'
 	import Section from '$lib/components/Section.svelte'
+	import { getStepPropPicker } from '../previousResults'
 
-	const { flowStateStore } = getContext<FlowEditorContext>('FlowEditorContext')
+	const { flowStateStore, flowStore, previewArgs } =
+		getContext<FlowEditorContext>('FlowEditorContext')
 
 	export let flowModule: FlowModule
 
 	let editor: SimpleEditor | undefined = undefined
+	$: stepPropPicker = getStepPropPicker(
+		$flowStateStore,
+		undefined,
+		undefined,
+		flowModule.id,
+		$flowStore,
+		$previewArgs,
+		false
+	)
 
 	$: isStopAfterIfEnabled = Boolean(flowModule.stop_after_if)
 	$: result = $flowStateStore[flowModule.id]?.previewResult ?? NEVER_TESTED_THIS_FAR
@@ -52,16 +63,19 @@
 		>
 			{#if flowModule.stop_after_if}
 				<Toggle
+					size="xs"
 					bind:checked={flowModule.stop_after_if.skip_if_stopped}
 					options={{
 						right: 'Label flow as "skipped" if stopped'
 					}}
 				/>
-				<span class="text-xs font-bold">Stop condition expression</span>
+				<span class="mt-2 text-xs font-bold">Stop condition expression</span>
 				<div class="border w-full">
 					<PropPickerWrapper
-						{result}
+						notSelectable
+						flow_input={stepPropPicker.pickableProperties.flow_input}
 						pickableProperties={undefined}
+						{result}
 						on:select={({ detail }) => {
 							editor?.insertAtCursor(detail)
 							editor?.focus()
@@ -79,10 +93,11 @@
 			{:else}
 				<Toggle
 					disabled
+					size="xs"
 					options={{
 						right: 'Label flow as "skipped" if stopped'
 					}}
-				/> <span class="text-xs font-bold">Stop condition expression</span>
+				/> <span class="mt-2 text-xs font-bold">Stop condition expression</span>
 				<textarea disabled rows="3" class="min-h-[80px]" />
 			{/if}
 		</div>

frontend/src/lib/components/flows/content/FlowModuleSleep.svelte

@@ -77,6 +77,7 @@
 			</div>
 		{:else}
 			<SecondsInput disabled />
+			<div class="text-secondary">OR use a dynamic expression</div>
 		{/if}
 	</Label>
 </Section>

frontend/src/lib/components/flows/propPicker/PropPickerWrapper.svelte

@@ -28,6 +28,7 @@
 
 	export let pickableProperties: PickableProperties | undefined
 	export let result: any = undefined
+	export let flow_input: any = undefined
 	export let error: boolean = false
 	export let displayContext = true
 	export let notSelectable = false
@@ -72,6 +73,7 @@
 			{#if result}
 				<PropPickerResult
 					{result}
+					{flow_input}
 					on:select={({ detail }) => {
 						if (!notSelectable && !$propPickerConfig) {
 							sendUserToast('Set cursor within an input or click on the plug first', true)

frontend/src/lib/components/propertyPicker/PropPickerResult.svelte

@@ -1,7 +1,11 @@
 <script lang="ts">
+	import { createEventDispatcher } from 'svelte'
 	import ObjectViewer from './ObjectViewer.svelte'
 
 	export let result: any
+	export let flow_input: any = undefined
+
+	const dispatch = createEventDispatcher()
 </script>
 
 <div class="w-full px-2">
@@ -9,4 +13,14 @@
 	<div class="overflow-y-auto mb-2 w-full">
 		<ObjectViewer allowCopy={false} json={{ result }} on:select />
 	</div>
+	{#if flow_input}
+		<span class="font-bold text-sm">Flow Input</span>
+		<div class="overflow-y-auto w-full">
+			<ObjectViewer
+				allowCopy={false}
+				json={flow_input}
+				on:select={(e) => dispatch('select', `flow_input.${e.detail}`)}
+			/>
+		</div>
+	{/if}
 </div>

frontend/src/routes/(root)/(logged)/flows/get/[...path]/+page.svelte

@@ -105,15 +105,21 @@
 	) {
 		loading = true
 		const scheduledFor = scheduledForStr ? new Date(scheduledForStr).toISOString() : undefined
-		let run = await JobService.runFlowByPath({
-			workspace: $workspaceStore!,
-			path,
-			invisibleToOwner,
-			requestBody: args,
-			scheduledFor,
-			tag: overrideTag
-		})
-		await goto('/run/' + run + '?workspace=' + $workspaceStore)
+		try {
+			let run = await JobService.runFlowByPath({
+				workspace: $workspaceStore!,
+				path,
+				invisibleToOwner,
+				requestBody: args,
+				scheduledFor,
+				tag: overrideTag
+			})
+			await goto('/run/' + run + '?workspace=' + $workspaceStore)
+		} catch (e) {
+			throw e
+		} finally {
+			loading = false
+		}
 	}
 
 	let args: Record<string, any> | undefined = undefined