Create dirs

Will be enabled later.
Needs proper testing and its better to split onto 2 PRs with first modifying python and second ansible

backend/.sqlx/query-0a686ca61444d7ad7484071727aa039a6ea6697e5a49a633b767c052aa3e0a18.json

@@ -1,83 +0,0 @@
-{
-  "db_name": "PostgreSQL",
-  "query": "WITH uuid_table as (\n            select gen_random_uuid() as uuid from generate_series(1, $11)\n        )\n        INSERT INTO queue \n            (id, script_hash, script_path, job_kind, language, args, tag, created_by, permissioned_as, email, scheduled_for, workspace_id, concurrent_limit, concurrency_time_window_s, timeout)\n            (SELECT uuid, $1, $2, $3, $4, ('{ \"uuid\": \"' || uuid || '\" }')::jsonb, $5, $6, $7, $8, $9, $10, $12, $13, $14 FROM uuid_table) \n        RETURNING id",
-  "describe": {
-    "columns": [
-      {
-        "ordinal": 0,
-        "name": "id",
-        "type_info": "Uuid"
-      }
-    ],
-    "parameters": {
-      "Left": [
-        "Int8",
-        "Varchar",
-        {
-          "Custom": {
-            "name": "job_kind",
-            "kind": {
-              "Enum": [
-                "script",
-                "preview",
-                "flow",
-                "dependencies",
-                "flowpreview",
-                "script_hub",
-                "identity",
-                "flowdependencies",
-                "http",
-                "graphql",
-                "postgresql",
-                "noop",
-                "appdependencies",
-                "deploymentcallback",
-                "singlescriptflow"
-              ]
-            }
-          }
-        },
-        {
-          "Custom": {
-            "name": "script_lang",
-            "kind": {
-              "Enum": [
-                "python3",
-                "deno",
-                "go",
-                "bash",
-                "postgresql",
-                "nativets",
-                "bun",
-                "mysql",
-                "bigquery",
-                "snowflake",
-                "graphql",
-                "powershell",
-                "mssql",
-                "php",
-                "bunnative",
-                "rust",
-                "ansible"
-              ]
-            }
-          }
-        },
-        "Varchar",
-        "Varchar",
-        "Varchar",
-        "Varchar",
-        "Timestamptz",
-        "Varchar",
-        "Int4",
-        "Int4",
-        "Int4",
-        "Int4"
-      ]
-    },
-    "nullable": [
-      false
-    ]
-  },
-  "hash": "0a686ca61444d7ad7484071727aa039a6ea6697e5a49a633b767c052aa3e0a18"
-}

backend/.sqlx/query-1e7ce0c140410ae799f9c0c5772e4be4506bfd238c88f1b1c3ddf39b29071446.json

@@ -0,0 +1,29 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT\n                flow_status AS \"flow_status!: Json<Box<RawValue>>\",\n                raw_flow->'modules'->(flow_status->'step')::int AS \"module: Json<Box<RawValue>>\"\n            FROM queue WHERE id = $1 AND workspace_id = $2 LIMIT 1",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "flow_status!: Json<Box<RawValue>>",
+        "type_info": "Jsonb"
+      },
+      {
+        "ordinal": 1,
+        "name": "module: Json<Box<RawValue>>",
+        "type_info": "Jsonb"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Uuid",
+        "Text"
+      ]
+    },
+    "nullable": [
+      true,
+      null
+    ]
+  },
+  "hash": "1e7ce0c140410ae799f9c0c5772e4be4506bfd238c88f1b1c3ddf39b29071446"
+}

backend/.sqlx/query-31dd23f6768052e486668172e15eba1a3f9b6a8e5678a7ee8e5456ff4405d6f9.json

@@ -0,0 +1,22 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT raw_flow->'failure_module' != 'null'::jsonb\n            FROM completed_job\n            WHERE id = $1",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "?column?",
+        "type_info": "Bool"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Uuid"
+      ]
+    },
+    "nullable": [
+      null
+    ]
+  },
+  "hash": "31dd23f6768052e486668172e15eba1a3f9b6a8e5678a7ee8e5456ff4405d6f9"
+}

backend/.sqlx/query-33cfb00f61dbae09467bd3732ce8e83c9835ccd30c90cb24788cdb54d1ceb575.json

@@ -1,20 +0,0 @@
-{
-  "db_name": "PostgreSQL",
-  "query": "SELECT value::BOOLEAN as value FROM global_settings WHERE name = 'critical_alert_mute_ui'",
-  "describe": {
-    "columns": [
-      {
-        "ordinal": 0,
-        "name": "value",
-        "type_info": "Bool"
-      }
-    ],
-    "parameters": {
-      "Left": []
-    },
-    "nullable": [
-      null
-    ]
-  },
-  "hash": "33cfb00f61dbae09467bd3732ce8e83c9835ccd30c90cb24788cdb54d1ceb575"
-}

backend/.sqlx/query-349396e8fdd96d45875110bc06767e6eb876792ec8f83e9b03c2fb46bb12e0b9.json

@@ -0,0 +1,15 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "UPDATE password SET login_type = $1 WHERE email = $2",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Varchar",
+        "Text"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "349396e8fdd96d45875110bc06767e6eb876792ec8f83e9b03c2fb46bb12e0b9"
+}

backend/.sqlx/query-aa6907b7266ee437dfbd77a9bf6c047fda88f3e72a0d10323a5e4020cf857c42.json

@@ -0,0 +1,22 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT raw_flow->'failure_module' != 'null'::jsonb\n            FROM queue\n            WHERE id = $1",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "?column?",
+        "type_info": "Bool"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Uuid"
+      ]
+    },
+    "nullable": [
+      null
+    ]
+  },
+  "hash": "aa6907b7266ee437dfbd77a9bf6c047fda88f3e72a0d10323a5e4020cf857c42"
+}

backend/.sqlx/query-ad03e5acf10ef94abc37cb9f56b1775c67f075c8bc83458e8be2e242347218d6.json

@@ -0,0 +1,22 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT wm_version FROM worker_ping WHERE wm_version != $1 AND ping_at > now() - interval '5 minutes'",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "wm_version",
+        "type_info": "Varchar"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Text"
+      ]
+    },
+    "nullable": [
+      false
+    ]
+  },
+  "hash": "ad03e5acf10ef94abc37cb9f56b1775c67f075c8bc83458e8be2e242347218d6"
+}

backend/.sqlx/query-de1abe57b6aa61155f747a3bcb98359f70eade6654b5a884915f07f3ef3fe15e.json

