Merge branch 'main' into all_runs

backend/Cargo.lock

@@ -10834,7 +10834,6 @@ dependencies = [
  "bigdecimal",
  "chrono",
  "chrono-tz 0.10.0",
- "const_format",
  "cron",
  "futures-core",
  "hex",

backend/migrations/20241004112544_create_params_args_result_tables.down.sql

@@ -1,3 +0,0 @@
-DROP TABLE job_params;
-DROP TABLE job_args;
-DROP TABLE completed_jobs_result;

backend/migrations/20241004112544_create_params_args_result_tables.up.sql

@@ -1,25 +0,0 @@
--- Add up migration script here
--- Add down migration script here
--- Add up migration script here
-CREATE TABLE job_params (
-  id UUID PRIMARY KEY,
-  raw_code TEXT,
-  raw_flow jsonb NULL,
-  tag VARCHAR(50),
-  workspace_id VARCHAR(50)
-);
-
--- Add up migration script here
-CREATE TABLE job_args (
-  id UUID PRIMARY KEY,
-  args JSONB,
-  tag VARCHAR(50),
-  workspace_id VARCHAR(50)
-);
-
-CREATE TABLE completed_jobs_result (
-  id UUID PRIMARY KEY,
-  result JSONB,
-  tag VARCHAR(50),
-  workspace_id VARCHAR(50)
-);

backend/src/monitor.rs

@@ -19,7 +19,6 @@ use tokio::{
     sync::{mpsc, RwLock},
 };
 
-use uuid::Uuid;
 #[cfg(feature = "embedding")]
 use windmill_api::embeddings::update_embeddings_db;
 use windmill_api::{
@@ -33,7 +32,7 @@ use windmill_common::{
     auth::JWT_SECRET,
     ee::CriticalErrorChannel,
     error,
-    flow_status::{FlowStatusModule, ParsedFlowStatusGetter as _},
+    flow_status::FlowStatusModule,
     global_settings::{
         BASE_URL_SETTING, BUNFIG_INSTALL_SCOPES_SETTING, CRITICAL_ERROR_CHANNELS_SETTING,
         DEFAULT_TAGS_PER_WORKSPACE_SETTING, DEFAULT_TAGS_WORKSPACES_SETTING,
@@ -1421,7 +1420,7 @@ async fn handle_zombie_flows(
                 }
             );
             report_critical_error(reason.clone(), db.clone()).await;
-            cancel_zombie_flow_job(db, &flow.id, &flow.workspace_id, &rsmq, reason).await?;
+            cancel_zombie_flow_job(db, flow, &rsmq, reason).await?;
         }
     }
 
@@ -1437,17 +1436,11 @@ async fn handle_zombie_flows(
     .fetch_all(db)
     .await?;
 
-    #[derive(sqlx::FromRow, Debug)]
-    struct InQueueJobResult {
-        id: uuid::Uuid,
-        workspace_id: String,
-    }
-
     for flow in flows2 {
-        let in_queue = sqlx::query_as!(InQueueJobResult,
-            "SELECT id, workspace_id FROM queue WHERE id = $1 AND running = true AND canceled = false",
-            flow.parent_flow_id
+        let in_queue = sqlx::query_as::<_, QueuedJob>(
+            "SELECT * FROM queue WHERE id = $1 AND running = true AND canceled = false",
         )
+        .bind(flow.parent_flow_id)
         .fetch_optional(db)
         .await?;
         if let Some(job) = in_queue {
@@ -1457,7 +1450,7 @@ async fn handle_zombie_flows(
                 job.workspace_id,
                 flow.last_ping
             );
-            cancel_zombie_flow_job(db, &job.id, &job.workspace_id, &rsmq,
+            cancel_zombie_flow_job(db, job, &rsmq,
                 format!("Flow {} cancelled as one of the parallel branch {} was unable to make the last transition ", flow.parent_flow_id, flow.job_id))
                 .await?;
         } else {
@@ -1469,22 +1462,21 @@ async fn handle_zombie_flows(
 
 async fn cancel_zombie_flow_job(
     db: &Pool<Postgres>,
-    job_id: &Uuid,
-    workspace_id: &str,
+    flow: QueuedJob,
     rsmq: &Option<MultiplexedRsmq>,
     message: String,
 ) -> Result<(), error::Error> {
     let tx = db.begin().await.unwrap();
     tracing::error!(
         "zombie flow detected: {} in workspace {}. Cancelling it.",
-        job_id,
-        workspace_id
+        flow.id,
+        flow.workspace_id
     );
     let (ntx, _) = cancel_job(
         "monitor",
         Some(message),
-        *job_id,
-        workspace_id,
+        flow.id,
+        flow.workspace_id.as_str(),
         tx,
         db,
         rsmq.clone(),

backend/windmill-api/src/inputs.rs

@@ -21,7 +21,11 @@ use std::{
     vec,
 };
 use windmill_common::{
-    db::UserDB, error::JsonResult, jobs::JobKind, query_scalar_with_fallback, scripts::to_i64, utils::{not_found_if_none, paginate, Pagination}
+    db::UserDB,
+    error::JsonResult,
+    jobs::JobKind,
+    scripts::to_i64,
+    utils::{not_found_if_none, paginate, Pagination},
 };
 pub fn workspaced_service() -> Router {
     Router::new()
@@ -174,7 +178,6 @@ struct GetArgs {
     input: Option<bool>,
     allow_large: Option<bool>,
 }
-
 async fn get_args_from_history_or_saved_input(
     authed: ApiAuthed,
     Extension(user_db): Extension<UserDB>,
@@ -182,7 +185,6 @@ async fn get_args_from_history_or_saved_input(
     Path((w_id, job_or_input_id)): Path<(String, Uuid)>,
 ) -> JsonResult<Option<Value>> {
     let mut tx = user_db.begin(&authed).await?;
-
     let result_o = if let Some(input) = g.input {
         if input {
             sqlx::query_scalar!(
@@ -194,20 +196,24 @@ async fn get_args_from_history_or_saved_input(
                 .fetch_optional(&mut *tx)
                 .await?
         } else {
-            query_scalar_with_fallback!(tx, 
-                "SELECT CASE WHEN pg_column_size(args) < 40000 OR $3 THEN args ELSE '\"WINDMILL_TOO_BIG\"'::jsonb END as args FROM job_args WHERE id = $1 AND workspace_id = $2", 
+            sqlx::query_scalar!(
                 "SELECT CASE WHEN pg_column_size(args) < 40000 OR $3 THEN args ELSE '\"WINDMILL_TOO_BIG\"'::jsonb END as args FROM completed_job WHERE id = $1 AND workspace_id = $2",
-                job_or_input_id, 
-                w_id,
-                g.allow_large.unwrap_or(true))?
-        }
-    } else {
-        query_scalar_with_fallback!(tx, 
-                 "SELECT CASE WHEN pg_column_size(args) < 40000 OR $3 THEN args ELSE '\"WINDMILL_TOO_BIG\"'::jsonb END as args FROM job_args WHERE id = $1 AND workspace_id = $2 UNION ALL SELECT CASE WHEN pg_column_size(args) < 40000 OR $3 THEN args ELSE '\"WINDMILL_TOO_BIG\"'::jsonb END as args FROM input WHERE id = $1 AND workspace_id = $2", 
-                "SELECT CASE WHEN pg_column_size(args) < 40000 OR $3 THEN args ELSE '\"WINDMILL_TOO_BIG\"'::jsonb END as args FROM completed_job WHERE id = $1 AND workspace_id = $2 UNION ALL SELECT CASE WHEN pg_column_size(args) < 40000 OR $3 THEN args ELSE '\"WINDMILL_TOO_BIG\"'::jsonb END as args FROM input WHERE id = $1 AND workspace_id = $2",
                 job_or_input_id,
                 w_id,
-                g.allow_large.unwrap_or(true))?
+                g.allow_large.unwrap_or(true)
+            )
+            .fetch_optional(&mut *tx)
+            .await?
+        }
+    } else {
+        sqlx::query_scalar!(
+        "SELECT CASE WHEN pg_column_size(args) < 40000 OR $3 THEN args ELSE '\"WINDMILL_TOO_BIG\"'::jsonb END as args FROM completed_job WHERE id = $1 AND workspace_id = $2 UNION ALL SELECT CASE WHEN pg_column_size(args) < 40000 OR $3 THEN args ELSE '\"WINDMILL_TOO_BIG\"'::jsonb END as args FROM input WHERE id = $1 AND workspace_id = $2",
+        job_or_input_id,
+        w_id,
+        g.allow_large.unwrap_or(true)
+    )
+    .fetch_optional(&mut *tx)
+    .await?
     };
 
     tx.commit().await?;

backend/windmill-api/src/jobs.rs

@@ -314,14 +314,9 @@ async fn get_result_by_id(
     Path((w_id, flow_id, node_id)): Path<(String, Uuid, String)>,
     Query(JsonPath { json_path, .. }): Query<JsonPath>,
 ) -> windmill_common::error::JsonResult<Box<JsonRawValue>> {
-    let res = windmill_queue::get_result_by_id(
-        db.clone(),
-        &w_id,
-        flow_id,
-        &node_id,
-        json_path.as_deref(),
-    )
-    .await?;
+    let res =
+        windmill_queue::get_result_by_id(db.clone(), w_id.clone(), flow_id, node_id, json_path)
+            .await?;
 
     log_job_view(&db, Some(&authed), &w_id, &flow_id).await?;
 
@@ -1354,8 +1349,6 @@ async fn cancel_jobs(
 ) -> error::JsonResult<Vec<Uuid>> {
     let mut uuids = vec![];
     let mut tx = db.begin().await?;
-
-    let result = serde_json::json!({"error": { "message": format!("Job canceled: cancel all by {username}"), "name": "Canceled", "reason": "cancel all", "canceler": username}});
     let trivial_jobs =  sqlx::query!("INSERT INTO completed_job AS cj
                    ( workspace_id
                    , id
@@ -1419,19 +1412,10 @@ async fn cancel_jobs(
                    , tag
                    , priority FROM queue 
         WHERE id = any($2) AND running = false AND parent_job IS NULL AND workspace_id = $3 AND schedule_path IS NULL FOR UPDATE SKIP LOCKED
-        ON CONFLICT (id) DO NOTHING RETURNING id", username, &jobs, w_id, &result)
+        ON CONFLICT (id) DO NOTHING RETURNING id", username, &jobs, w_id, serde_json::json!({"error": { "message": format!("Job canceled: cancel all by {username}"), "name": "Canceled", "reason": "cancel all", "canceler": username}}))
         .fetch_all(&mut *tx)
         .await?.into_iter().map(|x| x.id).collect::<Vec<Uuid>>();
 
-    sqlx::query!(
-        "INSERT INTO completed_jobs_result(id, result, tag, workspace_id) SELECT id, $1, tag, $2 FROM completed_job  WHERE id = any($3) ON CONFLICT (id) DO NOTHING",
-        result,
-        w_id,
-        &trivial_jobs,
-    )
-    .execute(&mut *tx)
-    .await?;
-
     sqlx::query!(
         "DELETE FROM queue WHERE id = any($1) AND workspace_id = $2",
         &trivial_jobs,
@@ -1439,7 +1423,6 @@ async fn cancel_jobs(
     )
     .execute(&mut *tx)
     .await?;
-
     tx.commit().await?;
 
     // sqlx::query!(
@@ -1453,10 +1436,18 @@ async fn cancel_jobs(
             continue;
         }
         let rsmq = rsmq.clone();
-        if let Ok(result) = tokio::time::timeout(tokio::time::Duration::from_secs(5), async move {
+        match tokio::time::timeout(tokio::time::Duration::from_secs(5), async move {
             let tx = db.begin().await?;
             let (tx, _) = windmill_queue::cancel_job(
-                username, None, job_id, w_id, tx, db, rsmq, false, false,
+                username,
+                None,
+                job_id.clone(),
+                w_id,
+                tx,
+                db,
+                rsmq,
+                false,
+                false,
             )
             .await?;
             tx.commit().await?;
@@ -1464,19 +1455,20 @@ async fn cancel_jobs(
         })
         .await
         {
-            match result {
+            Ok(result) => match result {
                 Ok(_) => {
                     uuids.push(job_id);
                 }
                 Err(e) => {
                     tracing::error!("Failed to cancel job {:?}: {:?}", job_id, e);
                 }
+            },
+            Err(_) => {
+                tracing::error!(
+                    "Timeout while trying to cancel job {:?} after 5 seconds",
+                    job_id
+                );
             }
-        } else {
-            tracing::error!(
-                "Timeout while trying to cancel job {:?} after 5 seconds",
-                job_id
-            );
         }
     }
 
@@ -3291,7 +3283,7 @@ async fn run_wait_result(
     };
 
     let fast_poll_duration = *WAIT_RESULT_FAST_POLL_DURATION_SECS as u64 * 1000;
-    let mut accumulated_delay = 0_u64;
+    let mut accumulated_delay = 0 as u64;
 
     loop {
         if let Some(node_id_for_empty_return) = node_id_for_empty_return.as_ref() {
@@ -3408,16 +3400,6 @@ async fn delete_job_metadata_after_use(db: &DB, job_uuid: Uuid) -> Result<(), Er
     )
     .execute(db)
     .await?;
-
-    sqlx::query!(
-        "UPDATE job_args
-        SET args = '{}'::jsonb
-        WHERE id = $1",
-        job_uuid,
-    )
-    .execute(db)
-    .await?;
-
     sqlx::query!(
         "UPDATE job_logs
         SET logs = '##DELETED##'
@@ -4402,8 +4384,11 @@ async fn add_batch_jobs(
                 tx = PushIsolationLevel::Transaction(ntx);
                 uuids.push(uuid);
             }
-            if let PushIsolationLevel::Transaction(tx) = tx {
-                tx.commit().await?;
+            match tx {
+                PushIsolationLevel::Transaction(tx) => {
+                    tx.commit().await?;
+                }
+                _ => (),
             }
             return Ok(Json(uuids));
         }

backend/windmill-api/src/resources.rs

@@ -31,6 +31,7 @@ use windmill_audit::ActionKind;
 use windmill_common::{
     db::UserDB,
     error::{Error, JsonResult, Result},
+    jobs::QueuedJob,
     utils::{not_found_if_none, paginate, require_admin, Pagination, StripPath},
     variables,
 };
@@ -535,29 +536,11 @@ pub async fn transform_json_value<'c>(
         }
         Value::String(y) if y.starts_with("$") && job_id.is_some() => {
             let mut tx = authed_transaction_or_default(authed, user_db.clone(), db).await?;
-
-            #[derive(sqlx::FromRow, Debug)]
-            struct QueuedJobLite {
-                pub id: Uuid,
-                pub workspace_id: String,
-                pub parent_job: Option<Uuid>,
-                pub created_by: String,
-                pub email: String,
-                pub permissioned_as: String,
-                pub script_path: Option<String>,
-                pub schedule_path: Option<String>,
-                pub root_job: Option<Uuid>,
-                pub flow_step_id: Option<String>,
-                pub scheduled_for: chrono::DateTime<chrono::Utc>,
-            }
-
-            let job = sqlx::query_as!(
-                QueuedJobLite,
-                "SELECT id, workspace_id,parent_job, created_by, email, permissioned_as, script_path, schedule_path, root_job, flow_step_id, scheduled_for
-                 FROM queue WHERE id = $1 AND workspace_id = $2",
-                job_id.unwrap(),
-                workspace
+            let job = sqlx::query_as::<_, QueuedJob>(
+                "SELECT * FROM queue WHERE id = $1 AND workspace_id = $2",
             )
+            .bind(job_id.unwrap())
+            .bind(workspace)
             .fetch_optional(&mut *tx)
             .await?;
             tx.commit().await?;

backend/windmill-common/src/flow_status.rs

@@ -45,33 +45,6 @@ pub struct FlowStatus {
     pub restarted_from: Option<RestartedFrom>,
 }
 
-pub trait FlowStatusGetter {
-    fn get_raw_flow_status(&self) -> Option<&sqlx::types::Json<Box<serde_json::value::RawValue>>>;
-}
-
-#[macro_export]
-macro_rules! impl_flow_status_getter {
-    ($struct_name:ident) => {
-        impl FlowStatusGetter for $struct_name {
-            fn get_raw_flow_status(
-                &self,
-            ) -> Option<&sqlx::types::Json<Box<serde_json::value::RawValue>>> {
-                self.flow_status.as_ref()
-            }
-        }
-    };
-}
-
-pub trait ParsedFlowStatusGetter {
-    fn parse_flow_status(&self) -> Option<FlowStatus>;
-}
-impl<I: FlowStatusGetter> ParsedFlowStatusGetter for I {
-    fn parse_flow_status(&self) -> Option<FlowStatus> {
-        self.get_raw_flow_status()
-            .and_then(|v| serde_json::from_str::<FlowStatus>((**v).get()).ok())
-    }
-}
-
 #[derive(Serialize, Deserialize, Debug, Clone, Default)]
 #[serde(default)]
 pub struct RetryStatus {

backend/windmill-common/src/flows.rs

@@ -125,34 +125,6 @@ pub struct FlowValue {
     pub concurrency_key: Option<String>,
 }
 
-pub trait FlowValueGetter {
-    fn get_raw_flow_value(&self) -> Option<&sqlx::types::Json<Box<serde_json::value::RawValue>>>;
-}
-
-#[macro_export]
-macro_rules! impl_flow_value_getter {
-    ($struct_name:ident) => {
-        impl FlowValueGetter for $struct_name {
-            fn get_raw_flow_value(
-                &self,
-            ) -> Option<&sqlx::types::Json<Box<serde_json::value::RawValue>>> {
-                self.raw_flow.as_ref()
-            }
-        }
-    };
-}
-
-pub trait ParsedFlowValueGetter {
-    fn parse_raw_flow(&self) -> Option<FlowValue>;
-}
-
-impl<I: FlowValueGetter> ParsedFlowValueGetter for I {
-    fn parse_raw_flow(&self) -> Option<FlowValue> {
-        self.get_raw_flow_value()
-            .and_then(|v| serde_json::from_str::<FlowValue>((**v).get()).ok())
-    }
-}
-
 #[derive(Deserialize, Serialize, Debug, Clone)]
 pub struct StopAfterIf {
     pub expr: String,

backend/windmill-common/src/jobs.rs

@@ -15,9 +15,9 @@ pub const PREPROCESSOR_FAKE_ENTRYPOINT: &str = "__WM_PREPROCESSOR";
 
 use crate::{
     error::{self, to_anyhow, Error},
-    flow_status::{FlowStatusGetter, RestartedFrom},
-    flows::{FlowValue, FlowValueGetter, Retry},
-    get_latest_deployed_hash_for_path, impl_flow_status_getter, impl_flow_value_getter,
+    flow_status::{FlowStatus, RestartedFrom},
+    flows::{FlowValue, Retry},
+    get_latest_deployed_hash_for_path,
     scripts::{ScriptHash, ScriptLang},
     worker::{to_raw_value, TMP_DIR},
 };
@@ -119,7 +119,10 @@ pub struct QueuedJob {
 
 impl QueuedJob {
     pub fn script_path(&self) -> &str {
-        self.script_path.as_deref().unwrap_or("tmp/main")
+        self.script_path
+            .as_ref()
+            .map(String::as_str)
+            .unwrap_or("tmp/main")
     }
     pub fn is_flow(&self) -> bool {
         matches!(
@@ -136,10 +139,21 @@ impl QueuedJob {
             self.script_path()
         )
     }
-}
 
-impl_flow_status_getter!(QueuedJob);
-impl_flow_value_getter!(QueuedJob);
+    pub fn parse_raw_flow(&self) -> Option<FlowValue> {
+        self.raw_flow.as_ref().and_then(|v| {
+            let str = (**v).get();
+            // tracing::error!("raw_flow: {}", str);
+            return serde_json::from_str::<FlowValue>(str).ok();
+        })
+    }
+
+    pub fn parse_flow_status(&self) -> Option<FlowStatus> {
+        self.flow_status
+            .as_ref()
+            .and_then(|v| serde_json::from_str::<FlowStatus>((**v).get()).ok())
+    }
+}
 
 impl Default for QueuedJob {
     fn default() -> Self {
@@ -252,13 +266,23 @@ impl CompletedJob {
     pub fn json_result(&self) -> Option<serde_json::Value> {
         self.result
             .as_ref()
-            .and_then(|r| serde_json::from_str(r.get()).ok())
+            .map(|r| serde_json::from_str(r.get()).ok())
+            .flatten()
+    }
+
+    pub fn parse_raw_flow(&self) -> Option<FlowValue> {
+        self.raw_flow
+            .as_ref()
+            .and_then(|v| serde_json::from_str::<FlowValue>((**v).get()).ok())
+    }
+
+    pub fn parse_flow_status(&self) -> Option<FlowStatus> {
+        self.flow_status
+            .as_ref()
+            .and_then(|v| serde_json::from_str::<FlowStatus>((**v).get()).ok())
     }
 }
 
-impl_flow_status_getter!(CompletedJob);
-impl_flow_value_getter!(CompletedJob);
-
 #[derive(sqlx::FromRow)]
 pub struct BranchResults {
     pub result: sqlx::types::Json<Box<RawValue>>,

backend/windmill-common/src/lib.rs

@@ -6,7 +6,6 @@
  * LICENSE-AGPL for a copy of the license.
  */
 
-pub mod macros;
 use std::{
     net::SocketAddr,
     sync::{atomic::AtomicBool, Arc},

backend/windmill-common/src/macros.rs

@@ -1,57 +0,0 @@
-#[macro_export]
-macro_rules! fetch_one_with_fallback {
-    ($db:expr, $query_method:ident, $row_type:ty, $query:literal, $table:literal || $fallback_table:literal, $( $param:expr ),* ) => {{
-        let primary_query = sqlx::$query_method::<_, $row_type>(const_format::formatcp!($query, $table))
-            $(.bind($param))*
-            .fetch_one($db)
-            .await;
-
-        if let Err(sqlx::Error::RowNotFound) = primary_query {
-            tracing::info!("Data not found in job_params, falling back to fetching from $fallback_table");
-            sqlx::$query_method::<_, $row_type>(const_format::formatcp!($query, $fallback_table))
-                $(.bind($param))*
-                .fetch_one($db)
-                .await
-        } else {
-            primary_query
-        }
-    }};
-}
-
-#[macro_export]
-macro_rules! fetch_optional_with_fallback {
-    ($db:expr, $query_method:ident, $row_type:ty, $query:literal, $table:literal || $fallback_table:literal, $( $param:expr ),* ) => {{
-        let primary_query = sqlx::$query_method::<_, $row_type>(const_format::formatcp!($query, $table))
-            $(.bind($param))*
-            .fetch_optional($db)
-            .await;
-
-        if let Ok(None) = primary_query {
-            tracing::info!("Data not found in job_params, falling back to fetching from $fallback_table");
-            sqlx::$query_method::<_, $row_type>(const_format::formatcp!($query, $fallback_table))
-                $(.bind($param))*
-                .fetch_optional($db)
-                .await
-        } else {
-            primary_query
-        }
-    }};
-}
-
-#[macro_export]
-macro_rules! query_scalar_with_fallback {
-    ($tx:expr, $query:literal, $fallback_query:literal,$( $param:expr ),* ) => {{
-        let primary_query = sqlx::query_scalar!($query,$($param),*)
-            .fetch_optional(&mut *$tx)
-            .await;
-
-        if let Ok(None) = primary_query {
-            tracing::info!("Data not found in job_args, falling back to fetching from queue");
-            sqlx::query_scalar!($fallback_query, $($param),*)
-                .fetch_optional(&mut *$tx)
-                .await
-        } else {
-            primary_query
-        }
-    }};
-}

backend/windmill-common/src/worker.rs

@@ -91,7 +91,7 @@ lazy_static::lazy_static! {
 }
 
 pub async fn make_suspended_pull_query(wc: &WorkerConfig) {
-    if wc.worker_tags.is_empty() {
+    if wc.worker_tags.len() == 0 {
         tracing::error!("Empty tags in worker tags, skipping");
         return;
     }
@@ -123,7 +123,7 @@ pub async fn make_suspended_pull_query(wc: &WorkerConfig) {
 pub async fn make_pull_query(wc: &WorkerConfig) {
     let mut queries = vec![];
     for tags in wc.priority_tags_sorted.iter() {
-        if tags.tags.is_empty() {
+        if tags.tags.len() == 0 {
             tracing::error!("Empty tags in priority tags, skipping");
             continue;
         }

backend/windmill-queue/Cargo.toml

@@ -42,5 +42,4 @@ async-recursion.workspace = true
 bigdecimal.workspace = true
 axum.workspace = true
 serde_urlencoded.workspace = true
-regex.workspace = true
-const_format.workspace = true
+regex.workspace = true

backend/windmill-queue/src/jobs.rs

@@ -6,6 +6,8 @@
  * LICENSE-AGPL for a copy of the license.
  */
 
+use std::{borrow::Borrow, collections::HashMap, sync::Arc, vec};
+
 use anyhow::Context;
 use async_recursion::async_recursion;
 use axum::{
@@ -29,14 +31,12 @@ use serde_json::{json, value::RawValue};
 use sqlx::{types::Json, FromRow, Pool, Postgres, Transaction};
 #[cfg(feature = "benchmark")]
 use std::time::Instant;
-use std::{borrow::Borrow, collections::HashMap, sync::Arc, vec};
 use tokio::{sync::RwLock, time::sleep};
 use tracing::{instrument, Instrument};
 use ulid::Ulid;
 use uuid::Uuid;
 use windmill_audit::audit_ee::{audit_log, AuditAuthor};
 use windmill_audit::ActionKind;
-use windmill_common::{fetch_optional_with_fallback, flows::FlowValueGetter};
 
 use windmill_common::{
     add_time,
@@ -44,15 +44,12 @@ use windmill_common::{
     db::{Authed, UserDB},
     error::{self, to_anyhow, Error},
     flow_status::{
-        BranchAllStatus, FlowCleanupModule, FlowStatus, FlowStatusGetter, FlowStatusModule,
-        FlowStatusModuleWParent, Iterator, JobResult, ParsedFlowStatusGetter, RestartedFrom,
-        RetryStatus, MAX_RETRY_ATTEMPTS, MAX_RETRY_INTERVAL,
+        BranchAllStatus, FlowCleanupModule, FlowStatus, FlowStatusModule, FlowStatusModuleWParent,
+        Iterator, JobResult, RestartedFrom, RetryStatus, MAX_RETRY_ATTEMPTS, MAX_RETRY_INTERVAL,
     },
     flows::{
         add_virtual_items_if_necessary, FlowModule, FlowModuleValue, FlowValue, InputTransform,
-        ParsedFlowValueGetter,
     },
-    impl_flow_status_getter, impl_flow_value_getter,
     jobs::{
         get_payload_tag_from_prefixed_path, CompletedJob, JobKind, JobPayload, QueuedJob, RawCode,
         ENTRYPOINT_OVERRIDE, PREPROCESSOR_FAKE_ENTRYPOINT,
@@ -145,14 +142,6 @@ pub struct CanceledBy {
     pub reason: Option<String>,
 }
 
-#[derive(Debug, sqlx::FromRow)]
-pub struct CompletedSubFlow {
-    pub id: Uuid,
-    pub flow_status: Option<sqlx::types::Json<Box<RawValue>>>,
-}
-
-impl_flow_status_getter!(CompletedSubFlow);
-
 pub async fn cancel_single_job<'c>(
     username: &str,
     reason: Option<String>,
@@ -233,23 +222,26 @@ pub async fn cancel_job<'c>(
     force_cancel: bool,
     require_anonymous: bool,
 ) -> error::Result<(Transaction<'c, Postgres>, Option<Uuid>)> {
-    let Some(mut job) = get_queued_job_tx(id, w_id, &mut tx).await? else {
-        return Ok((tx, None));
-    };
+    let job = get_queued_job_tx(id, &w_id, &mut tx).await?;
 
-    if require_anonymous && job.created_by != "anonymous" {
+    if job.is_none() {
+        return Ok((tx, None));
+    }
+
+    if require_anonymous && job.as_ref().unwrap().created_by != "anonymous" {
         return Err(Error::BadRequest(
             "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();
     if force_cancel {
         // if force canceling a flow step, make sure we force cancel from the highest parent
         loop {
             if job.parent_job.is_none() {
                 break;
             }
-            match get_queued_job_tx(job.parent_job.unwrap(), w_id, &mut tx).await? {
+            match get_queued_job_tx(job.parent_job.unwrap(), &w_id, &mut tx).await? {
                 Some(j) => {
                     job = j;
                 }
@@ -595,8 +587,6 @@ pub async fn add_completed_job<
 
     let mem_peak = mem_peak.max(queued_job.mem_peak.unwrap_or(0));
     add_time!(bench, "add_completed_job query START");
-
-    // On conflict (when id already exists), update the success and result fields.
     let _duration: i64 = sqlx::query_scalar!(
         "INSERT INTO completed_job AS cj
                    ( workspace_id
@@ -633,8 +623,8 @@ pub async fn add_completed_job<
             VALUES ($1, $2, $3, $4, $5, COALESCE($6, now()), (EXTRACT('epoch' FROM (now())) - EXTRACT('epoch' FROM (COALESCE($6, now()))))*1000, $7, $8, $9,\
                     $10, $11, $12, $13, $14, $15, $16, $17, $18, $19, $20, $21, $22, $23, $24, $25, $26, $27, $28, $29)
          ON CONFLICT (id) DO UPDATE SET success = $7, result = $11 RETURNING duration_ms",
-        &queued_job.workspace_id,
-        &queued_job.id,
+        queued_job.workspace_id,
+        queued_job.id,
         queued_job.parent_job,
         queued_job.created_by,
         queued_job.created_at,
@@ -643,7 +633,7 @@ pub async fn add_completed_job<
         queued_job.script_hash.map(|x| x.0),
         queued_job.script_path,
         &queued_job.args as &Option<Json<HashMap<String, Box<RawValue>>>>,
-        &result as &Json<&T>,
+        result as Json<&T>,
         queued_job.raw_code,
         queued_job.raw_lock,
         canceled_by.is_some(),
@@ -660,32 +650,16 @@ pub async fn add_completed_job<
         queued_job.email,
         queued_job.visible_to_owner,
         if mem_peak > 0 { Some(mem_peak) } else { None },
-        &queued_job.tag,
+        queued_job.tag,
         queued_job.priority,
     )
     .fetch_one(&mut tx)
     .await
     .map_err(|e| Error::InternalErr(format!("Could not add completed job {job_id}: {e:#}")))?;
+    // tracing::error!("2 {:?}", start.elapsed());
+
     add_time!(bench, "add_completed_job query END");
 
-    add_time!(bench, "completed_jobs_result query START");
-    sqlx::query!(
-        "INSERT INTO completed_jobs_result(id, result, tag, workspace_id) VALUES($1, $2, $3, $4) ON CONFLICT (id) DO UPDATE SET result = $2",
-        queued_job.id,
-        &result as &Json<&T>,
-        queued_job.tag,
-        queued_job.workspace_id,
-    )
-    .execute(&mut tx)
-    .await
-    .map_err(|e| {
-        Error::InternalErr(format!(
-            "Could not add completed job result {job_id}: {e:#}"
-        ))
-    })?;
-
-    add_time!(bench, "completed_jobs_result query END");
-
     if !queued_job.is_flow_step {
         if _duration > 500
             && (queued_job.job_kind == JobKind::Script || queued_job.job_kind == JobKind::Preview)
@@ -890,10 +864,10 @@ pub async fn add_completed_job<
     tx.commit().await?;
     tracing::info!(
         %job_id,
-        root_job = ?queued_job.root_job.map(|x| x.to_string()).unwrap_or_default(),
+        root_job = ?queued_job.root_job.map(|x| x.to_string()).unwrap_or_else(|| String::new()),
         path = &queued_job.script_path(),
         job_kind = ?queued_job.job_kind,
-        started_at = ?queued_job.started_at.map(|x| x.to_string()).unwrap_or_default(),
+        started_at = ?queued_job.started_at.map(|x| x.to_string()).unwrap_or_else(|| String::new()),
         duration = ?_duration,
         permissioned_as = ?queued_job.permissioned_as,
         email = ?queued_job.email,
@@ -2293,48 +2267,55 @@ pub struct ResultWithId {
 
 pub async fn get_result_by_id(
     db: Pool<Postgres>,
-    w_id: &str,
+    w_id: String,
     flow_id: Uuid,
-    node_id: &str,
-    json_path: Option<&str>,
+    node_id: String,
+    json_path: Option<String>,
 ) -> error::Result<Box<RawValue>> {
-    #[derive(sqlx::FromRow, Debug)]
-    struct RunningFlowJobResult {
-        pub id: Uuid,
-        pub flow_status: Option<Json<Box<RawValue>>>,
-    }
-    impl_flow_status_getter!(RunningFlowJobResult);
-
-    match get_result_by_id_from_running_flow(&db, w_id, &flow_id, node_id, json_path).await {
+    match get_result_by_id_from_running_flow(
+        &db,
+        w_id.as_str(),
+        &flow_id,
+        node_id.as_str(),
+        json_path.clone(),
+    )
+    .await
+    {
         Ok(res) => Ok(res),
         Err(_) => {
-            let running_flow_job =sqlx::query_as::<_, RunningFlowJobResult>(
-                "SELECT id, flow_status FROM queue WHERE COALESCE((SELECT root_job FROM queue WHERE id = $1), $1) = id AND workspace_id = $2"
+            let running_flow_job =sqlx::query_as::<_, QueuedJob>(
+                "SELECT * FROM queue WHERE COALESCE((SELECT root_job FROM queue WHERE id = $1), $1) = id AND workspace_id = $2"
             ).bind(flow_id)
             .bind(&w_id)
             .fetch_optional(&db).await?;
-
             match running_flow_job {
                 Some(job) => {
                     let restarted_from = windmill_common::utils::not_found_if_none(
                         job.parse_flow_status()
-                            .and_then(|status| status.restarted_from),
+                            .map(|status| status.restarted_from)
+                            .flatten(),
                         "Id not found in the result's mapping of the root job and root job had no restarted from information",
                         format!("parent: {}, root: {}, id: {}", flow_id, job.id, node_id),
                     )?;
 
                     get_result_by_id_from_original_flow(
                         &db,
-                        w_id,
+                        w_id.as_str(),
                         &restarted_from.flow_job_id,
-                        node_id,
-                        json_path,
+                        node_id.as_str(),
+                        json_path.clone(),
                     )
                     .await
                 }
                 None => {
-                    get_result_by_id_from_original_flow(&db, w_id, &flow_id, node_id, json_path)
-                        .await
+                    get_result_by_id_from_original_flow(
+                        &db,
+                        w_id.as_str(),
+                        &flow_id,
+                        node_id.as_str(),
+                        json_path.clone(),
+                    )
+                    .await
                 }
             }
         }
@@ -2353,7 +2334,7 @@ pub async fn get_result_by_id_from_running_flow(
     w_id: &str,
     flow_id: &Uuid,
     node_id: &str,
-    json_path: Option<&str>,
+    json_path: Option<String>,
 ) -> error::Result<Box<RawValue>> {
     let flow_job_result = sqlx::query_as::<_, FlowJobResult>(
         "SELECT leaf_jobs->$1::text as leaf_jobs, parent_job FROM queue WHERE COALESCE((SELECT root_job FROM queue WHERE id = $2), $2) = id AND workspace_id = $3")
@@ -2371,7 +2352,8 @@ pub async fn get_result_by_id_from_running_flow(
 
     let job_result = flow_job_result
         .leaf_jobs
-        .and_then(|x| serde_json::from_str(x.get()).ok());
+        .map(|x| serde_json::from_str(x.get()).ok())
+        .flatten();
 
     if job_result.is_none() && flow_job_result.parent_job.is_some() {
         let parent_job = flow_job_result.parent_job.unwrap();
@@ -2396,9 +2378,9 @@ pub async fn get_result_by_id_from_running_flow(
 async fn get_completed_flow_node_result_rec(
     db: &Pool<Postgres>,
     w_id: &str,
-    subflows: &[CompletedSubFlow],
+    subflows: Vec<CompletedJob>,
     node_id: &str,
-    json_path: Option<&str>,
+    json_path: Option<String>,
 ) -> error::Result<Option<Box<RawValue>>> {
     for subflow in subflows {
         let flow_status = subflow.parse_flow_status().ok_or_else(|| {
@@ -2415,7 +2397,7 @@ async fn get_completed_flow_node_result_rec(
                     db,
                     w_id,
                     JobResult::SingleJob(leaf_job_uuid),
-                    json_path,
+                    json_path.clone(),
                 )
                 .await
                 .map(Some),
@@ -2423,7 +2405,7 @@ async fn get_completed_flow_node_result_rec(
                     db,
                     w_id,
                     JobResult::ListJob(jobs),
-                    json_path,
+                    json_path.clone(),
                 )
                 .await
                 .map(Some),
@@ -2434,11 +2416,10 @@ async fn get_completed_flow_node_result_rec(
                 ))),
             };
         } else {
-            let subflows = sqlx::query_as::<_, CompletedSubFlow>(
-                "SELECT id, flow_status as labels FROM completed_job WHERE parent_job = $1 AND workspace_id = $2 AND flow_status IS NOT NULL",
+            let subflows = sqlx::query_as::<_, CompletedJob>(
+                "SELECT *, null as labels FROM completed_job WHERE parent_job = $1 AND workspace_id = $2 AND flow_status IS NOT NULL",
             ).bind(subflow.id).bind(w_id).fetch_all(db).await?;
-
-            match get_completed_flow_node_result_rec(db, w_id, &subflows, node_id, json_path)
+            match get_completed_flow_node_result_rec(db, w_id, subflows, node_id, json_path.clone())
                 .await?
             {
                 Some(res) => return Ok(Some(res)),
@@ -2455,10 +2436,10 @@ async fn get_result_by_id_from_original_flow(
     w_id: &str,
     completed_flow_id: &Uuid,
     node_id: &str,
-    json_path: Option<&str>,
+    json_path: Option<String>,
 ) -> error::Result<Box<RawValue>> {
-    let flow_job = sqlx::query_as::<_, CompletedSubFlow>(
-        "SELECT id, flow_status FROM completed_job WHERE id = $1 AND workspace_id = $2",
+    let flow_job = sqlx::query_as::<_, CompletedJob>(
+        "SELECT *, null as labels FROM completed_job WHERE id = $1 AND workspace_id = $2",
     )
     .bind(completed_flow_id)
     .bind(w_id)
@@ -2471,7 +2452,7 @@ async fn get_result_by_id_from_original_flow(
         format!("root: {}, id: {}", completed_flow_id, node_id),
     )?;
 
-    match get_completed_flow_node_result_rec(db, w_id, &[flow_job], node_id, json_path).await? {
+    match get_completed_flow_node_result_rec(db, w_id, vec![flow_job], node_id, json_path).await? {
         Some(res) => Ok(res),
         None => Err(error::Error::NotFound(format!(
             "Flow result by id not found going top-down from {}, (id: {})",
@@ -2484,33 +2465,31 @@ async fn extract_result_from_job_result(
     db: &Pool<Postgres>,
     w_id: &str,
     job_result: JobResult,
-    json_path: Option<&str>,
+    json_path: Option<String>,
 ) -> error::Result<Box<RawValue>> {
     match job_result {
         JobResult::ListJob(job_ids) => match json_path {
             Some(json_path) => {
-                let mut parts = json_path.split('.');
+                let mut parts = json_path.split(".");
 
-                let Some(ref idx) = parts.next().and_then(|x| x.parse::<usize>().ok()) else {
+                let Some(idx) = parts.next().map(|x| x.parse::<usize>().ok()).flatten() else {
                     return Ok(to_raw_value(&serde_json::Value::Null));
                 };
-                let Some(job_id) = job_ids.get(*idx) else {
+                let Some(job_id) = job_ids.get(idx).cloned() else {
                     return Ok(to_raw_value(&serde_json::Value::Null));
                 };
-
-                let parts = parts.map(|x| x.to_string()).collect_vec();
-
-                Ok(fetch_optional_with_fallback!(
-                    db,
-                    query_as,
-                    ResultR,
-                    "SELECT result #> $3 as result FROM {} WHERE id = $1 AND workspace_id = $2",
-                    "completed_jobs_result" || "completed_job",
-                    *job_id,
-                    w_id,
-                    &parts
-                )?
-                .and_then(|r| r.result.map(|x| x.0))
+                Ok(sqlx::query_as::<_, ResultR>(
+                    "SELECT result #> $3 as result FROM completed_job WHERE id = $1 AND workspace_id = $2",
+                )
+                .bind(job_id)
+                .bind(w_id)
+                .bind(
+                    parts.map(|x| x.to_string()).collect::<Vec<_>>()
+                )
+                .fetch_optional(db)
+                .await?
+                .map(|r| r.result.map(|x| x.0))
+                .flatten()
                 .unwrap_or_else(|| to_raw_value(&serde_json::Value::Null)))
             }
             None => {
@@ -2524,7 +2503,6 @@ async fn extract_result_from_job_result(
                 .into_iter()
                 .filter_map(|x| x.result.map(|y| (x.id, y)))
                 .collect::<HashMap<Uuid, Json<Box<RawValue>>>>();
-
                 let result = job_ids
                     .into_iter()
                     .map(|id| {
@@ -2532,31 +2510,25 @@ async fn extract_result_from_job_result(
                             .map(|x| x.0.clone())
                             .unwrap_or_else(|| to_raw_value(&serde_json::Value::Null))
                     })
-                    .collect_vec();
+                    .collect::<Vec<_>>();
                 Ok(to_raw_value(&result))
             }
         },
-        // ici
-        JobResult::SingleJob(x) => {
-            let path = json_path
+        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)
+        .bind(w_id)
+        .bind(
+            json_path
                 .map(|x| x.split(".").map(|x| x.to_string()).collect::<Vec<_>>())
-                .unwrap_or_default();
-
-            let res = fetch_optional_with_fallback!(
-                db,
-                query_as,
-                ResultR,
-                "SELECT result #> $3 as result FROM {} WHERE id = $1 AND workspace_id = $2",
-                "completed_jobs_result" || "completed_job",
-                x,
-                w_id,
-                &path
-            )?
-            .and_then(|r| r.result.map(|x| x.0))
-            .unwrap_or_else(|| to_raw_value(&serde_json::Value::Null));
-
-            Ok(res)
-        }
+                .unwrap_or_default(),
+        )
+        .fetch_optional(db)
+        .await?
+        .map(|r| r.result.map(|x| x.0))
+        .flatten()
+        .unwrap_or_else(|| to_raw_value(&serde_json::Value::Null))),
     }
 }
 
@@ -2671,7 +2643,7 @@ pub struct PushArgsOwned {
     pub args: HashMap<String, Box<RawValue>>,
 }
 
-#[derive(Debug, Clone)]
+#[derive(Debug)]
 pub struct PushArgs<'c> {
     pub extra: Option<HashMap<String, Box<RawValue>>>,
     pub args: &'c HashMap<String, Box<RawValue>>,
@@ -2782,7 +2754,7 @@ fn restructure_cloudevents_metadata(
         .unwrap_or_else(|| to_raw_value(&serde_json::Value::Null));
     let str = data.to_string();
 
-    let wrap_body = !str.is_empty() && !str.starts_with('{');
+    let wrap_body = str.len() > 0 && str.chars().next().unwrap() != '{';
 
     if wrap_body {
         let args = serde_json::from_str::<Option<Box<RawValue>>>(&str)
@@ -2812,7 +2784,7 @@ impl PushArgsOwned {
             extra.insert("raw_string".to_string(), to_raw_value(&str));
         }
 
-        let wrap_body = force_wrap_body || !str.is_empty() && !str.starts_with('{');
+        let wrap_body = force_wrap_body || str.len() > 0 && str.chars().next().unwrap() != '{';
 
         if wrap_body {
             let args = serde_json::from_str::<Option<Box<RawValue>>>(&str)
@@ -3923,10 +3895,6 @@ pub async fn push<'c, 'd, R: rsmq_async::RsmqConnection + Send + 'c>(
     };
 
     tracing::debug!("Pushing job {job_id} with tag {tag}, schedule_path {schedule_path:?}, script_path: {script_path:?}, email {email}, workspace_id {workspace_id}");
-
-    let raw_flow = raw_flow.map(Json);
-    let args = Json(args);
-
     let uuid = sqlx::query_scalar!(
         "INSERT INTO queue
             (workspace_id, id, running, parent_job, created_by, permissioned_as, scheduled_for, 
@@ -3945,12 +3913,12 @@ pub async fn push<'c, 'd, R: rsmq_async::RsmqConnection + Send + 'c>(
         scheduled_for_o,
         script_hash,
         script_path.clone(),
-        raw_code.clone(),
+        raw_code,
         raw_lock,
-        args.clone() as Json<PushArgs>,
+        Json(args) as Json<PushArgs>,
         job_kind.clone() as JobKind,
         schedule_path,
-        raw_flow.clone() as Option<Json<FlowValue>>,
+        raw_flow.map(Json) as Option<Json<FlowValue>>,
         flow_status.map(Json) as Option<Json<FlowStatus>>,
         is_flow_step,
         language as Option<ScriptLang>,
@@ -3959,7 +3927,7 @@ pub async fn push<'c, 'd, R: rsmq_async::RsmqConnection + Send + 'c>(
         email,
         visible_to_owner,
         root_job,
-        tag.clone(),
+        tag,
         concurrent_limit,
         if concurrent_limit.is_some() {  concurrency_time_window_s } else { None },
         custom_timeout,
@@ -3971,34 +3939,6 @@ pub async fn push<'c, 'd, 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:#}")))?;
 
-    // insert into args queue
-    sqlx::query!(
-        r#"
-        INSERT INTO job_args (id, workspace_id, args, tag)
-        VALUES ($1, $2, $3, $4)
-        "#,
-        uuid,
-        workspace_id,
-        args as Json<PushArgs>,
-        &tag
-    )
-    .execute(&mut tx)
-    .await?;
-
-    sqlx::query!(
-        r#"
-        INSERT INTO job_params (id, workspace_id, raw_code, raw_flow, tag)
-        VALUES ($1, $2, $3, $4, $5)
-        "#,
-        uuid,
-        workspace_id,
-        raw_code,
-        raw_flow as Option<Json<FlowValue>>,
-        tag
-    )
-    .execute(&mut tx)
-    .await?;
-
     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")]
@@ -4119,8 +4059,11 @@ pub async fn push<'c, 'd, R: rsmq_async::RsmqConnection + Send + 'c>(
 }
 
 pub fn canceled_job_to_result(job: &QueuedJob) -> serde_json::Value {
-    let reason = job.canceled_reason.as_deref().unwrap_or("no reason given");
-    let canceler = job.canceled_by.as_deref().unwrap_or("unknown");
+    let reason = job
+        .canceled_reason
+        .as_deref()
+        .unwrap_or_else(|| "no reason given");
+    let canceler = job.canceled_by.as_deref().unwrap_or_else(|| "unknown");
     serde_json::json!({"message": format!("Job canceled: {reason} by {canceler}"), "name": "Canceled", "reason": reason, "canceler": canceler})
 }
 
@@ -4143,19 +4086,8 @@ async fn restarted_flows_resolution(
     ),
     Error,
 > {
-    #[derive(Debug, sqlx::FromRow)]
-    struct QueryResults {
-        pub script_path: Option<String>,
-        pub priority: Option<i16>,
-        pub raw_flow: Option<sqlx::types::Json<Box<RawValue>>>,
-        pub flow_status: Option<sqlx::types::Json<Box<RawValue>>>,
-    }
-
-    impl_flow_status_getter!(QueryResults);
-    impl_flow_value_getter!(QueryResults);
-
-    let completed_job = sqlx::query_as::<_, QueryResults>(
-        "SELECT script_path, priority, raw_flow, flow_status FROM completed_job WHERE id = $1 and workspace_id = $2",
+    let completed_job = sqlx::query_as::<_, CompletedJob>(
+        "SELECT *, null as labels FROM completed_job WHERE id = $1 and workspace_id = $2",
     )
     .bind(completed_flow_id)
     .bind(workspace_id)
@@ -4188,9 +4120,10 @@ async fn restarted_flows_resolution(
         if flow_value_if_any
             .clone()
             .map(|fv| {
-                !fv.modules
+                fv.modules
                     .iter()
-                    .any(|flow_value_module| flow_value_module.id == module.id())
+                    .find(|flow_value_module| flow_value_module.id == module.id())
+                    .is_none()
             })
             .unwrap_or(false)
         {
@@ -4306,7 +4239,7 @@ async fn restarted_flows_resolution(
             truncated_modules.push(FlowStatusModule::WaitingForPriorSteps { id: module.id() });
         } else {
             // else we simply "transfer" the module from the completed flow to the new one if it's a success
-            step_n += 1;
+            step_n = step_n + 1;
             match module.clone() {
                 FlowStatusModule::Success { .. } => Ok(truncated_modules.push(module)),
                 _ => Err(Error::InternalErr(format!(

backend/windmill-worker/src/worker_flow.rs

@@ -29,20 +29,20 @@ use sqlx::FromRow;
 use tokio::sync::mpsc::Sender;
 use tracing::instrument;
 use uuid::Uuid;
+use windmill_common::add_time;
 use windmill_common::auth::JobPerms;
 #[cfg(feature = "benchmark")]
 use windmill_common::bench::BenchmarkIter;
 use windmill_common::db::Authed;
 use windmill_common::flow_status::{
-    ApprovalConditions, FlowStatusModuleWParent, Iterator, JobResult, ParsedFlowStatusGetter,
+    ApprovalConditions, FlowStatusModuleWParent, Iterator, JobResult,
 };
-use windmill_common::flows::{add_virtual_items_if_necessary, ParsedFlowValueGetter};
+use windmill_common::flows::add_virtual_items_if_necessary;
 use windmill_common::jobs::{
     script_hash_to_tag_and_limits, script_path_to_payload, BranchResults, JobPayload, QueuedJob,
     RawCode, ENTRYPOINT_OVERRIDE,
 };
 use windmill_common::worker::to_raw_value;
-use windmill_common::{add_time, fetch_one_with_fallback, fetch_optional_with_fallback};
 use windmill_common::{
     error::{self, to_anyhow, Error},
     flow_status::{
@@ -61,19 +61,6 @@ type DB = sqlx::Pool<sqlx::Postgres>;
 
 use windmill_queue::{canceled_job_to_result, get_queued_job_tx, push, QueueTransaction};
 
-async fn get_args_from_job_id(db: &DB, id: &Uuid) -> Result<RowArgs, Error> {
-    let args = fetch_one_with_fallback!(
-        db,
-        query_as,
-        RowArgs,
-        "SELECT args FROM {} WHERE id = $1",
-        "job_args" || "queue",
-        id
-    );
-
-    args.map_err(|e| Error::InternalErr(format!("retrieval of args from state: {e:#}")))
-}
-
 // #[instrument(level = "trace", skip_all)]
 pub async fn update_flow_status_after_job_completion<
     R: rsmq_async::RsmqConnection + Send + Sync + Clone,
@@ -242,10 +229,10 @@ pub async fn update_flow_status_after_job_completion_internal<
         })?;
 
         let old_status = serde_json::from_str::<FlowStatus>(old_status_json.flow_status.get())
-            .map_err(|e| {
-                Error::InternalErr(format!(
+            .or_else(|e| {
+                Err(Error::InternalErr(format!(
                     "requiring status to be parsable as FlowStatus: {e:?}"
-                ))
+                )))
             })?;
 
         let current_module = if let Some(x) = old_status_json.current_module {
@@ -302,7 +289,7 @@ pub async fn update_flow_status_after_job_completion_internal<
 
         // 0 length flows are not failure steps
         let is_failure_step =
-            old_status.step >= old_status.modules.len() as i32 && !old_status.modules.is_empty();
+            old_status.step >= old_status.modules.len() as i32 && old_status.modules.len() > 0;
 
         let (mut stop_early, mut skip_if_stop_early, continue_on_error) = if let Some(se) =
             stop_early_override
@@ -315,13 +302,14 @@ pub async fn update_flow_status_after_job_completion_internal<
             };
 
             let is_flow = if let Some(step) = step {
-                fetch_one_with_fallback!(db,
-                    query_scalar,
-                    Option<bool>,
-                    "SELECT raw_flow->'modules'->($1)->'value'->>'type' = 'flow' FROM {} WHERE id = $2",
-                    "job_params" || "queue",
+                sqlx::query_scalar!(
+                    "SELECT raw_flow->'modules'->($1)->'value'->>'type' = 'flow' FROM queue WHERE id = $2",
                     step as i32,
-                    &flow).map_err(|e| {
+                    &flow
+                )
+                    .fetch_one(db)
+                    .await
+                    .map_err(|e| {
                         Error::InternalErr(format!("error during retrieval of step's type: {e:#}"))
                     })?
                     .unwrap_or(false)
@@ -354,9 +342,19 @@ pub async fn update_flow_status_after_job_completion_internal<
                         }
                         _ => None,
                     };
-
-                    let args = get_args_from_job_id(db, &flow).await?;
-
+                    let args = sqlx::query_as::<_, RowArgs>(
+                        "SELECT
+                                        args
+                                    FROM queue
+                                    WHERE id = $2",
+                    )
+                    .bind(old_status.step)
+                    .bind(flow)
+                    .fetch_one(db)
+                    .await
+                    .map_err(|e| {
+                        Error::InternalErr(format!("retrieval of args from state: {e:#}"))
+                    })?;
                     compute_bool_from_expr(
                         expr.to_string(),
                         Marc::new(args.args.unwrap_or_default().0),
@@ -444,7 +442,7 @@ pub async fn update_flow_status_after_job_completion_internal<
                             None
                         };
 
-                        let nindex = if let Some(position) = position {
+                        let nindex = if let Some(position) = position { 
                             sqlx::query_scalar!(
                             "UPDATE queue
                             SET flow_status = JSONB_SET(
@@ -491,7 +489,7 @@ pub async fn update_flow_status_after_job_completion_internal<
                             None
                         };
 
-                        let nindex = if let Some(position) = position {
+                        let nindex = if let Some(position) = position { 
                             sqlx::query_scalar!(
                                 "UPDATE queue
                                 SET flow_status = JSONB_SET(
@@ -660,7 +658,7 @@ pub async fn update_flow_status_after_job_completion_internal<
                 flow_jobs_success,
                 flow_jobs,
                 ..
-            } if *branch < len - 1 && (success || skip_branch_failure) => {
+            } if branch.to_owned() < len - 1 && (success || skip_branch_failure) => {
                 if let Some(jobs) = flow_jobs {
                     set_success_in_flow_job_success(
                         flow_jobs_success,
@@ -872,7 +870,19 @@ pub async fn update_flow_status_after_job_completion_internal<
                     .as_ref()
                     .and_then(|m| m.stop_after_all_iters_if.as_ref().map(|x| x.expr.clone()))
                 {
-                    let args = get_args_from_job_id(db, &flow).await?;
+                    let args = sqlx::query_as::<_, RowArgs>(
+                        "SELECT
+                            args
+                        FROM queue
+                        WHERE id = $2",
+                    )
+                    .bind(old_status.step)
+                    .bind(flow)
+                    .fetch_one(db)
+                    .await
+                    .map_err(|e| {
+                        Error::InternalErr(format!("retrieval of args from state: {e:#}"))
+                    })?;
 
                     let should_stop = compute_bool_from_expr(
                         expr.to_string(),
@@ -1234,7 +1244,7 @@ fn get_module(flow_job: &QueuedJob, module_step: &Step) -> Option<FlowModule> {
     if let Some(raw_flow) = raw_flow {
         match module_step {
             Step::PreprocessorStep => raw_flow.preprocessor_module.map(|x| *x.clone()),
-            Step::Step(i) => raw_flow.modules.get(*i).cloned(),
+            Step::Step(i) => raw_flow.modules.get(*i).map(|x| x.clone()),
             Step::FailureStep => raw_flow.failure_module.map(|x| *x.clone()),
         }
     } else {
@@ -1259,7 +1269,8 @@ async fn compute_skip_branchall_failure<'c>(
             .map(|p| {
                 BRANCHALL_INDEX_RE
                     .captures(&p)
-                    .and_then(|x| x.get(1).unwrap().as_str().parse::<i32>().ok())
+                    .map(|x| x.get(1).unwrap().as_str().parse::<i32>().ok())
+                    .flatten()
                     .ok_or(Error::InternalErr(format!(
                         "could not parse branchall index from path: {p}"
                     )))
@@ -1280,14 +1291,14 @@ async fn compute_skip_branchall_failure<'c>(
 }
 
 async fn has_failure_module<'c>(flow: Uuid, db: &DB) -> Result<bool, Error> {
-    fetch_one_with_fallback!(
-        db,
-        query_scalar,
-        Option<bool>,
-        "SELECT raw_flow->'failure_module' != 'null'::jsonb FROM {} WHERE id = $1",
-        "job_params" || "queue",
-        flow
+    sqlx::query_scalar::<_, Option<bool>>(
+        "SELECT raw_flow->'failure_module' != 'null'::jsonb
+        FROM queue
+        WHERE id = $1",
     )
+    .bind(flow)
+    .fetch_one(db)
+    .await
     .map_err(|e| {
         Error::InternalErr(format!(
             "error during retrieval of has_failure_module: {e:#}"
@@ -1599,7 +1610,6 @@ pub struct ResumeRow {
 
 #[derive(FromRow)]
 pub struct RawArgs {
-    #[allow(dead_code)]
     pub args: Option<Json<HashMap<String, Box<RawValue>>>>,
 }
 
@@ -1628,7 +1638,6 @@ fn potentially_crash_for_testing() {
 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>(
@@ -1716,7 +1725,7 @@ async fn push_next_flow_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>
                     flow_job.workspace_id.as_str(),
                     flow_job.id
                 ).fetch_all(db).await?;
-                if !overlapping.is_empty() {
+                if overlapping.len() > 0 {
                     let overlapping_str = overlapping
                         .iter()
                         .map(|x| x.to_string())
@@ -1795,8 +1804,8 @@ async fn push_next_flow_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>
         // else pass the last job result. Either from the function arg if it's set, or manually fetch it from the previous job
         // 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 let Some(last_job_result) = last_job_result {
-            last_job_result
+        if last_job_result.is_some() {
+            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!("{}"))),
@@ -2251,18 +2260,17 @@ async fn push_next_flow_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>
             );
             Ok(Marc::new(hm))
         } else if let Some(id) = get_args_from_id {
-            let row = fetch_optional_with_fallback!(
-                db,
-                query_as,
-                RowArgs,
-                "SELECT args FROM {} WHERE id = $1 AND workspace_id = $2",
-                "job_args" || "completed_job",
-                id,
-                &flow_job.workspace_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(raw_args) = row {
-                Ok(Marc::new(raw_args.args.map(|x| x.0).unwrap_or_default()))
+                Ok(Marc::new(
+                    raw_args.args.map(|x| x.0).unwrap_or_else(HashMap::new),
+                ))
             } else {
                 Ok(Marc::new(HashMap::new()))
             }
@@ -2381,7 +2389,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 {
+    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}");
             sqlx::query!(
@@ -3894,15 +3902,13 @@ async fn get_previous_job_result(
             Ok(Some(retrieve_flow_jobs_results(db, w_id, flow_jobs).await?))
         }
         Some(FlowStatusModule::Success { job, .. }) => Ok(Some(
-            fetch_one_with_fallback!(
-                db,
-                query_scalar,
-                Json<Box<RawValue>>,
-                "SELECT result FROM {} WHERE id = $1 AND workspace_id = $2",
-                "completed_jobs_result" || "completed_job",
-                job,
-                w_id
-            )?
+            sqlx::query_scalar::<_, Json<Box<RawValue>>>(
+                "SELECT result FROM completed_job WHERE id = $1 AND workspace_id = $2",
+            )
+            .bind(job)
+            .bind(w_id)
+            .fetch_one(db)
+            .await?
             .0,
         )),
         _ => Ok(None),

frontend/.gitignore

@@ -9,5 +9,4 @@ tests-out/
 storageState.json
 .env.production
 dist/
-static/tsdocs/
-!build/.gitkeep
+static/tsdocs/

frontend/src/lib/components/SavedInputs.svelte

@@ -319,8 +319,7 @@
 
 		<Pane>
 			<div class="w-full flex flex-col gap-4 p-2">
-				<span class="text-sm font-semibold">Previous runs</span>
-
+				<div class="text-sm font-semibold">Previous runs</div>
 				<div class="w-full flex flex-col gap-1 p-0 h-full overflow-y-auto">
 					{#if loading && (jobs == undefined || jobs?.length == 0)}
 						<div class="text-left text-tertiary text-xs">Loading current runs...</div>
@@ -335,12 +334,12 @@
 									if (!$workspaceStore) {
 										return
 									}
-
+		
 									const args = await JobService.getJobArgs({
 										workspace: $workspaceStore,
 										id: i.id
 									})
-
+		
 									selectedInput = {
 										id: i.id,
 										name: 'Running job: ' + i.id,
@@ -348,7 +347,7 @@
 										created_by: i.created_by ?? '',
 										is_public: true
 									}
-
+		
 									selectArgs(args)
 								}}
 							>
@@ -444,6 +443,11 @@
 						<div class="text-center text-tertiary">No previous Runs</div>
 					{/if}
 				</div>
+				{#if scriptPath || flowPath}
+					<a href="/runs/{scriptPath || flowPath}?show_schedules=true&show_future_jobs=true" class="text-sm text-blue-600 hover:underline">
+						All runs
+					</a>
+				{/if}
 			</div>
 		</Pane>