Merge branch 'main' into di/stream-sql-to-s3

Author: diegoimbert

.cursor/rules/rust-best-practices.mdc

@@ -3,10 +3,6 @@ description:
 globs: backend/**/*.rs
 alwaysApply: false
 ---
----
-description: Rust best practices for the Windmill backend, covering code organization, error handling, performance optimizations, and common patterns to follow when adding new code.
-globs: **/*.rs
----
 # Windmill Backend - Rust Best Practices
 
 ## Project Structure

backend/src/monitor.rs

@@ -28,10 +28,10 @@ use windmill_api::{
     SCIM_TOKEN,
 };
 
-#[cfg(feature = "enterprise")]
-use windmill_common::ee::{jobs_waiting_alerts, worker_groups_alerts};
 #[cfg(feature = "enterprise")]
 use windmill_common::ee::low_disk_alerts;
+#[cfg(feature = "enterprise")]
+use windmill_common::ee::{jobs_waiting_alerts, worker_groups_alerts};
 
 #[cfg(feature = "oauth2")]
 use windmill_common::global_settings::OAUTH_SETTING;
@@ -60,13 +60,9 @@ use windmill_common::{
     server::load_smtp_config,
     tracing_init::JSON_FMT,
     users::truncate_token,
-    utils::empty_string_as_none,
-    utils::{now_from_db, rd_string, report_critical_error, Mode},
+    utils::{empty_string_as_none, now_from_db, rd_string, report_critical_error, Mode},
     worker::{
-        load_worker_config, reload_custom_tags_setting, store_pull_query,
-        store_suspended_pull_query, update_min_version, Connection, DEFAULT_TAGS_PER_WORKSPACE,
-        DEFAULT_TAGS_WORKSPACES, INDEXER_CONFIG, SCRIPT_TOKEN_EXPIRY, SMTP_CONFIG, TMP_DIR,
-        WORKER_CONFIG, WORKER_GROUP,
+        load_env_vars, load_init_bash_from_env, load_whitelist_env_vars_from_env, load_worker_config, reload_custom_tags_setting, store_pull_query, store_suspended_pull_query, update_min_version, Connection, WorkerConfig, DEFAULT_TAGS_PER_WORKSPACE, DEFAULT_TAGS_WORKSPACES, INDEXER_CONFIG, SCRIPT_TOKEN_EXPIRY, SMTP_CONFIG, TMP_DIR, WORKER_CONFIG, WORKER_GROUP
     },
     KillpillSender, BASE_URL, CRITICAL_ALERTS_ON_DB_OVERSIZE, CRITICAL_ALERT_MUTE_UI_ENABLED,
     CRITICAL_ERROR_CHANNELS, DB, DEFAULT_HUB_BASE_URL, HUB_BASE_URL, JOB_RETENTION_SECS,
@@ -203,10 +199,23 @@ pub async fn initial_load(
             }
             Connection::Http(_) => {
                 // TODO: reload worker config from http
-                WORKER_CONFIG.write().await.worker_tags = DECODED_AGENT_TOKEN
-                    .as_ref()
-                    .map(|x| x.tags.clone())
-                    .unwrap_or_default();
+                let mut config = WORKER_CONFIG.write().await;
+                *config = WorkerConfig {
+                    worker_tags: DECODED_AGENT_TOKEN
+                        .as_ref()
+                        .map(|x| x.tags.clone())
+                        .unwrap_or_default(),
+                    env_vars: load_env_vars(
+                        load_whitelist_env_vars_from_env(),
+                        &std::collections::HashMap::new(),
+                    ),
+                    priority_tags_sorted: vec![],
+                    dedicated_worker: None,
+                    init_bash: load_init_bash_from_env(),
+                    cache_clear: None,
+                    additional_python_paths: None,
+                    pip_local_dependencies: None,
+                };
             }
         }
     }

backend/windmill-api/src/args.rs

@@ -355,7 +355,7 @@ where
         let uri = request.uri();
         let request_query = Query::<RequestQuery>::try_from_uri(uri).unwrap().0;
         let headers = build_headers(&headers_map, request_query.include_header, is_http_trigger);
-        let query_decode = DecodeQueries::from_uri(uri);
+        let query_decode = DecodeQueries::from_uri(uri, is_http_trigger);
         let mut query = HashMap::new();
         if let Some(DecodeQueries(queries)) = query_decode {
             query.extend(queries);
@@ -500,37 +500,49 @@ where
     type Rejection = Response;
 
     async fn from_request_parts(parts: &mut Parts, _state: &S) -> Result<Self, Self::Rejection> {
-        Ok(DecodeQueries::from_uri(&parts.uri).unwrap_or_else(|| DecodeQueries(HashMap::new())))
+        Ok(DecodeQueries::from_uri(&parts.uri, false)
+            .unwrap_or_else(|| DecodeQueries(HashMap::new())))
     }
 }
 
 impl DecodeQueries {
-    pub fn from_uri(uri: &Uri) -> Option<Self> {
+    pub fn from_uri(uri: &Uri, is_http_trigger: bool) -> Option<Self> {
         let query = uri.query();
         if query.is_none() {
             return None;
         }
         let query = query.unwrap();
-        let include_query = serde_urlencoded::from_str::<IncludeQuery>(query)
-            .map(|x| x.include_query)
-            .ok()
-            .flatten()
-            .unwrap_or_default();
-        let parse_query_args = include_query
-            .split(",")
-            .map(|s| s.to_string())
-            .collect::<Vec<_>>();
-        let mut args = HashMap::new();
-        if !parse_query_args.is_empty() {
+        if is_http_trigger {
             let queries =
                 serde_urlencoded::from_str::<HashMap<String, String>>(query).unwrap_or_default();
-            parse_query_args.iter().for_each(|h| {
-                if let Some(v) = queries.get(h) {
-                    args.insert(h.to_string(), to_raw_value(v));
-                }
-            });
+            Some(DecodeQueries(
+                queries
+                    .into_iter()
+                    .map(|(k, v)| (k, to_raw_value(&v)))
+                    .collect(),
+            ))
+        } else {
+            let include_query = serde_urlencoded::from_str::<IncludeQuery>(query)
+                .map(|x| x.include_query)
+                .ok()
+                .flatten()
+                .unwrap_or_default();
+            let parse_query_args = include_query
+                .split(",")
+                .map(|s| s.to_string())
+                .collect::<Vec<_>>();
+            let mut args = HashMap::new();
+            if !parse_query_args.is_empty() {
+                let queries = serde_urlencoded::from_str::<HashMap<String, String>>(query)
+                    .unwrap_or_default();
+                parse_query_args.iter().for_each(|h| {
+                    if let Some(v) = queries.get(h) {
+                        args.insert(h.to_string(), to_raw_value(v));
+                    }
+                });
+            }
+            Some(DecodeQueries(args))
         }
-        Some(DecodeQueries(args))
     }
 }
 

backend/windmill-api/src/http_trigger_args.rs

@@ -158,9 +158,10 @@ impl HttpTriggerArgs {
         wrap_body: bool,
     ) -> Result<PushArgsOwned, Error> {
         let mut extra = HashMap::new();
-
-        extra.insert(
-            "wm_trigger".to_string(),
+        let mut wm_trigger = HashMap::new();
+        wm_trigger.insert("kind".to_string(), to_raw_value(&"http".to_string()));
+        wm_trigger.insert(
+            "http".to_string(),
             to_raw_value(&HttpTriggerWmTrigger {
                 route: route_path,
                 path: called_path,
@@ -170,6 +171,7 @@ impl HttpTriggerArgs {
                 headers: &self.0.metadata.headers,
             }),
         );
+        extra.insert("wm_trigger".to_string(), to_raw_value(&wm_trigger));
 
         let mut args = self.to_main_args(wrap_body)?;
 

backend/windmill-common/src/error.rs

@@ -66,6 +66,8 @@ pub enum Error {
     DatabaseMigration(#[from] MigrateError),
     #[error("Non-zero exit status for {0}: {1}")]
     ExitStatus(String, i32),
+    #[error("ExecutionRawError: {0}")]
+    ExecutionRawError(Box<serde_json::value::RawValue>),
     #[error("Error: {error:#} @{location:#}")]
     Anyhow { error: anyhow::Error, location: String },
     #[error("Error: {0:#?}")]

backend/windmill-common/src/worker.rs

@@ -382,10 +382,7 @@ fn normalize_path(path: &Path) -> PathBuf {
     ret
 }
 
-pub fn is_allowed_file_location(
-    job_dir: &str,
-    user_defined_path: &str,
-) -> error::Result<PathBuf> {
+pub fn is_allowed_file_location(job_dir: &str, user_defined_path: &str) -> error::Result<PathBuf> {
     let job_dir = Path::new(job_dir);
     let user_path = PathBuf::from(user_defined_path);
 
@@ -1398,18 +1395,66 @@ pub async fn load_worker_config(
     tracing::debug!("Custom tags priority set: {:?}", priority_tags_sorted);
 
     let env_vars_static = config.env_vars_static.unwrap_or_default().clone();
-    let resolved_env_vars: HashMap<String, String> = env_vars_static
-        .keys()
-        .map(|x| x.to_string())
-        .chain(config.env_vars_allowlist.unwrap_or_default())
-        .chain(
-            std::env::var("WHITELIST_ENVS")
-                .ok()
-                .map(|x| x.split(',').map(|x| x.to_string()).collect_vec())
-                .unwrap_or_default()
-                .into_iter(),
-        )
-        .sorted()
+    let resolved_env_vars: HashMap<String, String> = load_env_vars(
+        config
+            .env_vars_allowlist
+            .unwrap_or_default()
+            .into_iter()
+            .chain(load_whitelist_env_vars_from_env())
+            .chain(env_vars_static.keys().map(|x| x.to_string())),
+        &env_vars_static,
+    );
+
+    Ok(WorkerConfig {
+        worker_tags,
+        priority_tags_sorted,
+        dedicated_worker,
+        init_bash: config
+            .init_bash
+            .or_else(|| load_init_bash_from_env())
+            .and_then(|x| if x.is_empty() { None } else { Some(x) }),
+        cache_clear: config.cache_clear,
+        pip_local_dependencies: config
+            .pip_local_dependencies
+            .or_else(|| load_pip_local_dependencies_from_env()),
+        additional_python_paths: config
+            .additional_python_paths
+            .or_else(|| load_additional_python_paths_from_env()),
+        env_vars: resolved_env_vars,
+    })
+}
+
+pub fn load_init_bash_from_env() -> Option<String> {
+    std::env::var("INIT_SCRIPT")
+        .ok()
+        .and_then(|x| if x.is_empty() { None } else { Some(x) })
+}
+
+pub fn load_pip_local_dependencies_from_env() -> Option<Vec<String>> {
+    std::env::var("PIP_LOCAL_DEPENDENCIES")
+        .ok()
+        .map(|x| x.split(',').map(|x| x.to_string()).collect_vec())
+}
+
+pub fn load_additional_python_paths_from_env() -> Option<Vec<String>> {
+    std::env::var("ADDITIONAL_PYTHON_PATHS")
+        .ok()
+        .map(|x| x.split(':').map(|x| x.to_string()).collect_vec())
+}
+
+pub fn load_whitelist_env_vars_from_env() -> std::vec::IntoIter<String> {
+    std::env::var("WHITELIST_ENVS")
+        .ok()
+        .map(|x| x.split(',').map(|x| x.to_string()).collect_vec())
+        .unwrap_or_default()
+        .into_iter()
+}
+
+pub fn load_env_vars(
+    iter: impl Iterator<Item = String>,
+    env_vars_static: &HashMap<String, String>,
+) -> HashMap<String, String> {
+    iter.sorted()
         .unique()
         .map(|envvar_name| {
             (
@@ -1422,34 +1467,7 @@ pub async fn load_worker_config(
                     }),
             )
         })
-        .collect();
-
-    Ok(WorkerConfig {
-        worker_tags,
-        priority_tags_sorted,
-        dedicated_worker,
-        init_bash: config
-            .init_bash
-            .or_else(|| std::env::var("INIT_SCRIPT").ok())
-            .and_then(|x| if x.is_empty() { None } else { Some(x) }),
-        cache_clear: config.cache_clear,
-        pip_local_dependencies: config.pip_local_dependencies.or_else(|| {
-            let pip_local_dependencies = std::env::var("PIP_LOCAL_DEPENDENCIES")
-                .ok()
-                .map(|x| x.split(',').map(|x| x.to_string()).collect());
-            if pip_local_dependencies == Some(vec!["".to_string()]) {
-                None
-            } else {
-                pip_local_dependencies
-            }
-        }),
-        additional_python_paths: config.additional_python_paths.or_else(|| {
-            std::env::var("ADDITIONAL_PYTHON_PATHS")
-                .ok()
-                .map(|x| x.split(':').map(|x| x.to_string()).collect())
-        }),
-        env_vars: resolved_env_vars,
-    })
+        .collect()
 }
 
 #[derive(Clone, PartialEq, Debug)]

backend/windmill-worker/src/bigquery_executor.rs

@@ -472,7 +472,7 @@ pub async fn do_bigquery(
         conn,
         mem_peak,
         canceled_by,
-        result_f.map_err(to_anyhow),
+        result_f,
         worker_name,
         &job.workspace_id,
         &mut Some(occupancy_metrics),

backend/windmill-worker/src/graphql_executor.rs

@@ -1,6 +1,5 @@
 use std::collections::HashMap;
 
-use anyhow::anyhow;
 use futures::{stream, TryStreamExt};
 use serde_json::{json, value::RawValue};
 use sqlx::types::Json;
@@ -134,11 +133,13 @@ pub async fn do_graphql(
         .map_err(|e| Error::ExecutionErr(e.to_string()))?;
 
         if let Some(errors) = result.errors {
-            return Err(anyhow!(errors
-                .into_iter()
-                .map(|x| x.message)
-                .collect::<Vec<_>>()
-                .join("\n"),));
+            return Err(Error::ExecutionErr(
+                errors
+                    .into_iter()
+                    .map(|x| x.message)
+                    .collect::<Vec<_>>()
+                    .join("\n"),
+            ));
         }
 
         // And then check that we got back the same string we sent over.

backend/windmill-worker/src/handle_child.rs

@@ -526,7 +526,7 @@ pub async fn run_future_with_polling_update_job_poller<Fut, T, S>(
     get_mem: S,
 ) -> error::Result<T>
 where
-    Fut: Future<Output = anyhow::Result<T>>,
+    Fut: Future<Output = windmill_common::error::Result<T>>,
     S: stream::Stream<Item = i32> + Unpin,
 {
     let (tx, rx) = broadcast::channel::<()>(3);