@@ -1,6 +1,6 @@
 {
   "db_name": "PostgreSQL",
-  "query": "SELECT raw_flow->'modules'->($1)->'value'->>'type' = 'flow' FROM queue WHERE id = $2",
+  "query": "SELECT raw_flow->'modules'->($1)::text->'value'->>'type' = 'flow' FROM queue WHERE id = $2 LIMIT 1",
   "describe": {
     "columns": [
       {
@@ -19,5 +19,5 @@
       null
     ]
   },
-  "hash": "3e539fef054ad31bc1736e27276087775a721a6ee7ae35b03fd4ce3563ea3838"
+  "hash": "de1abe57b6aa61155f747a3bcb98359f70eade6654b5a884915f07f3ef3fe15e"
 }

backend/Cargo.lock

@@ -10811,6 +10811,7 @@ dependencies = [
  "rand 0.8.5",
  "regex",
  "reqwest 0.12.9",
+ "semver 1.0.23",
  "serde",
  "serde_json",
  "sha2 0.10.8",

backend/Cargo.toml

@@ -180,6 +180,7 @@ tokio-util = { version = "^0", features = ["io"] }
 json-pointer = "^0"
 itertools = "^0"
 regex = "^1"
+semver = "^1"
 
 deno_fetch = "0.195.0"
 deno_tls = "0.158.0"

backend/ee-repo-ref.txt

@@ -1 +1 @@
-e3df64af864f9d3546a97ff21f719c553e34ff1c
+0d9c8813acd28848515c736e7b684220b5a785a3

backend/src/main.rs

@@ -7,7 +7,6 @@
  */
 
 use anyhow::Context;
-use git_version::git_version;
 use monitor::{
     reload_timeout_wait_result_setting, send_current_log_file_to_object_store,
     send_logs_to_object_store,
@@ -31,18 +30,19 @@ use windmill_common::ee::{maybe_renew_license_key_on_start, LICENSE_KEY_ID, LICE
 
 use windmill_common::{
     global_settings::{
-        BASE_URL_SETTING, BUNFIG_INSTALL_SCOPES_SETTING, CRITICAL_ERROR_CHANNELS_SETTING,
-        CUSTOM_TAGS_SETTING, DEFAULT_TAGS_PER_WORKSPACE_SETTING, DEFAULT_TAGS_WORKSPACES_SETTING,
-        ENV_SETTINGS, EXPOSE_DEBUG_METRICS_SETTING, EXPOSE_METRICS_SETTING,
-        EXTRA_PIP_INDEX_URL_SETTING, HUB_BASE_URL_SETTING, JOB_DEFAULT_TIMEOUT_SECS_SETTING,
-        JWT_SECRET_SETTING, KEEP_JOB_DIR_SETTING, LICENSE_KEY_SETTING, NPM_CONFIG_REGISTRY_SETTING,
-        OAUTH_SETTING, PIP_INDEX_URL_SETTING, REQUEST_SIZE_LIMIT_SETTING, CRITICAL_ALERT_MUTE_UI_SETTING,
-        REQUIRE_PREEXISTING_USER_FOR_OAUTH_SETTING, RETENTION_PERIOD_SECS_SETTING,
-        SAML_METADATA_SETTING, SCIM_TOKEN_SETTING, SMTP_SETTING, TIMEOUT_WAIT_RESULT_SETTING,
+        BASE_URL_SETTING, BUNFIG_INSTALL_SCOPES_SETTING, CRITICAL_ALERT_MUTE_UI_SETTING,
+        CRITICAL_ERROR_CHANNELS_SETTING, CUSTOM_TAGS_SETTING, DEFAULT_TAGS_PER_WORKSPACE_SETTING,
+        DEFAULT_TAGS_WORKSPACES_SETTING, ENV_SETTINGS, EXPOSE_DEBUG_METRICS_SETTING,
+        EXPOSE_METRICS_SETTING, EXTRA_PIP_INDEX_URL_SETTING, HUB_BASE_URL_SETTING,
+        JOB_DEFAULT_TIMEOUT_SECS_SETTING, JWT_SECRET_SETTING, KEEP_JOB_DIR_SETTING,
+        LICENSE_KEY_SETTING, NPM_CONFIG_REGISTRY_SETTING, OAUTH_SETTING, PIP_INDEX_URL_SETTING,
+        REQUEST_SIZE_LIMIT_SETTING, REQUIRE_PREEXISTING_USER_FOR_OAUTH_SETTING,
+        RETENTION_PERIOD_SECS_SETTING, SAML_METADATA_SETTING, SCIM_TOKEN_SETTING, SMTP_SETTING,
+        TIMEOUT_WAIT_RESULT_SETTING,
     },
     scripts::ScriptLang,
     stats_ee::schedule_stats,
-    utils::{hostname, rd_string, Mode},
+    utils::{hostname, rd_string, Mode, GIT_VERSION},
     worker::{reload_custom_tags_setting, HUB_CACHE_DIR, TMP_DIR, WORKER_GROUP},
     DB, METRICS_ENABLED,
 };
@@ -67,24 +67,24 @@ use windmill_worker::{
     get_hub_script_content_and_requirements, BUN_BUNDLE_CACHE_DIR, BUN_CACHE_DIR,
     BUN_DEPSTAR_CACHE_DIR, DENO_CACHE_DIR, DENO_CACHE_DIR_DEPS, DENO_CACHE_DIR_NPM,
     GO_BIN_CACHE_DIR, GO_CACHE_DIR, LOCK_CACHE_DIR, PIP_CACHE_DIR, POWERSHELL_CACHE_DIR,
-    RUST_CACHE_DIR, TAR_PIP_CACHE_DIR, TMP_LOGS_DIR, UV_CACHE_DIR,
+    PY311_CACHE_DIR, RUST_CACHE_DIR, TAR_PIP_CACHE_DIR, TMP_LOGS_DIR, UV_CACHE_DIR,
 };
 
 use crate::monitor::{
     initial_load, load_keep_job_dir, load_metrics_debug_enabled, load_require_preexisting_user,
     load_tag_per_workspace_enabled, load_tag_per_workspace_workspaces, monitor_db, monitor_pool,
     reload_base_url_setting, reload_bunfig_install_scopes_setting,
-    reload_critical_error_channels_setting, reload_extra_pip_index_url_setting,
-    reload_hub_base_url_setting, reload_job_default_timeout_setting, reload_jwt_secret_setting,
-    reload_license_key, reload_npm_config_registry_setting, reload_pip_index_url_setting,
+    reload_critical_alert_mute_ui_setting, reload_critical_error_channels_setting,
+    reload_extra_pip_index_url_setting, reload_hub_base_url_setting,
+    reload_job_default_timeout_setting, reload_jwt_secret_setting, reload_license_key,
+    reload_npm_config_registry_setting, reload_pip_index_url_setting,
     reload_retention_period_setting, reload_scim_token_setting, reload_smtp_config,
-    reload_worker_config, reload_critical_alert_mute_ui_setting,
+    reload_worker_config,
 };
 
 #[cfg(feature = "parquet")]
 use crate::monitor::reload_s3_cache_setting;
 
-const GIT_VERSION: &str = git_version!(args = ["--tag", "--always"], fallback = "unknown-version");
 const DEFAULT_NUM_WORKERS: usize = 1;
 const DEFAULT_PORT: u16 = 8000;
 const DEFAULT_SERVER_BIND_ADDR: Ipv4Addr = Ipv4Addr::new(0, 0, 0, 0);
@@ -997,8 +997,9 @@ pub async fn run_workers<R: rsmq_async::RsmqConnection + Send + Sync + Clone + '
     for x in [
         LOCK_CACHE_DIR,
         TMP_LOGS_DIR,
-        PIP_CACHE_DIR,
         UV_CACHE_DIR,
+        PY311_CACHE_DIR,
+        PIP_CACHE_DIR,
         TAR_PIP_CACHE_DIR,
         DENO_CACHE_DIR,
         DENO_CACHE_DIR_DEPS,

backend/src/monitor.rs

@@ -51,8 +51,8 @@ use windmill_common::{
     utils::{now_from_db, rd_string, report_critical_error, Mode},
     worker::{
         load_worker_config, make_pull_query, make_suspended_pull_query, reload_custom_tags_setting,
-        DEFAULT_TAGS_PER_WORKSPACE, DEFAULT_TAGS_WORKSPACES, SMTP_CONFIG, WORKER_CONFIG,
-        WORKER_GROUP,
+        update_min_version, DEFAULT_TAGS_PER_WORKSPACE, DEFAULT_TAGS_WORKSPACES, SMTP_CONFIG,
+        WORKER_CONFIG, WORKER_GROUP,
     },
     BASE_URL, CRITICAL_ERROR_CHANNELS, DB, DEFAULT_HUB_BASE_URL, HUB_BASE_URL, JOB_RETENTION_SECS,
     METRICS_DEBUG_ENABLED, METRICS_ENABLED, CRITICAL_ALERT_MUTE_UI_ENABLED
@@ -1080,6 +1080,10 @@ pub async fn monitor_db(
         }
     };
 
+    let update_min_worker_version_f = async {
+        update_min_version(db).await;
+    };
+
     join!(
         expired_items_f,
         zombie_jobs_f,
@@ -1088,6 +1092,7 @@ pub async fn monitor_db(
         worker_groups_alerts_f,
         jobs_waiting_alerts_f,
         apply_autoscaling_f,
+        update_min_worker_version_f,
     );
 }
 

backend/windmill-api/openapi.yaml

@@ -283,6 +283,71 @@ paths:
             text/plain:
               schema:
                 type: string
+                
+  /users/set_password_of/{user}:
+    post:
+      summary: set password for a specific user (require super admin)
+      operationId: setPasswordForUser
+      tags:
+        - user
+      parameters:
+        - name: user
+          in: path
+          required: true
+          schema:
+            type: string
+      requestBody:
+        description: set password
+        required: true
+        content:
+          application/json:
+            schema:
+              type: object
+              properties:
+                password:
+                  type: string
+              required:
+                - password
+      responses:
+        "200":
+          description: password set
+          content:
+            text/plain:
+              schema:
+                type: string
+
+  /users/set_login_type/{user}:
+    post:
+      summary: set login type for a specific user (require super admin)
+      operationId: setLoginTypeForUser
+      tags:
+        - user
+      parameters:
+        - name: user
+          in: path
+          required: true
+          schema:
+            type: string
+      requestBody:
+        description: set login type
+        required: true
+        content:
+          application/json:
+            schema:
+              type: object
+              properties:
+                login_type:
+                  type: string
+              required:
+                - login_type
+      responses:
+        "200":
+          description: login type set
+          content:
+            text/plain:
+              schema:
+                type: string
+
 
   /users/create:
     post:
@@ -8188,6 +8253,23 @@ paths:
                     - id
                     - values
 
+  /workers/queue_counts:
+    get:
+      summary: get counts of jobs waiting for an executor per tag
+      operationId: getCountsOfJobsWaitingPerTag
+      tags:
+        - worker
+      responses:
+        "200":
+          description: queue counts
+          content:
+            application/json:
+              schema: 
+                type: object
+                additionalProperties:
+                  type: integer
+
+              
   /configs/list_worker_groups:
     get:
       summary: list worker groups

backend/windmill-api/src/jobs.rs

@@ -12,6 +12,7 @@ use quick_cache::sync::Cache;
 use serde_json::value::RawValue;
 use sqlx::Pool;
 use std::collections::HashMap;
+use std::ops::{Deref, DerefMut};
 #[cfg(feature = "prometheus")]
 use std::sync::atomic::Ordering;
 use tokio::io::AsyncReadExt;
@@ -600,7 +601,7 @@ async fn get_flow_job_debug_info(
     Extension(db): Extension<DB>,
     Path((w_id, id)): Path<(String, Uuid)>,
 ) -> error::Result<Response> {
-    let job = get_queued_job(&id, w_id.as_str(), &db).await?;
+    let job = get_queued_job_ex(&db, &w_id, id, false, None).await?;
     if let Some(job) = job {
         let is_flow = job.is_flow();
         if job.is_flow_step || !is_flow {
@@ -734,6 +735,67 @@ fn generate_get_job_query(no_logs: bool, table: &str) -> String {
     {join}
     WHERE id = $1 AND {table}.workspace_id = $2");
 }
+pub async fn get_queued_job_ex(
+    db: &DB,
+    workspace_id: &str,
+    job_id: Uuid,
+    no_logs: bool,
+    // first optional is if authed need to be checked, second is the opt_authed itself
+    opt_authed: Option<&Option<ApiAuthed>>,
+) -> error::Result<Option<JobExtended<QueuedJob>>> {
+    let query = if no_logs { &*GET_QUEUED_JOB_QUERY_NO_LOGS } else { &*GET_QUEUED_JOB_QUERY };
+    let job = sqlx::query_as::<_, JobExtended<QueuedJob>>(query)
+        .bind(job_id)
+        .bind(workspace_id)
+        .fetch_optional(db)
+        .await?;
+
+    if let Some(job) = job.as_ref() {
+        if opt_authed.is_some_and(|x| x.is_none()) && job.created_by != "anonymous" {
+            return Err(Error::BadRequest(
+                "As a non logged in user, you can only see jobs ran by anonymous users".to_string(),
+            ));
+        }
+    }
+
+    Ok(job)
+}
+pub async fn get_completed_job_ex(
+    db: &DB,
+    workspace_id: &str,
+    job_id: Uuid,
+    no_logs: bool,
+    // first optional is if authed need to be checked, second is the opt_authed itself
+    opt_authed: Option<&Option<ApiAuthed>>,
+) -> error::Result<Option<JobExtended<CompletedJob>>> {
+    let query = if no_logs { &*GET_COMPLETED_JOB_QUERY_NO_LOGS } else { &*GET_COMPLETED_JOB_QUERY };
+    let cjob = sqlx::query_as::<_, JobExtended<CompletedJob>>(query)
+        .bind(job_id)
+        .bind(workspace_id)
+        .fetch_optional(db)
+        .await?;
+
+    if let Some(job) = cjob.as_ref() {
+        if opt_authed.is_some_and(|x| x.is_none()) && job.created_by != "anonymous" {
+            return Err(Error::BadRequest(
+                "As a non logged in user, you can only see jobs ran by anonymous users".to_string(),
+            ));
+        }
+    }
+
+    if let Some(mut cjob) = cjob {
+        let CompletedJobWithFormattedResult { mut cj, result } = format_completed_job_result(cjob.inner);
+        cj.result = match result {
+            Some(FormattedResult::RawValue(rv)) => rv,
+            Some(FormattedResult::Vec(v)) => Some(to_raw_value(&v)),
+            None => None,
+        }.map(sqlx::types::Json);
+        cjob.inner = cj;
+        return Ok(Some(cjob));
+    }
+
+    Ok(cjob)
+}
 pub async fn get_job_internal(
     db: &DB,
     workspace_id: &str,
@@ -742,48 +804,18 @@ pub async fn get_job_internal(
     // first optional is if authed need to be checked, second is the opt_authed itself
     opt_authed: Option<&Option<ApiAuthed>>,
 ) -> error::Result<Job> {
-    let cjob_maybe = sqlx::query_as::<_, CompletedJob>(if no_logs {
-        &*GET_COMPLETED_JOB_QUERY_NO_LOGS
-    } else {
-        &*GET_COMPLETED_JOB_QUERY
-    })
-    .bind(job_id)
-    .bind(workspace_id)
-    .fetch_optional(db)
-    .await?
-    .map(Job::CompletedJob);
-
-    if let Some(cjob) = cjob_maybe {
-        Ok(match cjob {
-            Job::CompletedJob(cjob) => {
-                if opt_authed.is_some_and(|x| x.is_none()) && cjob.created_by != "anonymous" {
-                    return Err(Error::BadRequest(
-                        "As a non logged in user, you can only see jobs ran by anonymous users"
-                            .to_string(),
-                    ));
-                }
-                Job::CompletedJobWithFormattedResult(format_completed_job_result(cjob))
-            }
-            cjob => cjob,
-        })
-    } else {
-        let job_o = sqlx::query_as::<_, QueuedJob>(if no_logs {
-            &*GET_QUEUED_JOB_QUERY_NO_LOGS
-        } else {
-            &*GET_QUEUED_JOB_QUERY
-        })
-        .bind(job_id)
-        .bind(workspace_id)
-        .fetch_optional(db)
+    let cjob = get_completed_job_ex(db, workspace_id, job_id, no_logs, opt_authed.clone())
         .await?
-        .map(Job::QueuedJob);
-        let job: Job = not_found_if_none(job_o, "Job", job_id.to_string())?;
-        if opt_authed.is_some_and(|x| x.is_none()) && job.created_by() != "anonymous" {
-            return Err(Error::BadRequest(
-                "As a non logged in user, you can only see jobs ran by anonymous users".to_string(),
-            ));
+        .map(Job::CompletedJob);
+
+    match cjob {
+        Some(cjob) => Ok(cjob),
+        None => {
+            let job_maybe = get_queued_job_ex(db, workspace_id, job_id, no_logs, opt_authed)
+                .await?
+                .map(Job::QueuedJob);
+            not_found_if_none(job_maybe, "Job", job_id.to_string())
         }
-        Ok(job)
     }
 }
 
@@ -1735,7 +1767,6 @@ async fn resume_suspended_job_internal(
     let trigger_email = match &parent_flow {
         Job::CompletedJob(job) => &job.email,
         Job::QueuedJob(job) => &job.email,
-        Job::CompletedJobWithFormattedResult(job) => &job.cj.email,
     };
     conditionally_require_authed_user(authed.clone(), flow_status, trigger_email)?;
 
@@ -2010,7 +2041,6 @@ pub async fn get_suspended_job_flow(
     let trigger_email = match &flow {
         Job::CompletedJob(job) => &job.email,
         Job::QueuedJob(job) => &job.email,
-        Job::CompletedJobWithFormattedResult(job) => &job.cj.email,
     };
     conditionally_require_authed_user(authed.clone(), flow_status.clone(), trigger_email)?;
 
@@ -2225,13 +2255,45 @@ pub async fn get_resume_urls(
     Ok(Json(res))
 }
 
+#[derive(sqlx::FromRow, Debug, Serialize)]
+pub struct JobExtended<T> {
+    #[sqlx(flatten)]
+    #[serde(flatten)]
+    inner: T,
+
+    #[sqlx(skip)]
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub self_wait_time_ms: Option<i64>,
+    #[sqlx(skip)]
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub aggregate_wait_time_ms: Option<i64>,
+}
+
+impl<T> JobExtended<T> {
+    pub fn new(self_wait_time_ms: Option<i64>, aggregate_wait_time_ms: Option<i64>, inner: T) -> Self {
+        Self { inner, self_wait_time_ms, aggregate_wait_time_ms }
+    }
+}
+
+impl<T> Deref for JobExtended<T> {
+    type Target = T;
+
+    fn deref(&self) -> &Self::Target {
+        &self.inner
+    }
+}
+
+impl<T> DerefMut for JobExtended<T> {
+    fn deref_mut(&mut self) -> &mut Self::Target {
+        &mut self.inner
+    }
+}
+
 #[derive(Serialize, Debug)]
 #[serde(tag = "type")]
 pub enum Job {
-    QueuedJob(QueuedJob),
-    CompletedJob(CompletedJob),
-    #[serde(rename = "CompletedJob")]
-    CompletedJobWithFormattedResult(CompletedJobWithFormattedResult),
+    QueuedJob(JobExtended<QueuedJob>),
+    CompletedJob(JobExtended<CompletedJob>),
 }
 
 impl Job {
@@ -2239,27 +2301,6 @@ impl Job {
         match self {
             Job::QueuedJob(job) => &job.created_by,
             Job::CompletedJob(job) => &job.created_by,
-            Job::CompletedJobWithFormattedResult(job) => &job.cj.created_by,
-        }
-    }
-    pub fn raw_flow(&self) -> Option<FlowValue> {
-        match self {
-            Job::QueuedJob(job) => job
-                .raw_flow
-                .as_ref()
-                .map(|rf| serde_json::from_str(rf.0.get()).ok())
-                .flatten(),
-            Job::CompletedJob(job) => job
-                .raw_flow
-                .as_ref()
-                .map(|rf| serde_json::from_str(rf.0.get()).ok())
-                .flatten(),
-            Job::CompletedJobWithFormattedResult(job) => job
-                .cj
-                .raw_flow
-                .as_ref()
-                .map(|rf| serde_json::from_str(rf.0.get()).ok())
-                .flatten(),
         }
     }
 
@@ -2279,13 +2320,6 @@ impl Job {
                     job.logs = Some(logs.to_string());
                 }
             }
-            Job::CompletedJobWithFormattedResult(job) => {
-                if let Some(ref mut l) = job.cj.logs {
-                    l.push_str(logs);
-                } else {
-                    job.cj.logs = Some(logs.to_string());
-                }
-            }
         }
     }
 
@@ -2293,7 +2327,6 @@ impl Job {
         match self {
             Job::QueuedJob(job) => job.logs.as_ref().map(|l| l.len()),
             Job::CompletedJob(job) => job.logs.as_ref().map(|l| l.len()),
-            Job::CompletedJobWithFormattedResult(job) => job.cj.logs.as_ref().map(|l| l.len()),
         }
     }
 
@@ -2301,7 +2334,6 @@ impl Job {
         match self {
             Job::QueuedJob(job) => job.logs.clone(),
             Job::CompletedJob(job) => job.logs.clone(),
-            Job::CompletedJobWithFormattedResult(job) => job.cj.logs.clone(),
         }
     }
     pub fn flow_status(&self) -> Option<FlowStatus> {
@@ -2316,19 +2348,12 @@ impl Job {
                 .as_ref()
                 .map(|rf| serde_json::from_str(rf.0.get()).ok())
                 .flatten(),
-            Job::CompletedJobWithFormattedResult(job) => job
-                .cj
-                .flow_status
-                .as_ref()
-                .map(|rf| serde_json::from_str(rf.0.get()).ok())
-                .flatten(),
         }
     }
     pub fn is_flow_step(&self) -> bool {
         match self {
             Job::QueuedJob(job) => job.is_flow_step,
             Job::CompletedJob(job) => job.is_flow_step,
-            Job::CompletedJobWithFormattedResult(job) => job.cj.is_flow_step,
         }
     }
 
@@ -2343,7 +2368,6 @@ impl Job {
         match self {
             Job::QueuedJob(job) => &job.job_kind,
             Job::CompletedJob(job) => &job.job_kind,
-            Job::CompletedJobWithFormattedResult(job) => &job.cj.job_kind,
         }
     }
 
@@ -2351,7 +2375,6 @@ impl Job {
         match self {
             Job::QueuedJob(job) => job.id,
             Job::CompletedJob(job) => job.id,
-            Job::CompletedJobWithFormattedResult(job) => job.cj.id,
         }
     }
 
@@ -2359,7 +2382,6 @@ impl Job {
         match self {
             Job::QueuedJob(job) => &job.workspace_id,
             Job::CompletedJob(job) => &job.workspace_id,
-            Job::CompletedJobWithFormattedResult(job) => &job.cj.workspace_id,
         }
     }
 
@@ -2367,7 +2389,6 @@ impl Job {
         match self {
             Job::QueuedJob(job) => job.script_path.as_ref(),
             Job::CompletedJob(job) => job.script_path.as_ref(),
-            Job::CompletedJobWithFormattedResult(job) => job.cj.script_path.as_ref(),
         }
         .map(String::as_str)
         .unwrap_or("tmp/main")
@@ -2377,7 +2398,6 @@ impl Job {
         match self {
             Job::QueuedJob(job) => job.args.as_ref(),
             Job::CompletedJob(job) => job.args.as_ref(),
-            Job::CompletedJobWithFormattedResult(job) => job.cj.args.as_ref(),
         }
     }
 
@@ -2426,10 +2446,6 @@ impl Job {
                 job.self_wait_time_ms = self_wait_time;
                 job.aggregate_wait_time_ms = aggregate_wait_time;
             }
-            Job::CompletedJobWithFormattedResult(job) => {
-                job.cj.self_wait_time_ms = self_wait_time;
-                job.cj.aggregate_wait_time_ms = aggregate_wait_time;
-            }
         }
         Ok(())
     }
@@ -2557,86 +2573,82 @@ impl UnifiedJob {
 impl<'a> From<UnifiedJob> for Job {
     fn from(uj: UnifiedJob) -> Self {
         match uj.typ.as_ref() {
-            "CompletedJob" => Job::CompletedJob(CompletedJob {
-                workspace_id: uj.workspace_id,
-                id: uj.id,
-                parent_job: uj.parent_job,
-                created_by: uj.created_by,
-                created_at: uj.created_at,
-                started_at: uj.started_at.unwrap_or(uj.created_at),
-                duration_ms: uj.duration_ms.unwrap(),
-                success: uj.success.unwrap(),
-                script_hash: uj.script_hash,
-                script_path: uj.script_path,
-                args: None,
-                result: None,
-                logs: None,
-                flow_status: None,
-                deleted: uj.deleted,
-                canceled: uj.canceled,
-                canceled_by: uj.canceled_by,
-                raw_code: None,
-                canceled_reason: None,
-                job_kind: uj.job_kind,
-                schedule_path: uj.schedule_path,
-                permissioned_as: uj.permissioned_as,
-                raw_flow: None,
-                is_flow_step: uj.is_flow_step,
-                language: uj.language,
-                is_skipped: uj.is_skipped,
-                email: uj.email,
-                visible_to_owner: uj.visible_to_owner,
-                mem_peak: uj.mem_peak,
-                tag: uj.tag,
-                priority: uj.priority,
-                labels: uj.labels,
-                self_wait_time_ms: uj.self_wait_time_ms,
-                aggregate_wait_time_ms: uj.aggregate_wait_time_ms,
-            }),
-            "QueuedJob" => Job::QueuedJob(QueuedJob {
-                workspace_id: uj.workspace_id,
-                id: uj.id,
-                parent_job: uj.parent_job,
-                created_by: uj.created_by,
-                created_at: uj.created_at,
-                started_at: uj.started_at,
-                script_hash: uj.script_hash,
-                script_path: uj.script_path,
-                args: None,
-                running: uj.running.unwrap(),
-                scheduled_for: uj.scheduled_for.unwrap(),
-                logs: None,
-                flow_status: None,
-                raw_code: None,
-                raw_lock: None,
-                canceled: uj.canceled,
-                canceled_by: uj.canceled_by,
-                canceled_reason: None,
-                last_ping: None,
-                job_kind: uj.job_kind,
-                schedule_path: uj.schedule_path,
-                permissioned_as: uj.permissioned_as,
-                raw_flow: None,
-                is_flow_step: uj.is_flow_step,
-                language: uj.language,
-                same_worker: false,
-                pre_run_error: None,
-                email: uj.email,
-                visible_to_owner: uj.visible_to_owner,
-                suspend: uj.suspend,
-                mem_peak: uj.mem_peak,
-                root_job: None,
-                leaf_jobs: None,
-                tag: uj.tag,
-                concurrent_limit: uj.concurrent_limit,
-                concurrency_time_window_s: uj.concurrency_time_window_s,
-                timeout: None,
-                flow_step_id: None,
-                cache_ttl: None,
-                priority: uj.priority,
-                self_wait_time_ms: uj.self_wait_time_ms,
-                aggregate_wait_time_ms: uj.aggregate_wait_time_ms,
-            }),
+            "CompletedJob" => Job::CompletedJob(JobExtended::new(uj.self_wait_time_ms, uj.aggregate_wait_time_ms, CompletedJob {
+                    workspace_id: uj.workspace_id,
+                    id: uj.id,
+                    parent_job: uj.parent_job,
+                    created_by: uj.created_by,
+                    created_at: uj.created_at,
+                    started_at: uj.started_at.unwrap_or(uj.created_at),
+                    duration_ms: uj.duration_ms.unwrap(),
+                    success: uj.success.unwrap(),
+                    script_hash: uj.script_hash,
+                    script_path: uj.script_path,
+                    args: None,
+                    result: None,
+                    logs: None,
+                    flow_status: None,
+                    deleted: uj.deleted,
+                    canceled: uj.canceled,
+                    canceled_by: uj.canceled_by,
+                    raw_code: None,
+                    canceled_reason: None,
+                    job_kind: uj.job_kind,
+                    schedule_path: uj.schedule_path,
+                    permissioned_as: uj.permissioned_as,
+                    raw_flow: None,
+                    is_flow_step: uj.is_flow_step,
+                    language: uj.language,
+                    is_skipped: uj.is_skipped,
+                    email: uj.email,
+                    visible_to_owner: uj.visible_to_owner,
+                    mem_peak: uj.mem_peak,
+                    tag: uj.tag,
+                    priority: uj.priority,
+                    labels: uj.labels,
+            })),
+            "QueuedJob" => Job::QueuedJob(JobExtended::new(uj.self_wait_time_ms, uj.aggregate_wait_time_ms, QueuedJob {
+                    workspace_id: uj.workspace_id,
+                    id: uj.id,
+                    parent_job: uj.parent_job,
+                    created_by: uj.created_by,
+                    created_at: uj.created_at,
+                    started_at: uj.started_at,
+                    script_hash: uj.script_hash,
+                    script_path: uj.script_path,
+                    args: None,
+                    running: uj.running.unwrap(),
+                    scheduled_for: uj.scheduled_for.unwrap(),
+                    logs: None,
+                    flow_status: None,
+                    raw_code: None,
+                    raw_lock: None,
+                    canceled: uj.canceled,
+                    canceled_by: uj.canceled_by,
+                    canceled_reason: None,
+                    last_ping: None,
+                    job_kind: uj.job_kind,
+                    schedule_path: uj.schedule_path,
+                    permissioned_as: uj.permissioned_as,
+                    raw_flow: None,
+                    is_flow_step: uj.is_flow_step,
+                    language: uj.language,
+                    same_worker: false,
+                    pre_run_error: None,
+                    email: uj.email,
+                    visible_to_owner: uj.visible_to_owner,
+                    suspend: uj.suspend,
+                    mem_peak: uj.mem_peak,
+                    root_job: None,
+                    leaf_jobs: None,
+                    tag: uj.tag,
+                    concurrent_limit: uj.concurrent_limit,
+                    concurrency_time_window_s: uj.concurrency_time_window_s,
+                    timeout: None,
+                    flow_step_id: None,
+                    cache_ttl: None,
+                    priority: uj.priority,
+            })),
             t => panic!("job type {} not valid", t),
         }
     }
@@ -3079,7 +3091,7 @@ pub async fn run_workflow_as_code(
         i += 1;
     }
 
-    let job = get_queued_job(&job_id, &w_id, &db).await?;
+    let job = get_queued_job_ex(&db, &w_id, job_id, true, None).await?;
 
     if *CLOUD_HOSTED {
         tracing::info!("workflow_as_code_tracing id {i} ");
@@ -3087,6 +3099,7 @@ pub async fn run_workflow_as_code(
     }
 
     let job = not_found_if_none(job, "Queued Job", &job_id.to_string())?;
+    let JobExtended { inner: job, .. } = job;
     let (job_payload, tag, _delete_after_use, timeout) = match job.job_kind {
         JobKind::Preview => (
             JobPayload::Code(RawCode {
@@ -5058,24 +5071,10 @@ 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_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,
-    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")
-        .bind(id)
-        .bind(&w_id)
-        .fetch_optional(&db)
+    let job_o = get_completed_job_ex(&db, &w_id, id, false, Some(&opt_authed))
         .await?;
 
     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(
-            "As a non logged in user, you can only see jobs ran by anonymous users".to_string(),
-        ));
-    }
-    let cj = format_completed_job_result(cj);
-
     let response = Json(cj).into_response();
     // let extra_log = query_scalar!(
     //     "SELECT substr(logs, $1) as logs FROM large_logs WHERE workspace_id = $2 AND job_id = $3",

backend/windmill-api/src/lib.rs

@@ -25,7 +25,6 @@ use argon2::Argon2;
 use axum::extract::DefaultBodyLimit;
 use axum::{middleware::from_extractor, routing::get, Extension, Router};
 use db::DB;
-use git_version::git_version;
 use http::HeaderValue;
 use reqwest::Client;
 use std::collections::HashMap;
@@ -40,7 +39,7 @@ use tower_http::{
 };
 use windmill_common::db::UserDB;
 use windmill_common::worker::{ALL_TAGS, CLOUD_HOSTED};
-use windmill_common::{BASE_URL, INSTANCE_NAME};
+use windmill_common::{BASE_URL, INSTANCE_NAME, utils::GIT_VERSION};
 
 use crate::scim_ee::has_scim_token;
 use windmill_common::error::AppError;
@@ -93,9 +92,6 @@ mod workers;
 mod workspaces;
 mod workspaces_ee;
 
-pub const GIT_VERSION: &str =
-    git_version!(args = ["--tag", "--always"], fallback = "unknown-version");
-
 pub const DEFAULT_BODY_LIMIT: usize = 2097152 * 100; // 200MB
 
 lazy_static::lazy_static! {

backend/windmill-api/src/users.rs

@@ -90,6 +90,9 @@ pub fn global_service() -> Router {
         .route("/accept_invite", post(accept_invite))
         .route("/list_as_super_admin", get(list_users_as_super_admin))
         .route("/setpassword", post(set_password))
+        .route("/set_password_of/:user", post(set_password_of_user))
+        .route("/set_login_type/:user", post(set_login_type))
+
         .route("/create", post(create_user))
         .route("/update/:user", post(update_user))
         .route("/delete/:user", delete(delete_user))
@@ -866,6 +869,12 @@ pub struct EditPassword {
     pub password: String,
 }
 
+#[derive(Deserialize)]
+pub struct EditLoginType {
+    pub login_type: String,
+}
+
+
 #[derive(FromRow, Serialize)]
 pub struct TruncatedToken {
     pub label: Option<String>,
@@ -2028,7 +2037,52 @@ async fn set_password(
     authed: ApiAuthed,
     Json(ep): Json<EditPassword>,
 ) -> Result<String> {
-    crate::users_ee::set_password(db, argon2, authed, ep).await
+    let email = authed.email.clone();
+    crate::users_ee::set_password(db, argon2, authed, &email, ep).await
+}
+
+async fn set_password_of_user(
+    Extension(db): Extension<DB>,
+    Extension(argon2): Extension<Arc<Argon2<'_>>>,
+    Path(email): Path<String>,
+    authed: ApiAuthed,
+    Json(ep): Json<EditPassword>,
+) -> Result<String> {
+    require_super_admin(&db, &authed.email).await?;
+    crate::users_ee::set_password(db, argon2, authed, &email, ep).await
+}
+
+async fn set_login_type(
+    Extension(db): Extension<DB>,
+    Path(email): Path<String>,
+    authed: ApiAuthed,
+    Json(et): Json<EditLoginType>,
+) -> Result<String> {
+    require_super_admin(&db, &authed.email).await?;
+    let mut tx = db.begin().await?;
+
+    sqlx::query!(
+        "UPDATE password SET login_type = $1 WHERE email = $2",
+        et.login_type,
+        email
+    )
+    .execute(&mut *tx)
+    .await?;
+
+    audit_log(
+        &mut *tx,
+        &authed,
+        "users.set_login_type",
+        ActionKind::Update,
+        "global",
+        Some(&email),
+        None,
+    )
+    .await?;
+
+    tx.commit().await?;
+    Ok(format!("login type of {} updated to {}", email, et.login_type))
+    
 }
 
 async fn login(

backend/windmill-api/src/users_ee.rs

@@ -27,6 +27,7 @@ pub async fn set_password(
     _db: DB,
     _argon2: Arc<Argon2<'_>>,
     _authed: ApiAuthed,
+    _user_email: &str,
     _ep: EditPassword,
 ) -> Result<String> {
     Err(Error::InternalErr(

backend/windmill-api/src/workers.rs

@@ -36,6 +36,7 @@ pub fn global_service() -> Router {
         )
         .route("/get_default_tags", get(get_default_tags))
         .route("/queue_metrics", get(get_queue_metrics))
+        .route("/queue_counts", get(get_queue_counts))
 }
 
 #[derive(FromRow, Serialize, Deserialize)]
@@ -176,3 +177,12 @@ async fn get_queue_metrics(
 
     Ok(Json(queue_metrics))
 }
+
+async fn get_queue_counts(
+    authed: ApiAuthed,
+    Extension(db): Extension<DB>,
+) -> JsonResult<std::collections::HashMap<String, u32>> {
+    require_super_admin(&db, &authed.email).await?;
+    let queue_counts = windmill_common::queue::get_queue_counts(&db).await;
+    Ok(Json(queue_counts))
+}

backend/windmill-common/Cargo.toml

@@ -58,6 +58,7 @@ async-stream.workspace = true
 const_format.workspace = true
 crc.workspace = true
 windmill-macros.workspace = true
+semver.workspace = true
 
 [target.'cfg(not(target_env = "msvc"))'.dependencies]
 tikv-jemalloc-ctl = { optional = true, workspace = true }

backend/windmill-common/src/flows.rs

@@ -16,6 +16,7 @@ use rand::Rng;
 use serde::{Deserialize, Serialize, Serializer};
 
 use crate::{
+    error::Error,
     more_serde::{default_empty_string, default_id, default_null, default_true, is_default},
     scripts::{Schema, ScriptHash, ScriptLang},
 };
@@ -651,3 +652,29 @@ pub fn add_virtual_items_if_necessary(modules: &mut Vec<FlowModule>) {
         });
     }
 }
+
+pub async fn has_failure_module<'c>(flow: sqlx::types::Uuid, db: &sqlx::Pool<sqlx::Postgres>, completed: bool) -> Result<bool, Error> {
+    if completed {
+        sqlx::query_scalar!(
+            "SELECT raw_flow->'failure_module' != 'null'::jsonb
+            FROM completed_job
+            WHERE id = $1",
+            flow
+        )
+    } else {
+        sqlx::query_scalar!(
+            "SELECT raw_flow->'failure_module' != 'null'::jsonb
+            FROM queue
+            WHERE id = $1",
+            flow
+        )
+    }
+    .fetch_one(db)
+    .await
+    .map_err(|e| {
+        Error::InternalErr(format!(
+            "error during retrieval of has_failure_module: {e:#}"
+        ))
+    })
+    .map(|v| v.unwrap_or(false))
+}

backend/windmill-common/src/jobs.rs

@@ -108,13 +108,6 @@ pub struct QueuedJob {
     pub cache_ttl: Option<i32>,
     #[serde(skip_serializing_if = "Option::is_none")]
     pub priority: Option<i16>,
-
-    #[serde(skip_serializing_if = "Option::is_none")]
-    #[sqlx(skip)]
-    pub self_wait_time_ms: Option<i64>,
-    #[serde(skip_serializing_if = "Option::is_none")]
-    #[sqlx(skip)]
-    pub aggregate_wait_time_ms: Option<i64>,
 }
 
 impl QueuedJob {
@@ -198,8 +191,6 @@ impl Default for QueuedJob {
             flow_step_id: None,
             cache_ttl: None,
             priority: None,
-            self_wait_time_ms: None,
-            aggregate_wait_time_ms: None,
         }
     }
 }
@@ -253,13 +244,6 @@ pub struct CompletedJob {
     pub priority: Option<i16>,
     #[serde(skip_serializing_if = "Option::is_none")]
     pub labels: Option<serde_json::Value>,
-
-    #[serde(skip_serializing_if = "Option::is_none")]
-    #[sqlx(skip)]
-    pub self_wait_time_ms: Option<i64>,
-    #[serde(skip_serializing_if = "Option::is_none")]
-    #[sqlx(skip)]
-    pub aggregate_wait_time_ms: Option<i64>,
 }
 
 impl CompletedJob {

backend/windmill-common/src/utils.rs

@@ -21,6 +21,7 @@ use reqwest::Client;
 use serde::{Deserialize, Serialize};
 use sha2::{Digest, Sha256};
 use sqlx::{Pool, Postgres};
+use semver::Version;
 
 pub const MAX_PER_PAGE: usize = 10000;
 pub const DEFAULT_PER_PAGE: usize = 1000;
@@ -37,6 +38,10 @@ lazy_static::lazy_static! {
         .timeout(std::time::Duration::from_secs(20))
         .connect_timeout(std::time::Duration::from_secs(10))
         .build().unwrap();
+    pub static ref GIT_SEM_VERSION: Version = Version::parse(
+        // skip first `v` character.
+        GIT_VERSION.split_at(1).1
+    ).unwrap_or(Version::new(0, 1, 0));
 }
 
 #[derive(Deserialize, Clone)]

backend/windmill-common/src/worker.rs

@@ -1,6 +1,7 @@
 use const_format::concatcp;
 use itertools::Itertools;
 use regex::Regex;
+use semver::Version;
 use serde::{Deserialize, Serialize};
 use serde_json::value::RawValue;
 use std::{
@@ -88,6 +89,7 @@ lazy_static::lazy_static! {
     .and_then(|x| x.parse::<bool>().ok())
     .unwrap_or(false);
 
+    pub static ref MIN_VERSION: Arc<RwLock<Version>> = Arc::new(RwLock::new(Version::new(0, 0, 0)));
 }
 
 pub async fn make_suspended_pull_query(wc: &WorkerConfig) {
@@ -308,6 +310,8 @@ fn parse_file<T: FromStr>(path: &str) -> Option<T> {
 pub struct PythonAnnotations {
     pub no_cache: bool,
     pub no_uv: bool,
+    pub no_uv_install: bool,
+    pub no_uv_compile: bool,
 }
 
 #[annotations("//")]
@@ -548,6 +552,31 @@ pub fn get_windmill_memory_usage() -> Option<i64> {
     }
 }
 
+pub async fn update_min_version<'c, E: sqlx::Executor<'c, Database = sqlx::Postgres>>(executor: E) -> bool {
+    use crate::utils::{GIT_VERSION, GIT_SEM_VERSION};
+
+    // fetch all pings with a different version than self from the last 5 minutes.
+    let pings = sqlx::query_scalar!(
+        "SELECT wm_version FROM worker_ping WHERE wm_version != $1 AND ping_at > now() - interval '5 minutes'",
+        GIT_VERSION
+    ).fetch_all(executor).await.unwrap_or_default();
+
+    let cur_version = GIT_SEM_VERSION.clone();
+    let min_version = pings
+        .iter()
+        .filter(|x| !x.is_empty())
+        .filter_map(|x| semver::Version::parse(x.split_at(1).1).ok())
+        .min()
+        .unwrap_or_else(|| cur_version.clone());
+
+    if min_version != cur_version {
+        tracing::info!("Minimal worker version: {min_version}");
+    }
+
+    *MIN_VERSION.write().await = min_version.clone();
+    min_version >= cur_version
+}
+
 pub async fn update_ping(worker_instance: &str, worker_name: &str, ip: &str, db: &DB) {
     let (tags, dw) = {
         let wc = WORKER_CONFIG.read().await.clone();

backend/windmill-queue/src/jobs.rs

@@ -71,6 +71,9 @@ use windmill_common::BASE_URL;
 #[cfg(feature = "cloud")]
 use windmill_common::users::SUPERADMIN_SYNC_EMAIL;
 
+#[cfg(feature = "enterprise")]
+use windmill_common::flows::has_failure_module;
+
 #[cfg(feature = "enterprise")]
 use windmill_common::worker::CLOUD_HOSTED;
 
@@ -482,13 +485,6 @@ where
     }
 }
 
-#[cfg(feature = "enterprise")]
-#[derive(Deserialize)]
-struct RawFlowFailureModule {
-    #[cfg(feature = "enterprise")]
-    failure_module: Option<Box<RawValue>>,
-}
-
 #[instrument(level = "trace", skip_all)]
 pub async fn add_completed_job_error<R: rsmq_async::RsmqConnection + Clone + Send>(
     db: &Pool<Postgres>,
@@ -952,15 +948,7 @@ pub async fn add_completed_job<
         } else if !skip_downstream_error_handlers
             && (matches!(queued_job.job_kind, JobKind::Script)
                 || matches!(queued_job.job_kind, JobKind::Flow)
-                    && queued_job
-                        .raw_flow
-                        .as_ref()
-                        .and_then(|v| {
-                            serde_json::from_str::<RawFlowFailureModule>((**v).get())
-                                .ok()
-                                .and_then(|v| v.failure_module)
-                        })
-                        .is_none())
+                    && !has_failure_module(job_id, db, true).await.unwrap_or(false))
             && queued_job.parent_job.is_none()
         {
             let result = serde_json::from_str(

backend/windmill-worker/nsjail/download.py.pip.config.proto

@@ -0,0 +1,93 @@
+name: "python download pip"
+
+mode: ONCE
+hostname: "python"
+log_level: ERROR
+time_limit: 900
+
+rlimit_as: 2048
+rlimit_cpu: 1000
+rlimit_fsize: 1024
+rlimit_nofile: 64
+
+envar: "HOME=/user"
+envar: "LD_LIBRARY_PATH=/usr/local/lib:$LD_LIBRARY_PATH"
+
+cwd: "/tmp"
+
+clone_newnet: false
+clone_newuser: {CLONE_NEWUSER}
+
+keep_caps: true
+keep_env: true
+
+mount {
+    src: "/bin"
+    dst: "/bin"
+	is_bind: true
+}
+
+mount {
+    src: "/lib"
+    dst: "/lib"
+	is_bind: true
+}
+
+mount {
+    src: "/lib64"
+    dst: "/lib64"
+	is_bind: true
+    mandatory: false
+}
+
+mount {
+    src: "/usr"
+    dst: "/usr"
+	is_bind: true
+}
+
+mount {
+    src: "/etc"
+    dst: "/etc"
+	is_bind: true
+}
+
+mount {
+	src: "/dev/null"
+	dst: "/dev/null"
+	is_bind: true
+	rw: true
+}
+
+mount {
+	dst: "/tmp"
+	fstype: "tmpfs"
+	rw: true
+    options: "size=500000000"
+}
+
+
+mount {
+    src: "{WORKER_DIR}/download_deps.py.pip.sh"
+    dst: "/download_deps.sh"
+    is_bind: true
+}
+
+mount {
+    src: "{CACHE_DIR}"
+    dst: "{CACHE_DIR}"
+    is_bind: true
+    rw: true
+}
+
+mount {
+    src: "/dev/urandom"
+    dst: "/dev/urandom"
+    is_bind: true
+}
+
+exec_bin {
+ 	path: "/bin/sh"
+ 	arg: "/download_deps.sh"
+}
+

backend/windmill-worker/nsjail/download_deps.py.pip.sh

@@ -0,0 +1,24 @@
+#/bin/sh
+
+INDEX_URL_ARG=$([ -z "$INDEX_URL" ] && echo ""|| echo "--index-url $INDEX_URL" )
+EXTRA_INDEX_URL_ARG=$([ -z "$EXTRA_INDEX_URL" ] && echo ""|| echo "--extra-index-url $EXTRA_INDEX_URL" )
+TRUSTED_HOST_ARG=$([ -z "$TRUSTED_HOST" ] &&  echo "" || echo "--trusted-host $TRUSTED_HOST")
+
+if [ ! -z "$INDEX_URL" ]
+then
+      echo "\$INDEX_URL is set to $INDEX_URL"
+fi
+
+if [ ! -z "$EXTRA_INDEX_URL" ]
+then
+      echo "\$EXTRA_INDEX_URL is set to $EXTRA_INDEX_URL"
+fi
+
+if [ ! -z "$TRUSTED_HOST" ]
+then
+      echo "\$TRUSTED_HOST is set to $TRUSTED_HOST"
+fi
+
+CMD="/usr/local/bin/python3 -m pip install -v \"$REQ\" -I -t \"$TARGET\" --no-cache --no-color --no-deps --isolated --no-warn-conflicts --disable-pip-version-check $INDEX_URL_ARG $EXTRA_INDEX_URL_ARG $TRUSTED_HOST_ARG"
+echo $CMD
+eval $CMD

backend/windmill-worker/nsjail/download_deps.py.sh

@@ -19,6 +19,18 @@ then
       echo "\$TRUSTED_HOST is set to $TRUSTED_HOST"
 fi
 
-CMD="/usr/local/bin/python3 -m pip install -v \"$REQ\" -I -t \"$TARGET\" --no-cache --no-color --no-deps --isolated --no-warn-conflicts --disable-pip-version-check $INDEX_URL_ARG $EXTRA_INDEX_URL_ARG $TRUSTED_HOST_ARG"
+CMD="/usr/local/bin/uv pip install 
+\"$REQ\" 
+--target \"$TARGET\" 
+--no-cache 
+--no-config
+--no-color 
+--no-deps 
+--link-mode=copy
+$INDEX_URL_ARG $EXTRA_INDEX_URL_ARG $TRUSTED_HOST_ARG
+--index-strategy unsafe-best-match
+--system
+"
+
 echo $CMD
 eval $CMD

backend/windmill-worker/src/ansible_executor.rs

@@ -116,6 +116,8 @@ async fn handle_ansible_python_deps(
             job_dir,
             worker_dir,
             &mut Some(occupancy_metrics),
+            true,
+            true,
         )
         .await?;
         additional_python_paths.append(&mut venv_path);

backend/windmill-worker/src/global_cache.rs

@@ -20,13 +20,21 @@ use std::sync::Arc;
 pub async fn build_tar_and_push(
     s3_client: Arc<dyn ObjectStore>,
     folder: String,
+    no_uv: bool,
 ) -> error::Result<()> {
     use object_store::path::Path;
 
+    use crate::PY311_CACHE_DIR;
+
     tracing::info!("Started building and pushing piptar {folder}");
     let start = Instant::now();
     let folder_name = folder.split("/").last().unwrap();
-    let tar_path = format!("{PIP_CACHE_DIR}/{folder_name}_tar.tar",);
+    let prefix = if no_uv {
+        PIP_CACHE_DIR
+    } else {
+        PY311_CACHE_DIR
+    };
+    let tar_path = format!("{prefix}/{folder_name}_tar.tar",);
 
     let tar_file = std::fs::File::create(&tar_path)?;
     let mut tar = tar::Builder::new(tar_file);
@@ -46,7 +54,10 @@ pub async fn build_tar_and_push(
     // })?;
     if let Err(e) = s3_client
         .put(
-            &Path::from(format!("/tar/pip/{folder_name}.tar")),
+            &Path::from(format!(
+                "/tar/{}/{folder_name}.tar",
+                if no_uv { "pip" } else { "python_311" }
+            )),
             std::fs::read(&tar_path)?.into(),
         )
         .await
@@ -71,7 +82,11 @@ pub async fn build_tar_and_push(
 }
 
 #[cfg(all(feature = "enterprise", feature = "parquet"))]
-pub async fn pull_from_tar(client: Arc<dyn ObjectStore>, folder: String) -> error::Result<()> {
+pub async fn pull_from_tar(
+    client: Arc<dyn ObjectStore>,
+    folder: String,
+    no_uv: bool,
+) -> error::Result<()> {
     use windmill_common::s3_helpers::attempt_fetch_bytes;
 
     let folder_name = folder.split("/").last().unwrap();
@@ -79,7 +94,10 @@ pub async fn pull_from_tar(client: Arc<dyn ObjectStore>, folder: String) -> erro
     tracing::info!("Attempting to pull piptar {folder_name} from bucket");
 
     let start = Instant::now();
-    let tar_path = format!("tar/pip/{folder_name}.tar");
+    let tar_path = format!(
+        "tar/{}/{folder_name}.tar",
+        if no_uv { "pip" } else { "python_311" }
+    );
     let bytes = attempt_fetch_bytes(client, &tar_path).await?;
 
     // tracing::info!("B: {target} {folder}");

backend/windmill-worker/src/python_executor.rs

@@ -42,6 +42,10 @@ lazy_static::lazy_static! {
     static ref USE_PIP_COMPILE: bool = std::env::var("USE_PIP_COMPILE")
         .ok().map(|flag| flag == "true").unwrap_or(false);
 
+    /// Use pip install
+    static ref USE_PIP_INSTALL: bool = std::env::var("USE_PIP_INSTALL")
+        .ok().map(|flag| flag == "true").unwrap_or(false);
+
 
     static ref RELATIVE_IMPORT_REGEX: Regex = Regex::new(r#"(import|from)\s(((u|f)\.)|\.)"#).unwrap();
 
@@ -50,6 +54,8 @@ lazy_static::lazy_static! {
 }
 
 const NSJAIL_CONFIG_DOWNLOAD_PY_CONTENT: &str = include_str!("../nsjail/download.py.config.proto");
+const NSJAIL_CONFIG_DOWNLOAD_PY_CONTENT_FALLBACK: &str =
+    include_str!("../nsjail/download.py.pip.config.proto");
 const NSJAIL_CONFIG_RUN_PYTHON3_CONTENT: &str = include_str!("../nsjail/run.python3.config.proto");
 const RELATIVE_PYTHON_LOADER: &str = include_str!("../loader.py");
 
@@ -66,8 +72,8 @@ use crate::{
     },
     handle_child::handle_child,
     AuthedClientBackgroundTask, DISABLE_NSJAIL, DISABLE_NUSER, HOME_ENV, LOCK_CACHE_DIR,
-    NSJAIL_PATH, PATH_ENV, PIP_CACHE_DIR, PIP_EXTRA_INDEX_URL, PIP_INDEX_URL, PROXY_ENVS, TZ_ENV,
-    UV_CACHE_DIR,
+    NSJAIL_PATH, PATH_ENV, PIP_CACHE_DIR, PIP_EXTRA_INDEX_URL, PIP_INDEX_URL, PROXY_ENVS,
+    PY311_CACHE_DIR, TZ_ENV, UV_CACHE_DIR,
 };
 
 #[cfg(windows)]
@@ -901,10 +907,10 @@ async fn handle_python_deps(
         .unwrap_or_else(|| vec![])
         .clone();
 
+    let annotations = windmill_common::worker::PythonAnnotations::parse(inner_content);
     let requirements = match requirements_o {
         Some(r) => r,
         None => {
-            let annotation = windmill_common::worker::PythonAnnotations::parse(inner_content);
             let mut already_visited = vec![];
 
             let requirements = windmill_parser_py_imports::parse_python_imports(
@@ -929,8 +935,8 @@ async fn handle_python_deps(
                     worker_name,
                     w_id,
                     occupancy_metrics,
-                    annotation.no_uv,
-                    annotation.no_cache,
+                    annotations.no_uv || annotations.no_uv_compile,
+                    annotations.no_cache,
                 )
                 .await
                 .map_err(|e| {
@@ -955,6 +961,8 @@ async fn handle_python_deps(
             job_dir,
             worker_dir,
             occupancy_metrics,
+            annotations.no_uv || annotations.no_uv_install,
+            false,
         )
         .await?;
         additional_python_paths.append(&mut venv_path);
@@ -966,6 +974,7 @@ lazy_static::lazy_static! {
     static ref PIP_SECRET_VARIABLE: Regex = Regex::new(r"\$\{PIP_SECRET:([^\s\}]+)\}").unwrap();
 }
 
+/// pip install, include cached or pull from S3
 pub async fn handle_python_reqs(
     requirements: Vec<&str>,
     job_id: &Uuid,
@@ -977,12 +986,22 @@ pub async fn handle_python_reqs(
     job_dir: &str,
     worker_dir: &str,
     occupancy_metrics: &mut Option<&mut OccupancyMetrics>,
+    // TODO: Remove (Deprecated)
+    mut no_uv_install: bool,
+    is_ansible: bool,
 ) -> error::Result<Vec<String>> {
     let mut req_paths: Vec<String> = vec![];
     let mut vars = vec![("PATH", PATH_ENV.as_str())];
     let pip_extra_index_url;
     let pip_index_url;
 
+    no_uv_install |= *USE_PIP_INSTALL;
+
+    if no_uv_install && !is_ansible {
+        append_logs(&job_id, w_id, "\nFallback to pip (Deprecated!)\n", db).await;
+        tracing::warn!("Fallback to pip");
+    }
+
     if !*DISABLE_NSJAIL {
         pip_extra_index_url = PIP_EXTRA_INDEX_URL
             .read()
@@ -1013,10 +1032,21 @@ pub async fn handle_python_reqs(
         let _ = write_file(
             job_dir,
             "download.config.proto",
-            &NSJAIL_CONFIG_DOWNLOAD_PY_CONTENT
-                .replace("{WORKER_DIR}", &worker_dir)
-                .replace("{CACHE_DIR}", PIP_CACHE_DIR)
-                .replace("{CLONE_NEWUSER}", &(!*DISABLE_NUSER).to_string()),
+            &(if no_uv_install {
+                NSJAIL_CONFIG_DOWNLOAD_PY_CONTENT_FALLBACK
+            } else {
+                NSJAIL_CONFIG_DOWNLOAD_PY_CONTENT
+            })
+            .replace("{WORKER_DIR}", &worker_dir)
+            .replace(
+                "{CACHE_DIR}",
+                if no_uv_install {
+                    PIP_CACHE_DIR
+                } else {
+                    PY311_CACHE_DIR
+                },
+            )
+            .replace("{CLONE_NEWUSER}", &(!*DISABLE_NUSER).to_string()),
         )?;
     };
 
@@ -1026,8 +1056,14 @@ pub async fn handle_python_reqs(
         if req.starts_with('#') {
             continue;
         }
+        let py_prefix = if no_uv_install {
+            PIP_CACHE_DIR
+        } else {
+            PY311_CACHE_DIR
+        };
+
         let venv_p = format!(
-            "{PIP_CACHE_DIR}/{}",
+            "{py_prefix}/{}",
             req.replace(' ', "").replace('/', "").replace(':', "")
         );
         if metadata(&venv_p).await.is_ok() {
@@ -1073,7 +1109,10 @@ pub async fn handle_python_reqs(
                 .map(|(req, venv_p)| {
                     let os = os.clone();
                     async move {
-                        if pull_from_tar(os, venv_p.clone()).await.is_ok() {
+                        if pull_from_tar(os, venv_p.clone(), no_uv_install)
+                            .await
+                            .is_ok()
+                        {
                             PullFromTar::Pulled(venv_p.to_string())
                         } else {
                             PullFromTar::NotPulled(req.to_string(), venv_p.to_string())
@@ -1114,7 +1153,7 @@ pub async fn handle_python_reqs(
 
     for (req, venv_p) in req_with_penv {
         let mut logs1 = String::new();
-        logs1.push_str("\n\n--- PIP INSTALL ---\n");
+        logs1.push_str("\n\n--- UV PIP INSTALL ---\n");
         logs1.push_str(&format!("\n{req} is being installed for the first time.\n It will be cached for all ulterior uses."));
         append_logs(&job_id, w_id, logs1, db).await;
 
@@ -1150,21 +1189,47 @@ pub async fn handle_python_reqs(
             #[cfg(windows)]
             let req = format!("{}", req);
 
-            let mut command_args = vec![
-                PYTHON_PATH.as_str(),
-                "-m",
-                "pip",
-                "install",
-                &req,
-                "-I",
-                "--no-deps",
-                "--no-color",
-                "--isolated",
-                "--no-warn-conflicts",
-                "--disable-pip-version-check",
-                "-t",
-                venv_p.as_str(),
-            ];
+            let mut command_args = if no_uv_install {
+                vec![
+                    PYTHON_PATH.as_str(),
+                    "-m",
+                    "pip",
+                    "install",
+                    &req,
+                    "-I",
+                    "--no-deps",
+                    "--no-color",
+                    "--isolated",
+                    "--no-warn-conflicts",
+                    "--disable-pip-version-check",
+                    "-t",
+                    venv_p.as_str(),
+                ]
+            } else {
+                vec![
+                    UV_PATH.as_str(),
+                    "pip",
+                    "install",
+                    &req,
+                    "--no-deps",
+                    "--no-color",
+                    // "-p",
+                    // "3.11",
+                    // Prevent uv from discovering configuration files.
+                    "--no-config",
+                    "--link-mode=copy",
+                    // TODO: Doublecheck it
+                    "--system",
+                    // Prefer main index over extra
+                    // https://docs.astral.sh/uv/pip/compatibility/#packages-that-exist-on-multiple-indexes
+                    // TODO: Use env variable that can be toggled from UI
+                    "--index-strategy",
+                    "unsafe-best-match",
+                    "--target",
+                    venv_p.as_str(),
+                    "--no-cache",
+                ]
+            };
             let pip_extra_index_url = PIP_EXTRA_INDEX_URL
                 .read()
                 .await
@@ -1196,7 +1261,7 @@ pub async fn handle_python_reqs(
 
             envs.push(("HOME", HOME_ENV.as_str()));
 
-            tracing::debug!("pip install command: {:?}", command_args);
+            tracing::debug!("uv pip install command: {:?}", command_args);
 
             #[cfg(unix)]
             {
@@ -1207,7 +1272,12 @@ pub async fn handle_python_reqs(
                     .envs(envs)
                     .args([
                         "-x",
-                        &format!("{}/pip-{}.lock", LOCK_CACHE_DIR, fssafe_req),
+                        &format!(
+                            "{}/{}-{}.lock",
+                            LOCK_CACHE_DIR,
+                            if no_uv_install { "pip" } else { "py311" },
+                            fssafe_req
+                        ),
                         "--command",
                         &command_args.join(" "),
                     ])
@@ -1218,16 +1288,20 @@ pub async fn handle_python_reqs(
 
             #[cfg(windows)]
             {
-                let mut pip_cmd = Command::new(PYTHON_PATH.as_str());
-                pip_cmd
-                    .env_clear()
+                let installer_path = if no_uv_install { command_args[0] } else { "uv" };
+                let mut cmd: Command = Command::new(&installer_path);
+                cmd.env_clear()
                     .envs(envs)
                     .envs(PROXY_ENVS.clone())
                     .env("SystemRoot", SYSTEM_ROOT.as_str())
+                    .env(
+                        "TMP",
+                        std::env::var("TMP").unwrap_or_else(|_| String::from("/tmp")),
+                    )
                     .args(&command_args[1..])
                     .stdout(Stdio::piped())
                     .stderr(Stdio::piped());
-                start_child_process(pip_cmd, PYTHON_PATH.as_str()).await?
+                start_child_process(cmd, installer_path).await?
             }
         };
 
@@ -1240,7 +1314,7 @@ pub async fn handle_python_reqs(
             false,
             worker_name,
             &w_id,
-            &format!("pip install {req}"),
+            &format!("uv pip install {req}"),
             None,
             false,
             occupancy_metrics,
@@ -1260,7 +1334,7 @@ pub async fn handle_python_reqs(
                 tracing::warn!("S3 cache not available in the pro plan");
             } else {
                 let venv_p = venv_p.clone();
-                tokio::spawn(build_tar_and_push(os, venv_p));
+                tokio::spawn(build_tar_and_push(os, venv_p, no_uv_install));
             }
         }
         req_paths.push(venv_p);

backend/windmill-worker/src/worker.rs

@@ -236,7 +236,14 @@ pub const TMP_LOGS_DIR: &str = concatcp!(TMP_DIR, "/logs");
 pub const ROOT_CACHE_NOMOUNT_DIR: &str = concatcp!(TMP_DIR, "/cache_nomount/");
 
 pub const LOCK_CACHE_DIR: &str = concatcp!(ROOT_CACHE_DIR, "lock");
+// Used as fallback now
 pub const PIP_CACHE_DIR: &str = concatcp!(ROOT_CACHE_DIR, "pip");
+
+// pub const PY310_CACHE_DIR: &str = concatcp!(ROOT_CACHE_DIR, "python_310");
+pub const PY311_CACHE_DIR: &str = concatcp!(ROOT_CACHE_DIR, "python_311");
+// pub const PY312_CACHE_DIR: &str = concatcp!(ROOT_CACHE_DIR, "python_312");
+// pub const PY313_CACHE_DIR: &str = concatcp!(ROOT_CACHE_DIR, "python_313");
+
 pub const UV_CACHE_DIR: &str = concatcp!(ROOT_CACHE_DIR, "uv");
 pub const TAR_PIP_CACHE_DIR: &str = concatcp!(ROOT_CACHE_DIR, "tar/pip");
 pub const DENO_CACHE_DIR: &str = concatcp!(ROOT_CACHE_DIR, "deno");
@@ -257,6 +264,7 @@ const NUM_SECS_PING: u64 = 5;
 const NUM_SECS_READINGS: u64 = 60;
 
 const INCLUDE_DEPS_PY_SH_CONTENT: &str = include_str!("../nsjail/download_deps.py.sh");
+const INCLUDE_DEPS_PY_SH_CONTENT_FALLBACK: &str = include_str!("../nsjail/download_deps.py.pip.sh");
 
 pub const DEFAULT_CLOUD_TIMEOUT: u64 = 900;
 pub const DEFAULT_SELFHOSTED_TIMEOUT: u64 = 604800; // 7 days
@@ -311,6 +319,7 @@ lazy_static::lazy_static! {
     .and_then(|x| x.parse::<bool>().ok())
     .unwrap_or(false);
 
+    // pub static ref DISABLE_NSJAIL: bool = false;
     pub static ref DISABLE_NSJAIL: bool = std::env::var("DISABLE_NSJAIL")
         .ok()
         .and_then(|x| x.parse::<bool>().ok())
@@ -740,6 +749,13 @@ pub async fn run_worker<R: rsmq_async::RsmqConnection + Send + Sync + Clone + 's
             "download_deps.py.sh",
             INCLUDE_DEPS_PY_SH_CONTENT,
         );
+
+        // TODO: Remove (Deprecated)
+        let _ = write_file(
+            &worker_dir,
+            "download_deps.py.pip.sh",
+            INCLUDE_DEPS_PY_SH_CONTENT_FALLBACK,
+        );
     }
 
     let mut last_ping = Instant::now() - Duration::from_secs(NUM_SECS_PING + 1);
@@ -1903,8 +1919,10 @@ async fn handle_queued_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>(
         }
     };
     if job.is_flow() {
+        let flow = job.parse_raw_flow();
         handle_flow(
             job,
+            flow,
             db,
             &client.get_authed().await,
             None,

backend/windmill-worker/src/worker_flow.rs

@@ -8,7 +8,6 @@
 
 use std::collections::hash_map::DefaultHasher;
 use std::collections::HashMap;
-use std::hash::Hash;
 use std::sync::atomic::{AtomicUsize, Ordering};
 use std::sync::Arc;
 use std::time::Duration;
@@ -49,7 +48,7 @@ use windmill_common::{
         Approval, BranchAllStatus, BranchChosen, FlowStatus, FlowStatusModule, RetryStatus,
         MAX_RETRY_ATTEMPTS, MAX_RETRY_INTERVAL,
     },
-    flows::{FlowModule, FlowModuleValue, FlowValue, InputTransform, Retry, Suspend},
+    flows::{has_failure_module, FlowModule, FlowModuleValue, FlowValue, InputTransform, Retry, Suspend},
 };
 use windmill_queue::schedule::get_schedule_opt;
 use windmill_queue::{
@@ -177,11 +176,6 @@ struct RecoveryObject {
     recover: Option<bool>,
 }
 
-#[derive(sqlx::FromRow, Deserialize)]
-pub struct RowFlowStatus {
-    pub flow_status: sqlx::types::Json<Box<serde_json::value::RawValue>>,
-    pub current_module: Option<sqlx::types::Json<Box<serde_json::value::RawValue>>>,
-}
 // #[instrument(level = "trace", skip_all)]
 pub async fn update_flow_status_after_job_completion_internal<
     R: rsmq_async::RsmqConnection + Send + Sync + Clone,
@@ -207,6 +201,7 @@ pub async fn update_flow_status_after_job_completion_internal<
     let (
         should_continue_flow,
         flow_job,
+        flow_value,
         stop_early,
         skip_if_stop_early,
         nresult,
@@ -215,35 +210,28 @@ pub async fn update_flow_status_after_job_completion_internal<
     ) = {
         // tracing::debug!("UPDATE FLOW STATUS: {flow:?} {success} {result:?} {w_id} {depth}");
 
-        let old_status_json = sqlx::query_as::<_, RowFlowStatus>(
-            "SELECT flow_status, raw_flow->'modules'->(flow_status->'step')::int as current_module FROM queue WHERE id = $1 AND workspace_id = $2",
+        let (old_status, current_module) = sqlx::query!(
+            "SELECT
+                flow_status AS \"flow_status!: Json<Box<RawValue>>\",
+                raw_flow->'modules'->(flow_status->'step')::int AS \"module: Json<Box<RawValue>>\"
+            FROM queue WHERE id = $1 AND workspace_id = $2 LIMIT 1",
+            flow, w_id
         )
-        .bind(flow)
-        .bind(w_id)
         .fetch_one(db)
         .await
-        .map_err(|e| {
-            Error::InternalErr(format!(
-                "fetching flow status {flow} while reporting {success} {result:?}: {e:#}"
-            ))
-        })?;
-
-        let old_status = serde_json::from_str::<FlowStatus>(old_status_json.flow_status.get())
-            .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 {
-            Some(serde_json::from_str::<FlowModule>(x.0.get()).or_else(|e| {
-                Err(Error::InternalErr(format!(
+        .map_err(|e| Error::InternalErr(
+            format!("fetching flow status {flow} while reporting {success} {result:?}: {e:#}")
+        ))
+        .and_then(|record| Ok((
+            serde_json::from_str::<FlowStatus>(record.flow_status.0.get()).map_err(|e| Error::InternalErr(
+                format!("requiring current module to be parsable as FlowStatus: {e:?}")
+            ))?,
+            record.module.map(|json| {
+                serde_json::from_str::<FlowModule>(json.0.get()).map_err(|e| Error::InternalErr(format!(
                     "requiring current module to be parsable as FlowModule: {e:?}"
                 )))
-            })?)
-        } else {
-            None
-        };
+            }).transpose()?,
+        )))?;
 
         let module_step = Step::from_i32_and_len(old_status.step, old_status.modules.len());
 
@@ -303,16 +291,15 @@ pub async fn update_flow_status_after_job_completion_internal<
 
             let is_flow = if let Some(step) = step {
                 sqlx::query_scalar!(
-                    "SELECT raw_flow->'modules'->($1)->'value'->>'type' = 'flow' FROM queue WHERE id = $2",
-                    step as i32,
-                    &flow
+                    "SELECT raw_flow->'modules'->($1)::text->'value'->>'type' = 'flow' FROM queue WHERE id = $2 LIMIT 1",
+                    step as i32, flow
                 )
-                    .fetch_one(db)
-                    .await
-                    .map_err(|e| {
-                        Error::InternalErr(format!("error during retrieval of step's type: {e:#}"))
-                    })?
-                    .unwrap_or(false)
+                .fetch_one(db)
+                .await
+                .map_err(|e| {
+                    Error::InternalErr(format!("error during retrieval of step's type: {e:#}"))
+                })?
+                .unwrap_or(false)
             } else {
                 false
             };
@@ -938,10 +925,7 @@ pub async fn update_flow_status_after_job_completion_internal<
             .unwrap_or_else(|| "none".to_string());
         tracing::info!(id = %flow_job.id, root_id = %job_root, "update flow status");
 
-        let module = get_module(&flow_job, &module_step);
-        // tracing::error!(
-        //     "UPDATE FLOW STATUS 3: {module:#?} {unrecoverable} {} {is_last_step} {success} {skip_error_handler} is_failure_step {is_failure_step}", flow_job.canceled
-        // );
+        let flow_value = flow_job.parse_raw_flow();
 
         let should_continue_flow = match success {
             _ if stop_early => false,
@@ -953,7 +937,14 @@ pub async fn update_flow_status_after_job_completion_internal<
             }
             false
                 if next_retry(
-                    &module.and_then(|m| m.retry.clone()).unwrap_or_default(),
+                    flow_value
+                        .as_ref()
+                        .and_then(|value| match module_step {
+                            Step::PreprocessorStep => value.preprocessor_module.as_ref().and_then(|m| m.retry.as_ref()),
+                            Step::Step(i) => value.modules.get(i).as_ref().and_then(|m| m.retry.as_ref()),
+                            Step::FailureStep => value.failure_module.as_ref().and_then(|m| m.retry.as_ref()),
+                        })
+                        .unwrap_or(&Retry::default()),
                     &old_status.retry,
                 )
                 .is_some() =>
@@ -963,7 +954,7 @@ pub async fn update_flow_status_after_job_completion_internal<
             false
                 if !is_failure_step
                     && !skip_error_handler
-                    && has_failure_module(flow, db).await? =>
+                    && has_failure_module(flow, db, false).await? =>
             {
                 true
             }
@@ -975,6 +966,7 @@ pub async fn update_flow_status_after_job_completion_internal<
         (
             should_continue_flow,
             flow_job,
+            flow_value,
             stop_early,
             skip_if_stop_early,
             nresult,
@@ -1042,13 +1034,11 @@ pub async fn update_flow_status_after_job_completion_internal<
                     let args_hash =
                         hash_args(db, client, w_id, job_id_for_status, &flow_job.args).await;
                     let flow_path = flow_job.script_path();
-                    let version_hash = if let Some(rc) = flow_job.raw_flow.as_ref() {
-                        use std::hash::Hasher;
-                        let mut s = DefaultHasher::new();
-                        serde_json::to_string(&rc.0)
-                            .unwrap_or_default()
-                            .hash(&mut s);
-                        format!("flow_{}", hex::encode(s.finish().to_be_bytes()))
+                    let version_hash = if let Some(sqlx::types::Json(s)) = flow_job.raw_flow.as_ref() {
+                        use std::hash::{Hash, Hasher};
+                        let mut h = DefaultHasher::new();
+                        s.get().hash(&mut h);
+                        format!("flow_{}", hex::encode(h.finish().to_be_bytes()))
                     } else {
                         "flow_unknown".to_string()
                     };
@@ -1107,6 +1097,7 @@ pub async fn update_flow_status_after_job_completion_internal<
         tracing::debug!(id = %flow_job.id,  "start handle flow");
         match handle_flow(
             flow_job.clone(),
+            flow_value,
             db,
             client,
             Some(nresult.clone()),
@@ -1239,19 +1230,6 @@ async fn retrieve_flow_jobs_results(
     Ok(to_raw_value(&results))
 }
 
-fn get_module(flow_job: &QueuedJob, module_step: &Step) -> Option<FlowModule> {
-    let raw_flow = flow_job.parse_raw_flow();
-    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).map(|x| x.clone()),
-            Step::FailureStep => raw_flow.failure_module.map(|x| *x.clone()),
-        }
-    } else {
-        None
-    }
-}
-
 async fn compute_skip_branchall_failure<'c>(
     job: &Uuid,
     branch: usize,
@@ -1290,23 +1268,6 @@ async fn compute_skip_branchall_failure<'c>(
         }))
 }
 
-async fn has_failure_module<'c>(flow: Uuid, db: &DB) -> Result<bool, Error> {
-    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:#}"
-        ))
-    })
-    .map(|v| v.unwrap_or(false))
-}
-
 // async fn retrieve_cleanup_module<'c>(flow_uuid: Uuid, db: &DB) -> Result<FlowCleanupModule, Error> {
 //     tracing::warn!("Retrieving cleanup module of flow {}", flow_uuid);
 //     let raw_value = sqlx::query_scalar!(
@@ -1523,6 +1484,7 @@ async fn transform_input(
 #[instrument(level = "trace", skip_all)]
 pub async fn handle_flow<R: rsmq_async::RsmqConnection + Send + Sync + Clone>(
     flow_job: Arc<QueuedJob>,
+    flow_value: Option<FlowValue>,
     db: &sqlx::Pool<sqlx::Postgres>,
     client: &AuthedClient,
     last_result: Option<Arc<Box<RawValue>>>,
@@ -1531,8 +1493,7 @@ pub async fn handle_flow<R: rsmq_async::RsmqConnection + Send + Sync + Clone>(
     rsmq: Option<R>,
     job_completed_tx: Sender<SendResult>,
 ) -> anyhow::Result<()> {
-    let flow = flow_job
-        .parse_raw_flow()
+    let flow = flow_value
         .with_context(|| "Unable to parse flow definition")?;
     let status = flow_job
         .parse_flow_status()

backend/windmill-worker/src/worker_lockfiles.rs

@@ -1312,6 +1312,8 @@ async fn python_dep(
             job_dir,
             worker_dir,
             occupancy_metrics,
+            false,
+            false,
         )
         .await;
 

flake.nix

@@ -26,7 +26,10 @@
       ]);
     in {
       devShell = pkgs.mkShell {
-        buildInputs = buildInputs ++ (with pkgs; [ git go xcaddy sqlx-cli ]);
+        buildInputs = buildInputs ++ (with pkgs; [
+          git xcaddy sqlx-cli flock rust-analyzer
+          deno python3 python3Packages.pip go bun uv
+        ]);
         packages = [
           (pkgs.writeScriptBin "wm-caddy" ''
             cd ./frontend
@@ -56,6 +59,9 @@
             sqlx database create
             wm-migrate
           '')
+          (pkgs.writeScriptBin "wm-bench" ''
+            deno run -A benchmarks/main.ts -e admin@windmill.dev -p changeme $*
+          '')
           (pkgs.writeScriptBin "wm" ''
             cd ./frontend
             npm run dev $*
@@ -68,6 +74,12 @@
         DATABASE_URL = "postgres://postgres:changeme@127.0.0.1:5432/";
         REMOTE = "http://127.0.0.1:8000";
         REMOTE_LSP = "http://127.0.0.1:3001";
+        DENO_PATH = "${pkgs.deno}/bin/deno";
+        PYTHON_PATH = "${pkgs.python3}/bin/python3";
+        GO_PATH = "${pkgs.go}/bin/go";
+        BUN_PATH = "${pkgs.bun}/bin/bun";
+        UV_PATH = "${pkgs.uv}/bin/uv";
+        FLOCK_PATH = "${pkgs.flock}/bin/flock";
       };
       packages.default = self.packages.${system}.windmill;
       packages.windmill-client = pkgs.stdenv.mkDerivation {
@@ -82,7 +94,7 @@
           npm config set strict-ssl false
           cd frontend
           npm install --verbose
-          npm run generate-backend-client-mac
+          npm run ${if pkgs.stdenv.isDarwin then "generate-backend-client-mac" else "generate-backend-client"}
           NODE_OPTIONS="--max-old-space-size=8192" npm run build
         '';
 

frontend/src/lib/components/ChangeInstanceUsernameInner.svelte

@@ -72,11 +72,13 @@
 </script>
 
 <div class="flex flex-col max-w-2xl p-2">
+	{#if isConflict} 
 	<span class="text-sm mb-2 leading-6 font-semibold"
-		>{isConflict ? 'Fix username conflict' : 'Change username'}</span
+		>Fix username conflict</span
 	>
+	{/if}
 
-	<span class="text-xs mb-1 leading-6"
+	<span class="text-sm font-semibold mb-1 leading-6"
 		>{isConflict ? 'Auto-generated instance username' : 'New username'}</span
 	>
 	<input

frontend/src/lib/components/InstanceNameEditor.svelte

@@ -5,17 +5,36 @@
 	import Popup from './common/popup/Popup.svelte'
 	import { offset, flip, shift } from 'svelte-floating-ui/dom'
 	import ChangeInstanceUsernameInner from './ChangeInstanceUsernameInner.svelte'
+	import { UserService } from '$lib/gen'
+	import { sendUserToast } from '$lib/toast'
 
 	export let value: string | undefined
 	export let email: string
 	export let username: string | undefined = undefined
 	export let automateUsernameCreation: boolean = false
+	export let login_type: string
+
+	let password: string = ''
 
 	const dispatch = createEventDispatcher()
 
-	function save() {
+	function saveName() {
 		dispatch('save', value)
 	}
+	async function savePassword() {
+		if (password.length < 5) {
+			sendUserToast('Password must be at least 5 characters long', true)
+			return
+		}
+		await UserService.setPasswordForUser({user: email,  requestBody:{ password }})
+		sendUserToast(`Password updated for ${email}`)
+	}
+
+	async function saveLoginType() {
+		await UserService.setLoginTypeForUser({user: email,  requestBody:{ login_type }})
+		sendUserToast(`Login type updated for ${email}`)
+		dispatch('refresh')
+	}
 </script>
 
 <Popup
@@ -34,7 +53,7 @@
 			<ChangeInstanceUsernameInner {email} {username} on:renamed />
 		{/if}
 		<label class="block text-primary">
-			<div class="pb-1 text-xs text-secondary">Name</div>
+			<div class="pb-2 text-sm font-semibold text-primary">Name</div>
 			<div class="flex w-full">
 				<input
 					type="text"
@@ -44,7 +63,7 @@
 					on:keydown|stopPropagation
 					on:keypress|stopPropagation={({ key }) => {
 						if (key === 'Enter') {
-							save()
+							saveName()
 							close(null)
 						}
 					}}
@@ -57,12 +76,78 @@
 				btnClasses="mt-2 "
 				aria-label="Save ID"
 				on:click={() => {
-					save()
+					saveName()
 					close(null)
 				}}
 			>
 				Update name
 			</Button>
 		</label>
+		<label class="block text-primary">
+			<div class="pb-2 text-sm font-semibold text-primary">Password</div>
+			<div class="flex w-full">
+				<input
+					type="password"
+					bind:value={password}
+					class="!w-auto grow"
+					on:click|stopPropagation={() => {}}
+					on:keydown|stopPropagation
+					on:keypress|stopPropagation={({ key }) => {
+						if (key === 'Enter') {
+							savePassword()
+							close(null)
+						}
+					}}
+				/>
+			</div>
+			<Button
+				size="xs"
+				color="blue"
+				buttonType="button"
+				btnClasses="mt-2 "
+				aria-label="Save ID"
+				on:click={() => {
+					savePassword()
+					close(null)
+				}}
+			>
+				Update password
+			</Button>
+		</label>
+		<label class="block text-primary">
+			<div class="pb-2 text-sm font-semibold text-primary">Login type</div>
+			<div class="text-xs text-secondary mb-2">
+				Must match exact SSO name, "password" or "saml". Examples: password, google, saml, microsoft
+			</div>
+
+			<div class="flex w-full">
+				<input
+					type="text"
+					bind:value={login_type}
+					class="!w-auto grow"
+					on:click|stopPropagation={() => {}}
+					on:keydown|stopPropagation
+					on:keypress|stopPropagation={({ key }) => {
+						if (key === 'Enter') {
+							saveLoginType()
+							close(null)
+						}
+					}}
+				/>
+			</div>
+			<Button
+				size="xs"
+				color="blue"
+				buttonType="button"
+				btnClasses="mt-2 "
+				aria-label="Save login type"
+				on:click={() => {
+					saveLoginType()
+					close(null)
+				}}
+			>
+				Update login type
+			</Button>
+		</label>
 	</div>
 </Popup>

frontend/src/lib/components/SuperadminSettings.svelte

@@ -276,9 +276,13 @@
 														<td>
 															<div class="flex flex-row gap-x-1 justify-end">
 																<InstanceNameEditor
+																	login_type={login_type}
 																	value={name}
 																	{username}
 																	{email}
+																	on:refresh={() => {
+																		listUsers(activeOnly)
+																	}}
 																	on:save={(e) => {
 																		updateName(e.detail, email)
 																	}}