From c8915063c48dc89de7e0eee2fb855922a69f9be3 Mon Sep 17 00:00:00 2001 From: Ruben Fiszel Date: Sat, 8 Jun 2024 16:43:42 +0200 Subject: [PATCH] feat: track dependency map for bun + inline script of flows --- ...82fc2426a7abd277313dff326a667b956fdee.json | 17 + ...8174435d6df4f835cb0883ba40897cb99a410.json | 15 - ...46bee53a07c8d1264eeb44bc94233bc06bbfd.json | 16 + ...97ce7e470c6ba1905e7e877dca35d9d68968f.json | 23 - ...659cedf430fb7242faee4379b4a13e5fd763.json} | 6 +- ...9f61501068f4b027f58279e0c8508839607f4.json | 35 + .../20240607113729_add_importer_node.down.sql | 2 + .../20240607113729_add_importer_node.up.sql | 4 + backend/windmill-api/src/jobs.rs | 2 +- backend/windmill-queue/src/jobs.rs | 6 +- backend/windmill-worker/src/bash_executor.rs | 6 +- backend/windmill-worker/src/bun_executor.rs | 8 +- backend/windmill-worker/src/common.rs | 4 +- .../windmill-worker/src/dedicated_worker.rs | 475 ++++- backend/windmill-worker/src/deno_executor.rs | 2 +- backend/windmill-worker/src/go_executor.rs | 14 +- backend/windmill-worker/src/lib.rs | 1 + backend/windmill-worker/src/mssql_executor.rs | 2 +- backend/windmill-worker/src/php_executor.rs | 4 +- .../windmill-worker/src/python_executor.rs | 12 +- backend/windmill-worker/src/worker.rs | 1615 +---------------- backend/windmill-worker/src/worker_flow.rs | 8 +- .../windmill-worker/src/worker_lockfiles.rs | 1270 +++++++++++++ frontend/src/lib/ata/index.ts | 1 + frontend/src/lib/components/Editor.svelte | 5 +- .../flows/content/FlowModuleComponent.svelte | 2 +- 26 files changed, 1929 insertions(+), 1626 deletions(-) create mode 100644 backend/.sqlx/query-20db481797e151cb20e986d5aeb82fc2426a7abd277313dff326a667b956fdee.json delete mode 100644 backend/.sqlx/query-30e9250edd0c4c6025322f51b758174435d6df4f835cb0883ba40897cb99a410.json create mode 100644 backend/.sqlx/query-6c962f9471b0b1fe385a93789ec46bee53a07c8d1264eeb44bc94233bc06bbfd.json delete mode 100644 backend/.sqlx/query-aa44311f9bae8253c6c483d122597ce7e470c6ba1905e7e877dca35d9d68968f.json rename backend/.sqlx/{query-5fbd5dfb8df9aecce023c5d482ee58efcbc23c8682a3f0ec94977eb56023ed72.json => query-e87c317550af671e4b65752a9c0b659cedf430fb7242faee4379b4a13e5fd763.json} (52%) create mode 100644 backend/.sqlx/query-ec3359bbc309c2b893e9f68c09c9f61501068f4b027f58279e0c8508839607f4.json create mode 100644 backend/migrations/20240607113729_add_importer_node.down.sql create mode 100644 backend/migrations/20240607113729_add_importer_node.up.sql create mode 100644 backend/windmill-worker/src/worker_lockfiles.rs diff --git a/backend/.sqlx/query-20db481797e151cb20e986d5aeb82fc2426a7abd277313dff326a667b956fdee.json b/backend/.sqlx/query-20db481797e151cb20e986d5aeb82fc2426a7abd277313dff326a667b956fdee.json new file mode 100644 index 0000000000..42c56115cd --- /dev/null +++ b/backend/.sqlx/query-20db481797e151cb20e986d5aeb82fc2426a7abd277313dff326a667b956fdee.json @@ -0,0 +1,17 @@ +{ + "db_name": "PostgreSQL", + "query": "DELETE FROM dependency_map\n WHERE importer_path = $1 AND importer_kind = $3::text::IMPORTER_KIND\n AND workspace_id = $2 AND importer_node_id = $4", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Text", + "Text", + "Text", + "Text" + ] + }, + "nullable": [] + }, + "hash": "20db481797e151cb20e986d5aeb82fc2426a7abd277313dff326a667b956fdee" +} diff --git a/backend/.sqlx/query-30e9250edd0c4c6025322f51b758174435d6df4f835cb0883ba40897cb99a410.json b/backend/.sqlx/query-30e9250edd0c4c6025322f51b758174435d6df4f835cb0883ba40897cb99a410.json deleted file mode 100644 index 0efe3f75d5..0000000000 --- a/backend/.sqlx/query-30e9250edd0c4c6025322f51b758174435d6df4f835cb0883ba40897cb99a410.json +++ /dev/null @@ -1,15 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "DELETE FROM dependency_map\n WHERE importer_path = $1 AND importer_kind = 'script'\n AND workspace_id = $2", - "describe": { - "columns": [], - "parameters": { - "Left": [ - "Text", - "Text" - ] - }, - "nullable": [] - }, - "hash": "30e9250edd0c4c6025322f51b758174435d6df4f835cb0883ba40897cb99a410" -} diff --git a/backend/.sqlx/query-6c962f9471b0b1fe385a93789ec46bee53a07c8d1264eeb44bc94233bc06bbfd.json b/backend/.sqlx/query-6c962f9471b0b1fe385a93789ec46bee53a07c8d1264eeb44bc94233bc06bbfd.json new file mode 100644 index 0000000000..511fec6586 --- /dev/null +++ b/backend/.sqlx/query-6c962f9471b0b1fe385a93789ec46bee53a07c8d1264eeb44bc94233bc06bbfd.json @@ -0,0 +1,16 @@ +{ + "db_name": "PostgreSQL", + "query": "DELETE FROM dependency_map\n WHERE importer_path = $1 AND importer_kind = $3::text::IMPORTER_KIND\n AND workspace_id = $2", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Text", + "Text", + "Text" + ] + }, + "nullable": [] + }, + "hash": "6c962f9471b0b1fe385a93789ec46bee53a07c8d1264eeb44bc94233bc06bbfd" +} diff --git a/backend/.sqlx/query-aa44311f9bae8253c6c483d122597ce7e470c6ba1905e7e877dca35d9d68968f.json b/backend/.sqlx/query-aa44311f9bae8253c6c483d122597ce7e470c6ba1905e7e877dca35d9d68968f.json deleted file mode 100644 index 808f531923..0000000000 --- a/backend/.sqlx/query-aa44311f9bae8253c6c483d122597ce7e470c6ba1905e7e877dca35d9d68968f.json +++ /dev/null @@ -1,23 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "SELECT importer_path FROM dependency_map\n WHERE imported_path = $1 AND importer_kind = 'script'\n AND workspace_id = $2", - "describe": { - "columns": [ - { - "ordinal": 0, - "name": "importer_path", - "type_info": "Varchar" - } - ], - "parameters": { - "Left": [ - "Text", - "Text" - ] - }, - "nullable": [ - false - ] - }, - "hash": "aa44311f9bae8253c6c483d122597ce7e470c6ba1905e7e877dca35d9d68968f" -} diff --git a/backend/.sqlx/query-5fbd5dfb8df9aecce023c5d482ee58efcbc23c8682a3f0ec94977eb56023ed72.json b/backend/.sqlx/query-e87c317550af671e4b65752a9c0b659cedf430fb7242faee4379b4a13e5fd763.json similarity index 52% rename from backend/.sqlx/query-5fbd5dfb8df9aecce023c5d482ee58efcbc23c8682a3f0ec94977eb56023ed72.json rename to backend/.sqlx/query-e87c317550af671e4b65752a9c0b659cedf430fb7242faee4379b4a13e5fd763.json index c58167c782..83ae3520d7 100644 --- a/backend/.sqlx/query-5fbd5dfb8df9aecce023c5d482ee58efcbc23c8682a3f0ec94977eb56023ed72.json +++ b/backend/.sqlx/query-e87c317550af671e4b65752a9c0b659cedf430fb7242faee4379b4a13e5fd763.json @@ -1,16 +1,18 @@ { "db_name": "PostgreSQL", - "query": "INSERT INTO dependency_map (workspace_id, importer_path, importer_kind, imported_path)\n VALUES ($1, $2, 'script', $3)", + "query": "INSERT INTO dependency_map (workspace_id, importer_path, importer_kind, imported_path, importer_node_id)\n VALUES ($1, $2, $4::text::IMPORTER_KIND, $3, $5)", "describe": { "columns": [], "parameters": { "Left": [ "Varchar", "Varchar", + "Varchar", + "Text", "Varchar" ] }, "nullable": [] }, - "hash": "5fbd5dfb8df9aecce023c5d482ee58efcbc23c8682a3f0ec94977eb56023ed72" + "hash": "e87c317550af671e4b65752a9c0b659cedf430fb7242faee4379b4a13e5fd763" } diff --git a/backend/.sqlx/query-ec3359bbc309c2b893e9f68c09c9f61501068f4b027f58279e0c8508839607f4.json b/backend/.sqlx/query-ec3359bbc309c2b893e9f68c09c9f61501068f4b027f58279e0c8508839607f4.json new file mode 100644 index 0000000000..b2fa44659b --- /dev/null +++ b/backend/.sqlx/query-ec3359bbc309c2b893e9f68c09c9f61501068f4b027f58279e0c8508839607f4.json @@ -0,0 +1,35 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT importer_path, importer_kind::text, array_agg(importer_node_id) as importer_node_ids FROM dependency_map\n WHERE imported_path = $1\n AND workspace_id = $2\n GROUP BY importer_path, importer_kind", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "importer_path", + "type_info": "Varchar" + }, + { + "ordinal": 1, + "name": "importer_kind", + "type_info": "Text" + }, + { + "ordinal": 2, + "name": "importer_node_ids", + "type_info": "VarcharArray" + } + ], + "parameters": { + "Left": [ + "Text", + "Text" + ] + }, + "nullable": [ + false, + null, + null + ] + }, + "hash": "ec3359bbc309c2b893e9f68c09c9f61501068f4b027f58279e0c8508839607f4" +} diff --git a/backend/migrations/20240607113729_add_importer_node.down.sql b/backend/migrations/20240607113729_add_importer_node.down.sql new file mode 100644 index 0000000000..138619577d --- /dev/null +++ b/backend/migrations/20240607113729_add_importer_node.down.sql @@ -0,0 +1,2 @@ +-- Add down migration script here +ALTER TABLE dependency_map DROP COLUMN importer_node_id; \ No newline at end of file diff --git a/backend/migrations/20240607113729_add_importer_node.up.sql b/backend/migrations/20240607113729_add_importer_node.up.sql new file mode 100644 index 0000000000..1b91bbc277 --- /dev/null +++ b/backend/migrations/20240607113729_add_importer_node.up.sql @@ -0,0 +1,4 @@ +-- Add up migration script here +ALTER TABLE dependency_map ADD COLUMN importer_node_id VARCHAR(255) NOT NULL DEFAULT ''; +ALTER TABLE dependency_map DROP constraint dependency_map_pkey; +ALTER TABLE dependency_map ADD PRIMARY KEY (workspace_id, importer_node_id, importer_kind, importer_path, imported_path); \ No newline at end of file diff --git a/backend/windmill-api/src/jobs.rs b/backend/windmill-api/src/jobs.rs index c9cfc11dae..91cc430680 100644 --- a/backend/windmill-api/src/jobs.rs +++ b/backend/windmill-api/src/jobs.rs @@ -1252,7 +1252,7 @@ async fn cancel_all( if let Some(job_running) = job_running { append_logs( - j.id, + &j.id, w_id.clone(), format!("canceled by {username}: cancel_all"), db.clone(), diff --git a/backend/windmill-queue/src/jobs.rs b/backend/windmill-queue/src/jobs.rs index 9bdf599012..db2aedb8e9 100644 --- a/backend/windmill-queue/src/jobs.rs +++ b/backend/windmill-queue/src/jobs.rs @@ -163,7 +163,7 @@ pub async fn cancel_single_job<'c>( .unwrap_or_else(|| "unexplicited reasons".to_string()); let e = serde_json::json!({"message": format!("Job canceled: {reason} by {username}"), "name": "Canceled", "reason": reason, "canceler": username}); append_logs( - job_running.id, + &job_running.id, w_id.to_string(), format!("canceled by {username}: (force cancel: {force_cancel})"), db, @@ -293,7 +293,7 @@ pub async fn cancel_job<'c>( /* TODO retry this? */ #[tracing::instrument(level = "trace", skip_all)] pub async fn append_logs( - job_id: uuid::Uuid, + job_id: &uuid::Uuid, workspace: impl AsRef, logs: impl AsRef, db: impl Borrow>, @@ -1832,7 +1832,7 @@ pub async fn pull( let job_log_event = format!( "\nRe-scheduled job to {estimated_next_schedule_timestamp} due to concurrency limits with key {job_concurrency_key} and limit {job_custom_concurrent_limit} in the last {job_custom_concurrency_time_window_s} seconds", ); - let _ = append_logs(job_uuid, pulled_job.workspace_id, job_log_event, db).await; + let _ = append_logs(&job_uuid, pulled_job.workspace_id, job_log_event, db).await; if rsmq.is_some() { // if let Some(ref mut rsmq) = tx.rsmq { // if using redis, only one message at a time can be poped from the queue. Process only this message and move to the next elligible job diff --git a/backend/windmill-worker/src/bash_executor.rs b/backend/windmill-worker/src/bash_executor.rs index cf0a0297dc..9fc27532bb 100644 --- a/backend/windmill-worker/src/bash_executor.rs +++ b/backend/windmill-worker/src/bash_executor.rs @@ -45,7 +45,7 @@ pub async fn handle_bash_job( envs: HashMap, ) -> Result, Error> { let logs1 = "\n\n--- BASH CODE EXECUTION ---\n".to_string(); - append_logs(job.id, job.workspace_id.clone(), logs1, db).await; + append_logs(&job.id, &job.workspace_id, logs1, db).await; write_file(job_dir, "main.sh", &format!("set -e\n{content}")).await?; write_file( @@ -252,7 +252,7 @@ pub async fn handle_powershell_job( if !install_string.is_empty() { logs1.push_str("\n\nInstalling modules..."); - append_logs(job.id, job.workspace_id.clone(), logs1, db).await; + append_logs(&job.id, &job.workspace_id, logs1, db).await; let child = Command::new("pwsh") .args(&["-Command", &install_string]) .stdout(Stdio::piped()) @@ -277,7 +277,7 @@ pub async fn handle_powershell_job( let mut logs2 = "".to_string(); logs2.push_str("\n\n--- POWERSHELL CODE EXECUTION ---\n"); - append_logs(job.id, job.workspace_id.clone(), logs2, db).await; + append_logs(&job.id, &job.workspace_id, logs2, db).await; // make sure default (only allhostsallusers) modules are loaded, disable autoload (cache can be large to explore especially on cloud) and add /tmp/windmill/cache to PSModulePath let profile = format!( diff --git a/backend/windmill-worker/src/bun_executor.rs b/backend/windmill-worker/src/bun_executor.rs index acab8ac81e..7858210748 100644 --- a/backend/windmill-worker/src/bun_executor.rs +++ b/backend/windmill-worker/src/bun_executor.rs @@ -129,7 +129,7 @@ pub async fn gen_lockfile( "\ndetected trustedDependencies: {}\n", trusted_deps.join(", ") ); - append_logs(job_id.clone(), w_id.to_string(), logs1, db).await; + append_logs(&job_id, w_id, logs1, db).await; let mut content = "".to_string(); { @@ -280,7 +280,7 @@ pub async fn install_lockfile( }; if npm_mode { - append_logs(job_id.clone(), w_id.to_string(), npm_logs, db).await; + append_logs(&job_id.clone(), w_id, npm_logs, db).await; } let child_process = start_child_process(child_cmd, &*BUN_PATH).await?; @@ -575,7 +575,7 @@ pub async fn handle_bun_job( // if !*DISABLE_NSJAIL || !empty_trusted_deps || has_custom_config_registry { let logs1 = "\n\n--- BUN INSTALL ---\n".to_string(); - append_logs(job.id, job.workspace_id.clone(), logs1, db).await; + append_logs(&job.id, &job.workspace_id, logs1, db).await; let _ = gen_lockfile( mem_peak, @@ -607,7 +607,7 @@ pub async fn handle_bun_job( "\n\n--- BUN CODE EXECUTION ---\n".to_string() }; - append_logs(job.id.clone(), job.workspace_id.to_string(), init_logs, db).await; + append_logs(&job.id, &job.workspace_id, init_logs, db).await; let write_wrapper_f = async { // let mut start = Instant::now(); diff --git a/backend/windmill-worker/src/common.rs b/backend/windmill-worker/src/common.rs index 69f6680c06..675aca89be 100644 --- a/backend/windmill-worker/src/common.rs +++ b/backend/windmill-worker/src/common.rs @@ -872,7 +872,7 @@ async fn append_job_logs( .await; } } else { - append_logs(job_id, w_id, logs, db).await; + append_logs(&job_id, w_id, logs, db).await; } } @@ -949,7 +949,7 @@ pub async fn handle_child( let (timeout_duration, timeout_warn_msg) = resolve_job_timeout(&db, w_id, job_id, custom_timeout).await; if let Some(msg) = timeout_warn_msg { - append_logs(job_id, w_id.to_string(), msg.as_str(), db).await; + append_logs(&job_id, w_id, msg.as_str(), db).await; } /* a future that completes when the child process exits */ diff --git a/backend/windmill-worker/src/dedicated_worker.rs b/backend/windmill-worker/src/dedicated_worker.rs index e8a3a4d064..f31102ced8 100644 --- a/backend/windmill-worker/src/dedicated_worker.rs +++ b/backend/windmill-worker/src/dedicated_worker.rs @@ -4,15 +4,28 @@ // let (job_completed_tx, mut new_job) = mpsc::channel::(100); // } +use async_recursion::async_recursion; +use std::{collections::VecDeque, process::Stdio, sync::Arc}; +use tokio::sync::mpsc::Sender; use tokio::{ io::{AsyncBufReadExt, AsyncWriteExt, BufReader}, process::Command, + task::JoinHandle, +}; +use windmill_common::error::Error; +use windmill_common::flows::FlowValue; +use windmill_common::worker::WORKER_CONFIG; +use windmill_common::{ + error, + flows::{FlowModule, FlowModuleValue}, + jobs::QueuedJob, + scripts::{ScriptHash, ScriptLang}, + variables, + worker::to_raw_value, + DB, }; -use windmill_common::{error, jobs::QueuedJob, variables, worker::to_raw_value, DB}; use windmill_queue::append_logs; -use std::{collections::VecDeque, process::Stdio, sync::Arc}; - use anyhow::Context; use crate::{ @@ -45,6 +58,7 @@ async fn write_stdin(stdin: &mut tokio::process::ChildStdin, s: &str) -> error:: Ok(()) } +#[cfg(feature = "enterprise")] pub async fn handle_dedicated_process( command_path: &String, job_dir: &str, @@ -159,7 +173,7 @@ pub async fn handle_dedicated_process( tracing::info!("job completed on dedicated worker {script_path}: {}", job.id); match serde_json::from_str::>(&line.replace("wm_res[success]:", "").replace("wm_res[error]:", "")) { Ok(result) => { - append_logs(job.id, job.workspace_id.clone(), logs.clone(), db).await; + append_logs(&job.id, &job.workspace_id, logs.clone(), db).await; if line.starts_with("wm_res[success]:") { job_completed_tx.send(JobCompleted { job , result, mem_peak: 0, canceled_by: None, success: true, cached_res_path: None, token: token.to_string() }).await.unwrap() } else { @@ -207,3 +221,456 @@ pub async fn handle_dedicated_process( tracing::info!("dedicated worker child process exited successfully"); Ok(()) } + +type DedicatedWorker = (String, Sender>, Option>); + +// spawn one dedicated worker per compatible steps of the flow, associating the node id to the dedicated worker channel send +#[async_recursion] +#[cfg(feature = "enterprise")] +async fn spawn_dedicated_workers_for_flow( + modules: &Vec, + w_id: &str, + path: &str, + killpill_tx: tokio::sync::broadcast::Sender<()>, + killpill_rx: &tokio::sync::broadcast::Receiver<()>, + db: &DB, + worker_dir: &str, + base_internal_url: &str, + worker_name: &str, + job_completed_tx: &JobCompletedSender, +) -> Vec { + let mut workers = vec![]; + let mut script_path_to_worker: HashMap>> = HashMap::new(); + for module in modules.iter() { + let value = module.get_value(); + if let Ok(value) = value { + match &value { + FlowModuleValue::Script { path, hash, .. } => { + let key = format!( + "{}:{}", + path, + hash.clone() + .map(|x| x.to_string()) + .unwrap_or_else(|| "".to_string()) + ); + if let Some(sender) = script_path_to_worker.get(&key) { + workers.push((module.id.clone(), sender.clone(), None)); + } else { + if let Some(dedi_w) = spawn_dedicated_worker( + SpawnWorker::Script { path: path.to_string(), hash: hash.clone() }, + w_id, + killpill_tx.clone(), + killpill_rx, + db, + worker_dir, + base_internal_url, + worker_name, + job_completed_tx, + Some(module.id.clone()), + ) + .await + { + script_path_to_worker.insert(key, dedi_w.1.clone()); + workers.push(dedi_w); + } + } + } + FlowModuleValue::ForloopFlow { modules, .. } => { + let w = spawn_dedicated_workers_for_flow( + &modules, + path, + w_id, + killpill_tx.clone(), + killpill_rx, + db, + worker_dir, + base_internal_url, + worker_name, + job_completed_tx, + ) + .await; + workers.extend(w); + } + FlowModuleValue::WhileloopFlow { modules, .. } => { + let w = spawn_dedicated_workers_for_flow( + &modules, + path, + w_id, + killpill_tx.clone(), + killpill_rx, + db, + worker_dir, + base_internal_url, + worker_name, + job_completed_tx, + ) + .await; + workers.extend(w); + } + FlowModuleValue::BranchOne { branches, default } => { + for modules in branches + .iter() + .map(|x| &x.modules) + .chain(std::iter::once(default)) + { + let w = spawn_dedicated_workers_for_flow( + &modules, + path, + w_id, + killpill_tx.clone(), + killpill_rx, + db, + worker_dir, + base_internal_url, + worker_name, + job_completed_tx, + ) + .await; + workers.extend(w); + } + } + FlowModuleValue::BranchAll { branches, .. } => { + for branch in branches { + let w = spawn_dedicated_workers_for_flow( + &branch.modules, + path, + w_id, + killpill_tx.clone(), + killpill_rx, + db, + worker_dir, + base_internal_url, + worker_name, + job_completed_tx, + ) + .await; + workers.extend(w); + } + } + FlowModuleValue::RawScript { content, lock, path: spath, language, .. } => { + if let Some(dedi_w) = spawn_dedicated_worker( + SpawnWorker::RawScript { + path: spath.clone().unwrap_or(path.to_string()), + content: content.to_string(), + lock: lock.clone(), + lang: language.clone(), + }, + w_id, + killpill_tx.clone(), + killpill_rx, + db, + worker_dir, + base_internal_url, + worker_name, + job_completed_tx, + Some(module.id.clone()), + ) + .await + { + workers.push(dedi_w); + } + } + FlowModuleValue::Flow { .. } => (), + FlowModuleValue::Identity => (), + } + } else { + tracing::error!("failed to get value for module: {:?}", module); + } + } + workers +} + +pub async fn create_dedicated_worker_map( + killpill_tx: &tokio::sync::broadcast::Sender<()>, + killpill_rx: &tokio::sync::broadcast::Receiver<()>, + db: &DB, + worker_dir: &str, + base_internal_url: &str, + worker_name: &str, + job_completed_tx: &JobCompletedSender, +) -> ( + HashMap>>, + bool, + Vec>, +) { + let mut dedicated_handles = vec![]; + if let Some(_wp) = WORKER_CONFIG.read().await.dedicated_worker.clone() { + let mut hm = HashMap::new(); + let is_flow_worker; + if let Some(flow_path) = _wp.path.strip_prefix("flow/") { + is_flow_worker = true; + let value = sqlx::query_scalar!( + "SELECT value FROM flow WHERE path = $1 AND workspace_id = $2", + flow_path, + _wp.workspace_id + ) + .fetch_optional(db) + .await; + if let Ok(v) = value { + if let Some(v) = v { + let value = serde_json::from_value::(v).map_err(|err| { + Error::InternalErr(format!( + "could not convert json to flow for {flow_path}: {err:?}" + )) + }); + if let Ok(flow) = value { + let workers = spawn_dedicated_workers_for_flow( + &flow.modules, + &_wp.workspace_id, + &_wp.path, + killpill_tx.clone(), + &killpill_rx, + db, + &worker_dir, + base_internal_url, + &worker_name, + &job_completed_tx, + ) + .await; + workers.into_iter().for_each(|(path, sender, handle)| { + tracing::info!("spawned dedicated worker for flow: {}", path.as_str()); + if let Some(h) = handle { + dedicated_handles.push(h); + } + hm.insert(path, sender); + }); + } + } else { + tracing::error!( + "flow present but value not found for dedicated worker. {}", + flow_path + ); + } + } else { + tracing::error!("flow not found for dedicated worker: {}. Waiting for dependency job and expected to restart.", flow_path); + } + } else { + is_flow_worker = false; + if let Some((path, sender, handle)) = spawn_dedicated_worker( + SpawnWorker::Script { path: _wp.path.clone(), hash: None }, + &_wp.workspace_id, + killpill_tx.clone(), + &killpill_rx, + db, + &worker_dir, + base_internal_url, + &worker_name, + &job_completed_tx, + None, + ) + .await + { + if let Some(h) = handle { + dedicated_handles.push(h); + } + hm.insert(path, sender); + } else { + tracing::error!( + "failed to spawn dedicated worker for {}, script not found", + _wp.path + ); + } + } + (hm, is_flow_worker, dedicated_handles) + } else { + (HashMap::new(), false, dedicated_handles) + } +} +pub enum SpawnWorker { + Script { path: String, hash: Option }, + RawScript { path: String, content: String, lock: Option, lang: ScriptLang }, +} + +// spawn one dedicated worker and return the key, the channel sender and the join handle +// note that for it will return none for language that do not support dedicated workers +// note that go using cache binary does not need dedicated workers so all languages are supported +#[cfg(feature = "enterprise")] +async fn spawn_dedicated_worker( + sw: SpawnWorker, + w_id: &str, + killpill_tx: tokio::sync::broadcast::Sender<()>, + killpill_rx: &tokio::sync::broadcast::Receiver<()>, + db: &DB, + worker_dir: &str, + base_internal_url: &str, + worker_name: &str, + job_completed_tx: &JobCompletedSender, + node_id: Option, +) -> Option { + use windmill_common::{ + error::Error, + scripts::{ScriptHash, ScriptLang}, + utils::rd_string, + }; + + use crate::{build_envs, get_script_content_by_hash, ContentReqLangEnvs, JOB_TOKEN}; + + #[cfg(not(feature = "enterprise"))] + { + tracing::error!("Dedicated worker is an enterprise feature"); + killpill_tx.send(()).expect("send"); + return None; + } + + #[cfg(feature = "enterprise")] + { + let (dedicated_worker_tx, dedicated_worker_rx) = + tokio::sync::mpsc::channel::>(MAX_BUFFERED_DEDICATED_JOBS); + let killpill_rx = killpill_rx.resubscribe(); + let db = db.clone(); + let base_internal_url = base_internal_url.to_string(); + let worker_name = worker_name.to_string(); + let job_completed_tx = job_completed_tx.clone(); + let job_dir = format!("{}/dedicated", worker_dir); + tokio::fs::create_dir_all(&job_dir) + .await + .expect("create dir"); + + let path = match &sw { + SpawnWorker::RawScript { path, .. } => path.to_string(), + SpawnWorker::Script { path, .. } => path.to_string(), + }; + + let path2 = path.clone(); + let w_id = w_id.to_string(); + + let (content, lock, language, envs, codebase) = match sw { + SpawnWorker::Script { path, hash } => { + let q = if let Some(hash) = hash { + get_script_content_by_hash(&hash, &w_id, &db).await.map( + |r: ContentReqLangEnvs| { + Some((r.content, r.lockfile, r.language, r.envs, r.codebase)) + }, + ) + } else { + sqlx::query_as::<_, (String, Option, Option, Option>, bool, Option)>( + "SELECT content, lock, language, envs, codebase IS NOT NULL, hash FROM script WHERE path = $1 AND workspace_id = $2 AND + created_at = (SELECT max(created_at) FROM script WHERE path = $1 AND workspace_id = $2 AND + deleted = false AND lock IS not NULL AND lock_error_logs IS NULL)", + ) + .bind(&path) + .bind(&w_id) + .fetch_optional(&db) + .await + .map_err(|e| Error::InternalErr(format!("expected content and lock: {e:#}"))) + .map(|x| x.map(|y| (y.0, y.1, y.2, y.3, if y.4 { y.5.map(|z| z.to_string()) } else { None }))) + }; + if let Ok(q) = q { + if let Some(wp) = q { + wp + } else { + tracing::error!( + "Failed to fetch script `{}` in workspace {} for dedicated worker.", + path, + w_id + ); + return None; + } + } else { + tracing::error!("Failed to fetch script for dedicated worker"); + killpill_tx.send(()).expect("send"); + return None; + } + } + SpawnWorker::RawScript { content, lock, lang, .. } => { + (content, lock, Some(lang), None, None) + } + }; + + match language { + Some(ScriptLang::Python3) | Some(ScriptLang::Bun) | Some(ScriptLang::Deno) => {} + _ => return None, + } + + let handle = tokio::spawn(async move { + let token = if let Some(token) = JOB_TOKEN.as_ref() { + token.clone() + } else { + let token = rd_string(32); + if let Err(e) = sqlx::query_scalar!( + "INSERT INTO token + (token, label, super_admin, email) + VALUES ($1, $2, $3, $4)", + token, + "dedicated_worker", + true, + "dedicated_worker@windmill.dev" + ) + .execute(&db) + .await + { + tracing::error!("failed to create token for dedicated worker: {:?}", e); + killpill_tx.clone().send(()).expect("send"); + }; + token + }; + + let worker_envs = build_envs(envs).expect("failed to build envs"); + + if let Err(e) = match language { + Some(ScriptLang::Python3) => { + crate::python_executor::start_worker( + lock, + &db, + &content, + &base_internal_url, + &job_dir, + &worker_name, + worker_envs, + &w_id, + &path, + &token, + job_completed_tx, + dedicated_worker_rx, + killpill_rx, + ) + .await + } + Some(ScriptLang::Bun) => { + crate::bun_executor::start_worker( + lock, + codebase, + &db, + &content, + &base_internal_url, + &job_dir, + &worker_name, + worker_envs, + &w_id, + &path, + &token, + job_completed_tx, + dedicated_worker_rx, + killpill_rx, + ) + .await + } + Some(ScriptLang::Deno) => { + crate::deno_executor::start_worker( + &content, + &base_internal_url, + &job_dir, + &worker_name, + worker_envs, + &w_id, + &path, + &token, + job_completed_tx, + dedicated_worker_rx, + killpill_rx, + &db, + ) + .await + } + _ => unreachable!("Non supported language for dedicated worker"), + } { + tracing::error!("error in dedicated worker: {:?}", e); + }; + if let Err(e) = killpill_tx.clone().send(()) { + tracing::error!("failed to send final killpill to dedicated worker: {:?}", e); + } + }); + return Some((node_id.unwrap_or(path2), dedicated_worker_tx, Some(handle))); + // (Some(dedi_path), Some(dedicated_worker_tx), Some(handle)) + } +} diff --git a/backend/windmill-worker/src/deno_executor.rs b/backend/windmill-worker/src/deno_executor.rs index 5888313b92..f082c113ea 100644 --- a/backend/windmill-worker/src/deno_executor.rs +++ b/backend/windmill-worker/src/deno_executor.rs @@ -172,7 +172,7 @@ pub async fn handle_deno_job( ) -> error::Result> { // let mut start = Instant::now(); let logs1 = "\n\n--- DENO CODE EXECUTION ---\n".to_string(); - append_logs(job.id.clone(), job.workspace_id.to_string(), logs1, db).await; + append_logs(&job.id, &job.workspace_id, logs1, db).await; let main_override = get_main_override(job.args.as_ref()); diff --git a/backend/windmill-worker/src/go_executor.rs b/backend/windmill-worker/src/go_executor.rs index 875874d800..a8e7919a40 100644 --- a/backend/windmill-worker/src/go_executor.rs +++ b/backend/windmill-worker/src/go_executor.rs @@ -70,8 +70,8 @@ pub async fn save_cache( if !*CLOUD_HOSTED { tokio::fs::copy(&job_main_path, bin_path).await?; append_logs( - job.id.clone(), - job.workspace_id.to_string(), + &job.id, + &job.workspace_id, format!( "\nwrite cached binary: {} (backed by object store: {_cached_to_s3})\n", bin_path @@ -81,8 +81,8 @@ pub async fn save_cache( .await; } else if _cached_to_s3 { append_logs( - job.id.clone(), - job.workspace_id.to_string(), + &job.id, + &job.workspace_id, format!("write cached binary to object store {}\n", bin_path), db, ) @@ -178,7 +178,7 @@ pub async fn handle_go_job( let cache_logs = if !cache { let logs1 = format!("{cache_logs}\n\n--- GO DEPENDENCIES SETUP ---\n"); - append_logs(job.id.clone(), job.workspace_id.to_string(), logs1, db).await; + append_logs(&job.id, &job.workspace_id, logs1, db).await; install_go_dependencies( &job.id, @@ -326,7 +326,7 @@ func Run(req Req) (interface{{}}, error){{ }; let logs2 = format!("{cache_logs}\n\n--- GO CODE EXECUTION ---\n"); - append_logs(job.id.clone(), job.workspace_id.to_string(), logs2, db).await; + append_logs(&job.id, &job.workspace_id, logs2, db).await; let client = &client.get_authed().await; @@ -494,7 +494,7 @@ pub async fn install_go_dependencies( .await? { let logs1 = format!("\nfound cached resolution: {}", hash); - append_logs(job_id.clone(), w_id.to_string(), logs1, db).await; + append_logs(&job_id, w_id, logs1, db).await; gen_go_mod(code, job_dir, &cached).await?; skip_tidy = true; new_lockfile = false; diff --git a/backend/windmill-worker/src/lib.rs b/backend/windmill-worker/src/lib.rs index 7ee01679dc..ede73f08ec 100644 --- a/backend/windmill-worker/src/lib.rs +++ b/backend/windmill-worker/src/lib.rs @@ -22,4 +22,5 @@ mod php_executor; mod python_executor; mod worker; mod worker_flow; +mod worker_lockfiles; pub use worker::*; diff --git a/backend/windmill-worker/src/mssql_executor.rs b/backend/windmill-worker/src/mssql_executor.rs index f788b86310..beeb62c2b0 100644 --- a/backend/windmill-worker/src/mssql_executor.rs +++ b/backend/windmill-worker/src/mssql_executor.rs @@ -84,7 +84,7 @@ pub async fn do_mssql( if readonly_intent { let logs = format!("\nSetting ApplicationIntent to ReadOnly"); - append_logs(job.id, job.workspace_id.clone(), logs, db).await; + append_logs(&job.id, &job.workspace_id, logs, db).await; } // Using SQL Server authentication. diff --git a/backend/windmill-worker/src/php_executor.rs b/backend/windmill-worker/src/php_executor.rs index 730ebacf29..9c4f8fc40d 100644 --- a/backend/windmill-worker/src/php_executor.rs +++ b/backend/windmill-worker/src/php_executor.rs @@ -155,7 +155,7 @@ pub async fn handle_php_job( let autoload_line = if let Some(composer_json) = composer_json { let logs1 = "\n\n--- COMPOSER INSTALL ---\n".to_string(); - append_logs(job.id, job.workspace_id.clone(), logs1, db).await; + append_logs(&job.id, &job.workspace_id, logs1, db).await; composer_install( mem_peak, @@ -176,7 +176,7 @@ pub async fn handle_php_job( let init_logs = "\n\n--- PHP CODE EXECUTION ---\n".to_string(); - append_logs(job.id.clone(), job.workspace_id.to_string(), init_logs, db).await; + append_logs(&job.id, job.workspace_id.to_string(), init_logs, db).await; let _ = write_file(job_dir, "main.php", inner_content).await?; diff --git a/backend/windmill-worker/src/python_executor.rs b/backend/windmill-worker/src/python_executor.rs index 4f33688491..5efa15d780 100644 --- a/backend/windmill-worker/src/python_executor.rs +++ b/backend/windmill-worker/src/python_executor.rs @@ -197,7 +197,7 @@ pub async fn pip_compile( .stdout(Stdio::piped()) .stderr(Stdio::piped()); let child_process = start_child_process(child_cmd, "pip-compile").await?; - append_logs(job_id.clone(), w_id.to_string(), logs, db).await; + append_logs(&job_id, &w_id, logs, db).await; handle_child( job_id, db, @@ -264,8 +264,8 @@ pub async fn handle_python_job( .await?; append_logs( - job.id.clone(), - job.workspace_id.to_string(), + &job.id, + &job.workspace_id, "\n\n--- PYTHON CODE EXECUTION ---\n".to_string(), db, ) @@ -855,8 +855,8 @@ pub async fn handle_python_reqs( } if pulled.len() > 0 { append_logs( - job_id.clone(), - w_id.to_string(), + &job_id, + &w_id, format!( "pulled {} from distributed cache in {}ms", pulled.join(", "), @@ -873,7 +873,7 @@ pub async fn handle_python_reqs( let mut logs1 = String::new(); logs1.push_str("\n\n--- 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.clone(), w_id.to_string(), logs1, db).await; + append_logs(&job_id, w_id, logs1, db).await; tracing::info!( workspace_id = %w_id, diff --git a/backend/windmill-worker/src/worker.rs b/backend/windmill-worker/src/worker.rs index 0133db3dee..17682b5a42 100644 --- a/backend/windmill-worker/src/worker.rs +++ b/backend/windmill-worker/src/worker.rs @@ -27,31 +27,23 @@ use sqlx::{types::Json, Pool, Postgres}; use std::{ collections::{hash_map::DefaultHasher, HashMap}, hash::Hash, - path::{Component, Path, PathBuf}, sync::{ atomic::{AtomicBool, AtomicUsize, Ordering}, Arc, }, time::Duration, }; -use windmill_git_sync::{handle_deployment_metadata, DeployedObject}; -use windmill_parser_py_imports::parse_relative_imports; -use windmill_parser_ts::parse_expr_for_imports; use uuid::Uuid; use windmill_common::{ error::{self, to_anyhow, Error}, - flows::{FlowModule, FlowModuleValue, FlowValue}, get_latest_deployed_hash_for_path, - jobs::{JobKind, JobPayload, QueuedJob}, + jobs::{JobKind, QueuedJob}, scripts::{get_full_hub_script_by_path, ScriptHash, ScriptLang, PREVIEW_IS_CODEBASE_HASH}, users::{SUPERADMIN_NOTIFICATION_EMAIL, SUPERADMIN_SECRET_EMAIL, SUPERADMIN_SYNC_EMAIL}, utils::{rd_string, StripPath}, - worker::{ - to_raw_value, to_raw_value_owned, update_ping, CLOUD_HOSTED, NO_LOGS, WORKER_CONFIG, - WORKER_GROUP, - }, + worker::{to_raw_value, update_ping, CLOUD_HOSTED, NO_LOGS, WORKER_CONFIG, WORKER_GROUP}, DB, IS_READY, }; @@ -63,7 +55,7 @@ use windmill_queue::{ #[cfg(feature = "prometheus")] use windmill_queue::register_metric; -use serde_json::{json, value::RawValue, Value}; +use serde_json::{json, value::RawValue}; #[cfg(any(target_os = "linux", target_os = "macos"))] use tokio::fs::symlink; @@ -83,34 +75,36 @@ use tokio::{ use futures::future::FutureExt; -use async_recursion::async_recursion; - use rand::Rng; use windmill_queue::{add_completed_job, add_completed_job_error}; use crate::{ bash_executor::{handle_bash_job, handle_powershell_job, ANSI_ESCAPE_RE}, - bun_executor::{gen_lockfile, get_trusted_deps, handle_bun_job}, + bun_executor::handle_bun_job, common::{ build_args_map, get_cached_resource_value_if_valid, get_reserved_variables, hash_args, read_result, save_in_cache, write_file, NO_LOGS_AT_ALL, SLOW_LOGS, }, - deno_executor::{generate_deno_lock, handle_deno_job}, - go_executor::{handle_go_job, install_go_dependencies}, + deno_executor::handle_deno_job, + go_executor::handle_go_job, graphql_executor::do_graphql, js_eval::{eval_fetch_timeout, transpile_ts}, mysql_executor::do_mysql, pg_executor::do_postgresql, - php_executor::{composer_install, handle_php_job, parse_php_imports}, - python_executor::{ - create_dependencies_dir, handle_python_job, handle_python_reqs, pip_compile, - }, + php_executor::handle_php_job, + python_executor::handle_python_job, worker_flow::{ handle_flow, update_flow_status_after_job_completion, update_flow_status_in_progress, }, + worker_lockfiles::{ + handle_app_dependency_job, handle_dependency_job, handle_flow_dependency_job, + }, }; +#[cfg(feature = "enterprise")] +use crate::dedicated_worker::create_dedicated_worker_map; + #[cfg(feature = "enterprise")] use crate::{ bigquery_executor::do_bigquery, mssql_executor::do_mssql, snowflake_executor::do_snowflake, @@ -256,7 +250,7 @@ lazy_static::lazy_static! { lazy_static::lazy_static! { - static ref JOB_TOKEN: Option = std::env::var("JOB_TOKEN").ok(); + pub static ref JOB_TOKEN: Option = std::env::var("JOB_TOKEN").ok(); static ref SLEEP_QUEUE: u64 = std::env::var("SLEEP_QUEUE") .ok() @@ -1344,92 +1338,28 @@ pub async fn run_worker>>, // Option>, - let mut dedicated_handles: Vec> = vec![]; - let (dedicated_workers, is_flow_worker): (HashMap>>, bool) = - if let Some(_wp) = WORKER_CONFIG.read().await.dedicated_worker.clone() { - let mut hm = HashMap::new(); - let is_flow_worker; - if let Some(flow_path) = _wp.path.strip_prefix("flow/") { - is_flow_worker = true; - let value = sqlx::query_scalar!( - "SELECT value FROM flow WHERE path = $1 AND workspace_id = $2", - flow_path, - _wp.workspace_id - ) - .fetch_optional(db) - .await; - if let Ok(v) = value { - if let Some(v) = v { - let value = serde_json::from_value::(v).map_err(|err| { - Error::InternalErr(format!( - "could not convert json to flow for {flow_path}: {err:?}" - )) - }); - if let Ok(flow) = value { - let workers = spawn_dedicated_workers_for_flow( - &flow.modules, - &_wp.workspace_id, - &_wp.path, - killpill_tx.clone(), - &killpill_rx, - db, - &worker_dir, - base_internal_url, - &worker_name, - &job_completed_tx, - ) - .await; - workers.into_iter().for_each(|(path, sender, handle)| { - tracing::info!( - "spawned dedicated worker for flow: {}", - path.as_str() - ); - if let Some(h) = handle { - dedicated_handles.push(h); - } - hm.insert(path, sender); - }); - } - } else { - tracing::error!( - "flow present but value not found for dedicated worker. {}", - flow_path - ); - } - } else { - tracing::error!("flow not found for dedicated worker: {}. Waiting for dependency job and expected to restart.", flow_path); - } - } else { - is_flow_worker = false; - if let Some((path, sender, handle)) = spawn_dedicated_worker( - SpawnWorker::Script { path: _wp.path.clone(), hash: None }, - &_wp.workspace_id, - killpill_tx.clone(), - &killpill_rx, - db, - &worker_dir, - base_internal_url, - &worker_name, - &job_completed_tx, - None, - ) - .await - { - if let Some(h) = handle { - dedicated_handles.push(h); - } - hm.insert(path, sender); - } else { - tracing::error!( - "failed to spawn dedicated worker for {}, script not found", - _wp.path - ); - } - } - (hm, is_flow_worker) - } else { - (HashMap::new(), false) - }; + #[cfg(feature = "enterprise")] + let (dedicated_workers, is_flow_worker, dedicated_handles): ( + HashMap>>, + bool, + Vec>, + ) = create_dedicated_worker_map( + &killpill_tx, + &killpill_rx, + db, + &worker_dir, + base_internal_url, + &worker_name, + &job_completed_tx, + ) + .await; + + #[cfg(not(feature = "enterprise"))] + let (dedicated_workers, is_flow_worker, dedicated_handles): ( + HashMap>>, + bool, + Vec>, + ) = (HashMap::new(), false, vec![]); #[cfg(feature = "benchmark")] tracing::info!("pre loop time {}s", start.elapsed().as_secs_f64()); @@ -1943,372 +1873,6 @@ pub async fn run_worker>, Option>); - -// spawn one dedicated worker per compatible steps of the flow, associating the node id to the dedicated worker channel send -#[async_recursion] -async fn spawn_dedicated_workers_for_flow( - modules: &Vec, - w_id: &str, - path: &str, - killpill_tx: tokio::sync::broadcast::Sender<()>, - killpill_rx: &tokio::sync::broadcast::Receiver<()>, - db: &Pool, - worker_dir: &str, - base_internal_url: &str, - worker_name: &str, - job_completed_tx: &JobCompletedSender, -) -> Vec { - let mut workers = vec![]; - let mut script_path_to_worker: HashMap>> = HashMap::new(); - for module in modules.iter() { - let value = module.get_value(); - if let Ok(value) = value { - match &value { - FlowModuleValue::Script { path, hash, .. } => { - let key = format!( - "{}:{}", - path, - hash.clone() - .map(|x| x.to_string()) - .unwrap_or_else(|| "".to_string()) - ); - if let Some(sender) = script_path_to_worker.get(&key) { - workers.push((module.id.clone(), sender.clone(), None)); - } else { - if let Some(dedi_w) = spawn_dedicated_worker( - SpawnWorker::Script { path: path.to_string(), hash: hash.clone() }, - w_id, - killpill_tx.clone(), - killpill_rx, - db, - worker_dir, - base_internal_url, - worker_name, - job_completed_tx, - Some(module.id.clone()), - ) - .await - { - script_path_to_worker.insert(key, dedi_w.1.clone()); - workers.push(dedi_w); - } - } - } - FlowModuleValue::ForloopFlow { modules, .. } => { - let w = spawn_dedicated_workers_for_flow( - &modules, - path, - w_id, - killpill_tx.clone(), - killpill_rx, - db, - worker_dir, - base_internal_url, - worker_name, - job_completed_tx, - ) - .await; - workers.extend(w); - } - FlowModuleValue::WhileloopFlow { modules, .. } => { - let w = spawn_dedicated_workers_for_flow( - &modules, - path, - w_id, - killpill_tx.clone(), - killpill_rx, - db, - worker_dir, - base_internal_url, - worker_name, - job_completed_tx, - ) - .await; - workers.extend(w); - } - FlowModuleValue::BranchOne { branches, default } => { - for modules in branches - .iter() - .map(|x| &x.modules) - .chain(std::iter::once(default)) - { - let w = spawn_dedicated_workers_for_flow( - &modules, - path, - w_id, - killpill_tx.clone(), - killpill_rx, - db, - worker_dir, - base_internal_url, - worker_name, - job_completed_tx, - ) - .await; - workers.extend(w); - } - } - FlowModuleValue::BranchAll { branches, .. } => { - for branch in branches { - let w = spawn_dedicated_workers_for_flow( - &branch.modules, - path, - w_id, - killpill_tx.clone(), - killpill_rx, - db, - worker_dir, - base_internal_url, - worker_name, - job_completed_tx, - ) - .await; - workers.extend(w); - } - } - FlowModuleValue::RawScript { content, lock, path: spath, language, .. } => { - if let Some(dedi_w) = spawn_dedicated_worker( - SpawnWorker::RawScript { - path: spath.clone().unwrap_or(path.to_string()), - content: content.to_string(), - lock: lock.clone(), - lang: language.clone(), - }, - w_id, - killpill_tx.clone(), - killpill_rx, - db, - worker_dir, - base_internal_url, - worker_name, - job_completed_tx, - Some(module.id.clone()), - ) - .await - { - workers.push(dedi_w); - } - } - FlowModuleValue::Flow { .. } => (), - FlowModuleValue::Identity => (), - } - } else { - tracing::error!("failed to get value for module: {:?}", module); - } - } - workers -} - -pub enum SpawnWorker { - Script { path: String, hash: Option }, - RawScript { path: String, content: String, lock: Option, lang: ScriptLang }, -} - -#[cfg(not(feature = "enterprise"))] -async fn spawn_dedicated_worker( - _sw: SpawnWorker, - _w_id: &str, - killpill_tx: tokio::sync::broadcast::Sender<()>, - _killpill_rx: &tokio::sync::broadcast::Receiver<()>, - _db: &Pool, - _worker_dir: &str, - _base_internal_url: &str, - _worker_name: &str, - _job_completed_tx: &JobCompletedSender, - _node_id: Option, -) -> Option { - tracing::error!("Dedicated worker is an enterprise feature"); - killpill_tx.send(()).expect("send"); - return None; -} - -// spawn one dedicated worker and return the key, the channel sender and the join handle -// note that for it will return none for language that do not support dedicated workers -// note that go using cache binary does not need dedicated workers so all languages are supported -#[cfg(feature = "enterprise")] -async fn spawn_dedicated_worker( - sw: SpawnWorker, - w_id: &str, - killpill_tx: tokio::sync::broadcast::Sender<()>, - killpill_rx: &tokio::sync::broadcast::Receiver<()>, - db: &Pool, - worker_dir: &str, - base_internal_url: &str, - worker_name: &str, - job_completed_tx: &JobCompletedSender, - node_id: Option, -) -> Option { - #[cfg(not(feature = "enterprise"))] - { - tracing::error!("Dedicated worker is an enterprise feature"); - killpill_tx.send(()).expect("send"); - return None; - } - - #[cfg(feature = "enterprise")] - { - let (dedicated_worker_tx, dedicated_worker_rx) = - mpsc::channel::>(MAX_BUFFERED_DEDICATED_JOBS); - let killpill_rx = killpill_rx.resubscribe(); - let db = db.clone(); - let base_internal_url = base_internal_url.to_string(); - let worker_name = worker_name.to_string(); - let job_completed_tx = job_completed_tx.clone(); - let job_dir = format!("{}/dedicated", worker_dir); - tokio::fs::create_dir_all(&job_dir) - .await - .expect("create dir"); - - let path = match &sw { - SpawnWorker::RawScript { path, .. } => path.to_string(), - SpawnWorker::Script { path, .. } => path.to_string(), - }; - - let path2 = path.clone(); - let w_id = w_id.to_string(); - - let (content, lock, language, envs, codebase) = match sw { - SpawnWorker::Script { path, hash } => { - let q = if let Some(hash) = hash { - get_script_content_by_hash(&hash, &w_id, &db).await.map( - |r: ContentReqLangEnvs| { - Some((r.content, r.lockfile, r.language, r.envs, r.codebase)) - }, - ) - } else { - sqlx::query_as::<_, (String, Option, Option, Option>, bool, Option)>( - "SELECT content, lock, language, envs, codebase IS NOT NULL, hash FROM script WHERE path = $1 AND workspace_id = $2 AND - created_at = (SELECT max(created_at) FROM script WHERE path = $1 AND workspace_id = $2 AND - deleted = false AND lock IS not NULL AND lock_error_logs IS NULL)", - ) - .bind(&path) - .bind(&w_id) - .fetch_optional(&db) - .await - .map_err(|e| Error::InternalErr(format!("expected content and lock: {e:#}"))) - .map(|x| x.map(|y| (y.0, y.1, y.2, y.3, if y.4 { y.5.map(|z| z.to_string()) } else { None }))) - }; - if let Ok(q) = q { - if let Some(wp) = q { - wp - } else { - tracing::error!( - "Failed to fetch script `{}` in workspace {} for dedicated worker.", - path, - w_id - ); - return None; - } - } else { - tracing::error!("Failed to fetch script for dedicated worker"); - killpill_tx.send(()).expect("send"); - return None; - } - } - SpawnWorker::RawScript { content, lock, lang, .. } => { - (content, lock, Some(lang), None, None) - } - }; - - match language { - Some(ScriptLang::Python3) | Some(ScriptLang::Bun) | Some(ScriptLang::Deno) => {} - _ => return None, - } - - let handle = tokio::spawn(async move { - let token = if let Some(token) = JOB_TOKEN.as_ref() { - token.clone() - } else { - let token = rd_string(32); - if let Err(e) = sqlx::query_scalar!( - "INSERT INTO token - (token, label, super_admin, email) - VALUES ($1, $2, $3, $4)", - token, - "dedicated_worker", - true, - "dedicated_worker@windmill.dev" - ) - .execute(&db) - .await - { - tracing::error!("failed to create token for dedicated worker: {:?}", e); - killpill_tx.clone().send(()).expect("send"); - }; - token - }; - - let worker_envs = build_envs(envs).expect("failed to build envs"); - - if let Err(e) = match language { - Some(ScriptLang::Python3) => { - crate::python_executor::start_worker( - lock, - &db, - &content, - &base_internal_url, - &job_dir, - &worker_name, - worker_envs, - &w_id, - &path, - &token, - job_completed_tx, - dedicated_worker_rx, - killpill_rx, - ) - .await - } - Some(ScriptLang::Bun) => { - crate::bun_executor::start_worker( - lock, - codebase, - &db, - &content, - &base_internal_url, - &job_dir, - &worker_name, - worker_envs, - &w_id, - &path, - &token, - job_completed_tx, - dedicated_worker_rx, - killpill_rx, - ) - .await - } - Some(ScriptLang::Deno) => { - crate::deno_executor::start_worker( - &content, - &base_internal_url, - &job_dir, - &worker_name, - worker_envs, - &w_id, - &path, - &token, - job_completed_tx, - dedicated_worker_rx, - killpill_rx, - &db, - ) - .await - } - _ => unreachable!("Non supported language for dedicated worker"), - } { - tracing::error!("error in dedicated worker: {:?}", e); - }; - if let Err(e) = killpill_tx.clone().send(()) { - tracing::error!("failed to send final killpill to dedicated worker: {:?}", e); - } - }); - return Some((node_id.unwrap_or(path2), dedicated_worker_tx, Some(handle))); - // (Some(dedi_path), Some(dedicated_worker_tx), Some(handle)) - } -} - async fn queue_init_bash_maybe<'c, R: rsmq_async::RsmqConnection + Send + 'c>( db: &Pool, same_worker_tx: Sender, @@ -2544,8 +2108,8 @@ pub async fn handle_job_error( { let logs = "Job skipped because args & path found in cache and not expired".to_string(); - append_logs(job.id, job.workspace_id.clone(), logs, db).await; + append_logs(&job.id, &job.workspace_id, logs, db).await; } job_completed_tx .send(JobCompleted { @@ -2902,7 +2466,7 @@ async fn handle_queued_job( "handling job {}", job.id ); - append_logs(job.id, job.workspace_id.clone(), logs, db).await; + append_logs(&job.id, &job.workspace_id, logs, db).await; let mut column_order: Option> = None; let result = match job.job_kind { @@ -2921,20 +2485,21 @@ async fn handle_queued_job( ) .await } - JobKind::FlowDependencies => handle_flow_dependency_job( - &job, - &mut mem_peak, - &mut canceled_by, - job_dir, - db, - worker_name, - worker_dir, - base_internal_url, - &client.get_token().await, - rsmq.clone(), - ) - .await - .map(|()| serde_json::from_str("{}").unwrap()), + JobKind::FlowDependencies => { + handle_flow_dependency_job( + &job, + &mut mem_peak, + &mut canceled_by, + job_dir, + db, + worker_name, + worker_dir, + base_internal_url, + &client.get_token().await, + rsmq.clone(), + ) + .await + } JobKind::AppDependencies => handle_app_dependency_job( &job, &mut mem_peak, @@ -3100,7 +2665,7 @@ async fn process_result( Ok(()) } -fn build_envs( +pub fn build_envs( envs: Option>, ) -> windmill_common::error::Result> { let mut envs = if *CLOUD_HOSTED || envs.is_none() { @@ -3131,12 +2696,12 @@ fn build_envs( Ok(envs) } -struct ContentReqLangEnvs { - content: String, - lockfile: Option, - language: Option, - envs: Option>, - codebase: Option, +pub struct ContentReqLangEnvs { + pub content: String, + pub lockfile: Option, + pub language: Option, + pub envs: Option>, + pub codebase: Option, } async fn get_hub_script_content_and_requirements( @@ -3194,7 +2759,7 @@ async fn get_script_content_by_path( }; } -async fn get_script_content_by_hash( +pub async fn get_script_content_by_hash( script_hash: &ScriptHash, w_id: &str, db: &DB, @@ -3387,8 +2952,8 @@ async fn handle_code_execution_job( .await; } else if language == Some(ScriptLang::Nativets) { append_logs( - job.id, - job.workspace_id.clone(), + &job.id, + &job.workspace_id, "\n--- FETCH TS EXECUTION ---\n", db, ) @@ -3414,7 +2979,7 @@ async fn handle_code_execution_job( worker_name, ) .await?; - append_logs(job.id, job.workspace_id.clone(), ts_logs, db).await; + append_logs(&job.id, &job.workspace_id, ts_logs, db).await; return Ok(result); } @@ -3587,1045 +3152,3 @@ mount {{ result } - -fn try_normalize(path: &Path) -> Option { - let mut ret = PathBuf::new(); - - for component in path.components() { - match component { - Component::Prefix(..) | Component::RootDir => return None, - Component::CurDir => {} - Component::ParentDir => { - if !ret.pop() { - return None; - } - } - Component::Normal(c) => { - ret.push(c); - } - } - } - - Some(ret) -} - -fn parse_bun_relative_imports(raw_code: &str, script_path: &str) -> error::Result> { - let mut relative_imports = vec![]; - let r = parse_expr_for_imports(raw_code)?; - for import in r { - let import = import.trim_end_matches(".ts"); - if import.starts_with("/") { - relative_imports.push(import.trim_start_matches("/").to_string()); - } else if import.starts_with(".") { - let normalized = try_normalize(std::path::Path::new(&format!( - "{}/../{}", - script_path, import - ))); - if let Some(normalized) = normalized { - let normalized = normalized.to_str().unwrap().to_string(); - relative_imports.push(normalized); - } else { - tracing::error!("error canonicalizing path: {:?}", normalized); - } - } - } - - Ok(relative_imports) -} - -#[tracing::instrument(level = "trace", skip_all)] -async fn handle_dependency_job( - job: &QueuedJob, - mem_peak: &mut i32, - canceled_by: &mut Option, - job_dir: &str, - db: &sqlx::Pool, - worker_name: &str, - worker_dir: &str, - base_internal_url: &str, - token: &str, - rsmq: Option, -) -> error::Result> { - let raw_code = match job.raw_code { - Some(ref code) => code.to_owned(), - None => sqlx::query_scalar!( - "SELECT content FROM script WHERE hash = $1 AND workspace_id = $2", - &job.script_hash.unwrap_or(ScriptHash(0)).0, - &job.workspace_id - ) - .fetch_optional(db) - .await? - .unwrap_or_else(|| "No script found at this hash".to_string()), - }; - - let script_path = job.script_path(); - let raw_deps = job - .args - .as_ref() - .map(|x| { - x.get("raw_deps") - .is_some_and(|y| y.to_string().as_str() == "true") - }) - .unwrap_or(false); - - let content = capture_dependency_job( - &job.id, - job.language.as_ref().map(|v| Ok(v)).unwrap_or_else(|| { - Err(Error::InternalErr( - "Job Language required for dependency jobs".to_owned(), - )) - })?, - &raw_code, - mem_peak, - canceled_by, - job_dir, - db, - worker_name, - &job.workspace_id, - worker_dir, - base_internal_url, - token, - script_path, - raw_deps, - ) - .await; - - match content { - Ok(content) => { - if job.script_hash.is_none() { - // it a one-off raw script dependency job, no need to update the db - return Ok(to_raw_value_owned( - json!({ "success": "Successful lock file generation", "lock": content }), - )); - } - - let hash = job.script_hash.unwrap_or(ScriptHash(0)); - let w_id = &job.workspace_id; - sqlx::query!( - "UPDATE script SET lock = $1 WHERE hash = $2 AND workspace_id = $3", - &content, - &hash.0, - w_id - ) - .execute(db) - .await?; - - let (deployment_message, parent_path) = - get_deployment_msg_and_parent_path_from_args(job.args.clone()); - - if let Err(e) = handle_deployment_metadata( - &job.email, - &job.created_by, - &db, - &w_id, - DeployedObject::Script { - hash, - path: script_path.to_string(), - parent_path: parent_path.clone(), - }, - deployment_message.clone(), - rsmq.clone(), - false, - ) - .await - { - tracing::error!(%e, "error handling deployment metadata"); - } - - let relative_imports = match job.language { - Some(ScriptLang::Python3) => parse_relative_imports(&raw_code, script_path).ok(), - Some(ScriptLang::Bun) => parse_bun_relative_imports(&raw_code, script_path).ok(), - _ => None, - }; - if let Some(relative_imports) = relative_imports { - if !relative_imports.is_empty() { - let mut logs = "".to_string(); - logs.push_str("\n--- RELATIVE IMPORTS ---\n\n"); - logs.push_str(&relative_imports.join("\n")); - - let mut tx = db.begin().await?; - sqlx::query!( - "DELETE FROM dependency_map - WHERE importer_path = $1 AND importer_kind = 'script' - AND workspace_id = $2", - script_path, - w_id - ) - .execute(&mut *tx) - .await?; - if parent_path - .as_ref() - .is_some_and(|x| !x.is_empty() && x != script_path) - { - sqlx::query!( - "DELETE FROM dependency_map - WHERE importer_path = $1 AND importer_kind = 'script' - AND workspace_id = $2", - parent_path.clone().unwrap(), - w_id - ) - .execute(&mut *tx) - .await?; - } - - for import in relative_imports { - sqlx::query!( - "INSERT INTO dependency_map (workspace_id, importer_path, importer_kind, imported_path) - VALUES ($1, $2, 'script', $3)", - w_id, - script_path, - import - ) - .execute(&mut *tx) - .await?; - logs.push_str(&format!("{}\n", import)); - } - tx.commit().await?; - append_logs(job.id, job.workspace_id.clone(), logs, db).await; - } - - if let Err(e) = trigger_dependents_to_recompute_dependencies( - w_id, - script_path, - deployment_message, - parent_path, - &job.email, - &job.created_by, - &job.permissioned_as, - db, - rsmq, - ) - .await - { - tracing::error!(%e, "error triggering dependents to recompute dependencies"); - } - } - - Ok(to_raw_value_owned( - json!({ "success": "Successful lock file generation", "lock": content }), - )) - } - Err(error) => { - let logs2 = sqlx::query_scalar!( - "SELECT logs FROM job_logs WHERE job_id = $1 AND workspace_id = $2", - &job.id, - &job.workspace_id - ) - .fetch_optional(db) - .await? - .flatten() - .unwrap_or_else(|| "no logs".to_string()); - sqlx::query!( - "UPDATE script SET lock_error_logs = $1 WHERE hash = $2 AND workspace_id = $3", - &format!("{logs2}\n{error}"), - &job.script_hash.unwrap_or(ScriptHash(0)).0, - &job.workspace_id - ) - .execute(db) - .await?; - Err(Error::ExecutionErr(format!("Error locking file: {error}")))? - } - } -} - -async fn trigger_dependents_to_recompute_dependencies< - R: rsmq_async::RsmqConnection + Send + Sync + Clone, ->( - w_id: &str, - script_path: &str, - deployment_message: Option, - parent_path: Option, - email: &str, - created_by: &str, - permissioned_as: &str, - db: &sqlx::Pool, - rsmq: Option, -) -> error::Result<()> { - let script_importers = sqlx::query_scalar!( - "SELECT importer_path FROM dependency_map - WHERE imported_path = $1 AND importer_kind = 'script' - AND workspace_id = $2", - script_path, - w_id - ) - .fetch_all(db) - .await?; - for s in script_importers.iter() { - let tx: PushIsolationLevel<'_, R> = - PushIsolationLevel::IsolatedRoot(db.clone(), rsmq.clone()); - let r = get_latest_deployed_hash_for_path(db, w_id, s.as_str()).await; - if let Ok(r) = r { - let mut args: HashMap> = HashMap::new(); - if let Some(ref dm) = deployment_message { - args.insert("deployment_message".to_string(), to_raw_value(&dm)); - } - if let Some(ref p_path) = parent_path { - args.insert("common_dependency_path".to_string(), to_raw_value(&p_path)); - } - - let (job_uuid, new_tx) = windmill_queue::push( - db, - tx, - &w_id, - JobPayload::Dependencies { - path: s.clone(), - hash: r.0, - language: r.6, - dedicated_worker: r.7, - }, - windmill_queue::PushArgs { args, extra: HashMap::new() }, - &created_by, - email, - permissioned_as.to_string(), - None, - None, - None, - None, - None, - false, - false, - None, - true, - None, - None, - None, - None, - ) - .await?; - tracing::info!( - "pushed dependency job due to common python path: {job_uuid} for path {path} with hash {hash}", - path = s, - hash = r.0 - ); - new_tx.commit().await?; - } else { - tracing::error!( - "error getting latest deployed hash for path {path}: {err}", - path = s, - err = r.unwrap_err() - ); - } - } - Ok(()) -} - -async fn handle_flow_dependency_job( - job: &QueuedJob, - mem_peak: &mut i32, - canceled_by: &mut Option, - job_dir: &str, - db: &sqlx::Pool, - worker_name: &str, - worker_dir: &str, - base_internal_url: &str, - token: &str, - rsmq: Option, -) -> error::Result<()> { - let job_path = job.script_path.clone().ok_or_else(|| { - error::Error::InternalErr( - "Cannot resolve flow dependencies for flow without path".to_string(), - ) - })?; - let raw_flow = job.raw_flow.clone().map(|v| Ok(v)).unwrap_or_else(|| { - Err(Error::InternalErr( - "Flow Dependency requires raw flow".to_owned(), - )) - })?; - let mut flow = serde_json::from_str::((*raw_flow.0).get()).map_err(to_anyhow)?; - - flow.modules = lock_modules( - flow.modules, - job, - mem_peak, - canceled_by, - job_dir, - db, - worker_name, - worker_dir, - &job_path, - base_internal_url, - token, - ) - .await?; - let new_flow_value = serde_json::to_value(flow).map_err(to_anyhow)?; - - // Re-check cancelation to ensure we don't accidentially override a flow. - if sqlx::query_scalar!("SELECT canceled FROM queue WHERE id = $1", job.id) - .fetch_optional(db) - .await - .map(|v| Some(true) == v) - .unwrap_or_else(|err| { - tracing::error!(%job.id, %err, "error checking cancelation for job {0}: {err}", job.id); - false - }) - { - return Ok(()); - } - - sqlx::query!( - "UPDATE flow SET value = $1 WHERE path = $2 AND workspace_id = $3", - new_flow_value, - job_path, - job.workspace_id - ) - .execute(db) - .await?; - - let (deployment_message, parent_path) = - get_deployment_msg_and_parent_path_from_args(job.args.clone()); - - if let Err(e) = handle_deployment_metadata( - &job.email, - &job.created_by, - &db, - &job.workspace_id, - DeployedObject::Flow { path: job_path, parent_path }, - deployment_message, - rsmq.clone(), - false, - ) - .await - { - tracing::error!(%e, "error handling deployment metadata"); - } - - Ok(()) -} - -fn get_deployment_msg_and_parent_path_from_args( - args: Option>>>, -) -> (Option, Option) { - let args_map = args.map(|json_hashmap| json_hashmap.0); - let deployment_message = args_map - .clone() - .map(|hashmap| { - hashmap - .get("deployment_message") - .map(|map_value| serde_json::from_str::(map_value.get()).ok()) - .flatten() - }) - .flatten(); - let parent_path = args_map - .clone() - .map(|hashmap| { - hashmap - .get("parent_path") - .map(|map_value| serde_json::from_str::(map_value.get()).ok()) - .flatten() - }) - .flatten(); - (deployment_message, parent_path) -} - -#[async_recursion] -async fn lock_modules( - modules: Vec, - job: &QueuedJob, - mem_peak: &mut i32, - canceled_by: &mut Option, - job_dir: &str, - db: &sqlx::Pool, - worker_name: &str, - worker_dir: &str, - job_path: &str, - base_internal_url: &str, - token: &str, -) -> Result> { - let mut new_flow_modules = Vec::new(); - for mut e in modules.into_iter() { - let FlowModuleValue::RawScript { - lock, - path, - content, - language, - input_transforms, - tag, - custom_concurrency_key, - concurrent_limit, - concurrency_time_window_s, - } = e.get_value()? - else { - match e.get_value()? { - FlowModuleValue::ForloopFlow { - iterator, - modules, - skip_failures, - parallel, - parallelism, - } => { - e.value = FlowModuleValue::ForloopFlow { - iterator, - modules: lock_modules( - modules, - job, - mem_peak, - canceled_by, - job_dir, - db, - worker_name, - worker_dir, - job_path, - base_internal_url, - token, - ) - .await?, - skip_failures, - parallel, - parallelism, - } - .into() - } - FlowModuleValue::BranchAll { branches, parallel } => { - let mut nbranches = vec![]; - for mut b in branches { - b.modules = lock_modules( - b.modules, - job, - mem_peak, - canceled_by, - job_dir, - db, - worker_name, - worker_dir, - job_path, - base_internal_url, - token, - ) - .await?; - nbranches.push(b) - } - e.value = FlowModuleValue::BranchAll { branches: nbranches, parallel }.into() - } - FlowModuleValue::WhileloopFlow { modules, skip_failures } => { - e.value = FlowModuleValue::WhileloopFlow { - modules: lock_modules( - modules, - job, - mem_peak, - canceled_by, - job_dir, - db, - worker_name, - worker_dir, - job_path, - base_internal_url, - token, - ) - .await?, - skip_failures, - } - .into() - } - FlowModuleValue::BranchOne { branches, default } => { - let mut nbranches = vec![]; - for mut b in branches { - b.modules = lock_modules( - b.modules, - job, - mem_peak, - canceled_by, - job_dir, - db, - worker_name, - worker_dir, - job_path, - base_internal_url, - token, - ) - .await?; - nbranches.push(b) - } - let default = lock_modules( - default, - job, - mem_peak, - canceled_by, - job_dir, - db, - worker_name, - worker_dir, - job_path, - base_internal_url, - token, - ) - .await?; - e.value = FlowModuleValue::BranchOne { branches: nbranches, default }.into(); - } - _ => (), - }; - new_flow_modules.push(e); - continue; - }; - if lock.as_ref().is_some_and(|x| !x.trim().is_empty()) { - new_flow_modules.push(e); - continue; - } - let new_lock = capture_dependency_job( - &job.id, - &language, - &content, - mem_peak, - canceled_by, - job_dir, - db, - worker_name, - &job.workspace_id, - worker_dir, - base_internal_url, - token, - &path.clone().unwrap_or_else(|| job_path.to_string()), - false, - ) - .await; - match new_lock { - Ok(new_lock) => { - e.value = windmill_common::worker::to_raw_value(&FlowModuleValue::RawScript { - lock: Some(new_lock), - path, - input_transforms, - content, - language, - tag, - custom_concurrency_key, - concurrent_limit, - concurrency_time_window_s, - }); - new_flow_modules.push(e); - continue; - } - Err(error) => { - // TODO: Record flow raw script error lock logs - tracing::warn!( - path = path, - language = ?language, - error = ?error, - "Failed to generate flow lock for raw script" - ); - e.value = windmill_common::worker::to_raw_value(&FlowModuleValue::RawScript { - lock: None, - path, - input_transforms, - content, - language, - tag, - custom_concurrency_key, - concurrent_limit, - concurrency_time_window_s, - }); - new_flow_modules.push(e); - continue; - } - } - } - Ok(new_flow_modules) -} - -#[async_recursion] -async fn lock_modules_app( - value: Value, - job: &QueuedJob, - mem_peak: &mut i32, - canceled_by: &mut Option, - job_dir: &str, - db: &sqlx::Pool, - worker_name: &str, - worker_dir: &str, - job_path: &str, - base_internal_url: &str, - token: &str, -) -> Result { - match value { - Value::Object(mut m) => { - if m.contains_key("inlineScript") { - let v = m.get_mut("inlineScript").unwrap(); - if let Some(v) = v.as_object_mut() { - if v.contains_key("content") && v.contains_key("language") { - if let Ok(language) = - serde_json::from_value::(v.get("language").unwrap().clone()) - { - let content = v - .get("content") - .unwrap() - .as_str() - .unwrap_or_default() - .to_string(); - let mut logs = "".to_string(); - if v.get("lock") - .is_some_and(|x| !x.as_str().unwrap().trim().is_empty()) - { - logs.push_str( - "Found already locked inline script. Skipping lock...\n", - ); - return Ok(Value::Object(m.clone())); - } - logs.push_str("Found lockable inline script. Generating lock...\n"); - let new_lock = capture_dependency_job( - &job.id, - &language, - &content, - mem_peak, - canceled_by, - job_dir, - db, - worker_name, - &job.workspace_id, - worker_dir, - base_internal_url, - token, - job.script_path(), - false, - ) - .await; - match new_lock { - Ok(new_lock) => { - append_logs(job.id, job.workspace_id.clone(), logs, db).await; - v.insert( - "lock".to_string(), - serde_json::Value::String(new_lock), - ); - return Ok(Value::Object(m.clone())); - } - Err(e) => { - tracing::warn!( - language = ?language, - error = ?e, - logs = ?logs, - "Failed to generate flow lock for inline script" - ); - () - } - } - } - } - } - } - for (a, b) in m.clone().into_iter() { - m.insert( - a.clone(), - lock_modules_app( - b, - job, - mem_peak, - canceled_by, - job_dir, - db, - worker_name, - worker_dir, - job_path, - base_internal_url, - token, - ) - .await?, - ); - } - Ok(Value::Object(m)) - } - Value::Array(a) => { - let mut nv = vec![]; - for b in a.clone().into_iter() { - nv.push( - lock_modules_app( - b, - job, - mem_peak, - canceled_by, - job_dir, - db, - worker_name, - worker_dir, - job_path, - base_internal_url, - token, - ) - .await?, - ); - } - Ok(Value::Array(nv)) - } - a @ _ => Ok(a), - } -} - -async fn handle_app_dependency_job( - job: &QueuedJob, - mem_peak: &mut i32, - canceled_by: &mut Option, - job_dir: &str, - db: &sqlx::Pool, - worker_name: &str, - worker_dir: &str, - base_internal_url: &str, - token: &str, - rsmq: Option, -) -> error::Result<()> { - let job_path = job.script_path.clone().ok_or_else(|| { - error::Error::InternalErr( - "Cannot resolve flow dependencies for flow without path".to_string(), - ) - })?; - - let id = job - .script_hash - .clone() - .ok_or_else(|| Error::InternalErr("Flow Dependency requires script hash".to_owned()))? - .0; - let value = sqlx::query_scalar!("SELECT value FROM app_version WHERE id = $1", id) - .fetch_optional(db) - .await?; - - if let Some(value) = value { - let value = lock_modules_app( - value, - job, - mem_peak, - canceled_by, - job_dir, - db, - worker_name, - worker_dir, - &job_path, - base_internal_url, - token, - ) - .await?; - - // Re-check cancelation to ensure we don't accidentially override a flow. - if sqlx::query_scalar!("SELECT canceled FROM queue WHERE id = $1", job.id) - .fetch_optional(db) - .await - .map(|v| Some(true) == v) - .unwrap_or_else(|err| { - tracing::error!(%job.id, %err, "error checking cancelation for job {0}: {err}", job.id); - false - }) - { - return Ok(()); - } - - sqlx::query!("UPDATE app_version SET value = $1 WHERE id = $2", value, id,) - .execute(db) - .await?; - - let (deployment_message, parent_path) = - get_deployment_msg_and_parent_path_from_args(job.args.clone()); - - if let Err(e) = handle_deployment_metadata( - &job.email, - &job.created_by, - &db, - &job.workspace_id, - DeployedObject::App { path: job_path, version: id, parent_path }, - deployment_message, - rsmq.clone(), - false, - ) - .await - { - tracing::error!(%e, "error handling deployment metadata"); - } - - // tx = PushIsolationLevel::Transaction(new_tx); - // tx = handle_deployment_metadata( - // tx, - // &authed, - // &db, - // &w_id, - // DeployedObject::App { path: app.path.clone(), version: v_id }, - // app.deployment_message, - // ) - // .await?; - - // match tx { - // PushIsolationLevel::Transaction(tx) => tx.commit().await?, - // _ => { - // return Err(Error::InternalErr( - // "Expected a transaction here".to_string(), - // )); - // } - // } - - Ok(()) - } else { - Ok(()) - } -} - -async fn capture_dependency_job( - job_id: &Uuid, - job_language: &ScriptLang, - job_raw_code: &str, - mem_peak: &mut i32, - canceled_by: &mut Option, - job_dir: &str, - db: &sqlx::Pool, - worker_name: &str, - w_id: &str, - worker_dir: &str, - base_internal_url: &str, - token: &str, - script_path: &str, - raw_deps: bool, -) -> error::Result { - match job_language { - ScriptLang::Python3 => { - let reqs = if raw_deps { - job_raw_code.to_string() - } else { - let mut already_visited = vec![]; - - windmill_parser_py_imports::parse_python_imports( - job_raw_code, - &w_id, - script_path, - &db, - &mut already_visited, - ) - .await? - .join("\n") - }; - create_dependencies_dir(job_dir).await; - let req: std::result::Result = pip_compile( - job_id, - &reqs, - mem_peak, - canceled_by, - job_dir, - db, - worker_name, - w_id, - ) - .await; - // install the dependencies to pre-fill the cache - if let Ok(req) = req.as_ref() { - let r = handle_python_reqs( - req.split("\n").filter(|x| !x.starts_with("--")).collect(), - job_id, - w_id, - mem_peak, - canceled_by, - db, - worker_name, - job_dir, - worker_dir, - ) - .await; - - if let Err(e) = r { - tracing::error!( - "Failed to install python dependencies to prefill the cache: {:?} \n", - e - ); - } - } - req - } - ScriptLang::Go => { - if raw_deps { - return Err(Error::ExecutionErr( - "Raw dependencies not supported for go".to_string(), - )); - } - install_go_dependencies( - job_id, - job_raw_code, - mem_peak, - canceled_by, - job_dir, - db, - false, - false, - false, - worker_name, - w_id, - ) - .await - } - ScriptLang::Deno => { - if raw_deps { - return Err(Error::ExecutionErr( - "Raw dependencies not supported for deno".to_string(), - )); - } - generate_deno_lock( - job_id, - job_raw_code, - mem_peak, - canceled_by, - job_dir, - db, - w_id, - worker_name, - base_internal_url, - ) - .await - } - ScriptLang::Bun => { - let trusted_deps = if !raw_deps { - let _ = write_file(job_dir, "main.ts", job_raw_code).await?; - //TODO: remove once bun provides sane default fot it - get_trusted_deps(job_raw_code) - } else { - vec![] - }; - let req = gen_lockfile( - mem_peak, - canceled_by, - job_id, - w_id, - db, - token, - script_path, - job_dir, - base_internal_url, - worker_name, - true, - trusted_deps, - if raw_deps { - Some(job_raw_code.to_string()) - } else { - None - }, - false, - ) - .await?; - Ok(req.unwrap_or_else(String::new)) - } - ScriptLang::Php => { - let reqs = if raw_deps { - if job_raw_code.is_empty() { - return Ok("".to_string()); - } - job_raw_code.to_string() - } else { - match parse_php_imports(job_raw_code)? { - Some(reqs) => reqs, - None => { - return Ok("".to_string()); - } - } - }; - - composer_install( - mem_peak, - canceled_by, - job_id, - w_id, - db, - job_dir, - worker_name, - reqs, - None, - ) - .await - } - ScriptLang::Postgresql => Ok("".to_owned()), - ScriptLang::Mysql => Ok("".to_owned()), - ScriptLang::Bigquery => Ok("".to_owned()), - ScriptLang::Snowflake => Ok("".to_owned()), - ScriptLang::Mssql => Ok("".to_owned()), - ScriptLang::Graphql => Ok("".to_owned()), - ScriptLang::Bash => Ok("".to_owned()), - ScriptLang::Powershell => Ok("".to_owned()), - ScriptLang::Nativets => Ok("".to_owned()), - } -} diff --git a/backend/windmill-worker/src/worker_flow.rs b/backend/windmill-worker/src/worker_flow.rs index 338144b540..aef56aaa45 100644 --- a/backend/windmill-worker/src/worker_flow.rs +++ b/backend/windmill-worker/src/worker_flow.rs @@ -742,7 +742,7 @@ pub async fn update_flow_status_after_job_completion_internal< } else { "Flow job completed with error\n".to_string() }; - append_logs(flow_job.id, w_id.to_string(), logs, db).await; + append_logs(&flow_job.id, w_id, logs, db).await; } #[cfg(feature = "enterprise")] if flow_job.parent_job.is_none() { @@ -852,8 +852,8 @@ pub async fn update_flow_status_after_job_completion_internal< Err(err) => { let e = json!({"message": err.to_string(), "name": "InternalError"}); append_logs( - flow_job.id, - w_id.to_string(), + &flow_job.id, + w_id, format!("Unexpected error during flow chaining:\n{:#?}", e), db, ) @@ -1681,7 +1681,7 @@ async fn push_next_flow_job let result: Value = json!({ "error": {"message": logs, "name": error_name}}); - append_logs(flow_job.id, flow_job.workspace_id.clone(), logs.clone(), db).await; + append_logs(&flow_job.id, &flow_job.workspace_id, logs.clone(), db).await; job_completed_tx .send(SendResult::UpdateFlow { diff --git a/backend/windmill-worker/src/worker_lockfiles.rs b/backend/windmill-worker/src/worker_lockfiles.rs new file mode 100644 index 0000000000..eb724d0223 --- /dev/null +++ b/backend/windmill-worker/src/worker_lockfiles.rs @@ -0,0 +1,1270 @@ +use std::collections::HashMap; +use std::path::{Component, Path, PathBuf}; + +use async_recursion::async_recursion; +use serde_json::value::RawValue; +use serde_json::{json, Value}; +use sqlx::types::Json; +use uuid::Uuid; +use windmill_common::error::Error; +use windmill_common::error::Result; +use windmill_common::flows::{FlowModule, FlowModuleValue}; +use windmill_common::get_latest_deployed_hash_for_path; +use windmill_common::jobs::JobPayload; +use windmill_common::scripts::ScriptHash; +use windmill_common::worker::{to_raw_value, to_raw_value_owned}; +use windmill_common::{ + error::{self, to_anyhow}, + flows::FlowValue, + jobs::QueuedJob, + scripts::ScriptLang, + DB, +}; +use windmill_git_sync::{handle_deployment_metadata, DeployedObject}; +use windmill_parser_py_imports::parse_relative_imports; +use windmill_parser_ts::parse_expr_for_imports; +use windmill_queue::{append_logs, CanceledBy, PushIsolationLevel}; + +use crate::python_executor::{create_dependencies_dir, handle_python_reqs, pip_compile}; +use crate::{ + bun_executor::{gen_lockfile, get_trusted_deps}, + common::write_file, + deno_executor::generate_deno_lock, + go_executor::install_go_dependencies, + php_executor::{composer_install, parse_php_imports}, +}; + +pub async fn update_script_dependency_map( + job_id: &Uuid, + db: &DB, + w_id: &str, + parent_path: &Option, + script_path: &str, + relative_imports: Vec, +) -> error::Result<()> { + let importer_kind = "script"; + if !relative_imports.is_empty() { + let mut logs = "".to_string(); + logs.push_str("\n--- RELATIVE IMPORTS ---\n\n"); + logs.push_str(&relative_imports.join("\n")); + + let mut tx = db.begin().await?; + tx = + clear_dependency_parent_path(parent_path, script_path, w_id, importer_kind, tx).await?; + + tx = clear_dependency_map_for_item(script_path, w_id, importer_kind, tx, &None).await?; + + tx = add_relative_imports_to_dependency_map( + script_path, + w_id, + relative_imports, + importer_kind, + tx, + &mut logs, + None, + ) + .await?; + tx.commit().await?; + append_logs(job_id, w_id, logs, db).await; + } + Ok(()) +} + +async fn add_relative_imports_to_dependency_map<'c>( + script_path: &str, + w_id: &str, + relative_imports: Vec, + importer_kind: &str, + mut tx: sqlx::Transaction<'c, sqlx::Postgres>, + logs: &mut String, + node_id: Option, +) -> error::Result> { + for import in relative_imports { + sqlx::query!( + "INSERT INTO dependency_map (workspace_id, importer_path, importer_kind, imported_path, importer_node_id) + VALUES ($1, $2, $4::text::IMPORTER_KIND, $3, $5)", + w_id, + script_path, + import, + importer_kind, + node_id + ) + .execute(&mut *tx) + .await?; + logs.push_str(&format!("{}\n", import)); + } + Ok(tx) +} + +async fn clear_dependency_map_for_item<'c>( + item_path: &str, + w_id: &str, + importer_kind: &str, + mut tx: sqlx::Transaction<'c, sqlx::Postgres>, + importer_node_id: &Option, +) -> Result> { + sqlx::query!( + "DELETE FROM dependency_map + WHERE importer_path = $1 AND importer_kind = $3::text::IMPORTER_KIND + AND workspace_id = $2 AND importer_node_id = $4", + item_path, + w_id, + importer_kind, + importer_node_id.clone() + ) + .execute(&mut *tx) + .await?; + Ok(tx) +} + +async fn clear_dependency_parent_path<'c>( + parent_path: &Option, + item_path: &str, + w_id: &str, + importer_kind: &str, + mut tx: sqlx::Transaction<'c, sqlx::Postgres>, +) -> Result> { + if parent_path + .as_ref() + .is_some_and(|x| !x.is_empty() && x != item_path) + { + sqlx::query!( + "DELETE FROM dependency_map + WHERE importer_path = $1 AND importer_kind = $3::text::IMPORTER_KIND + AND workspace_id = $2", + parent_path.clone().unwrap(), + w_id, + importer_kind + ) + .execute(&mut *tx) + .await?; + } + Ok(tx) +} + +fn try_normalize(path: &Path) -> Option { + let mut ret = PathBuf::new(); + + for component in path.components() { + match component { + Component::Prefix(..) | Component::RootDir => return None, + Component::CurDir => {} + Component::ParentDir => { + if !ret.pop() { + return None; + } + } + Component::Normal(c) => { + ret.push(c); + } + } + } + + Some(ret) +} + +fn parse_bun_relative_imports(raw_code: &str, script_path: &str) -> error::Result> { + let mut relative_imports = vec![]; + let r = parse_expr_for_imports(raw_code)?; + for import in r { + let import = import.trim_end_matches(".ts"); + if import.starts_with("/") { + relative_imports.push(import.trim_start_matches("/").to_string()); + } else if import.starts_with(".") { + let normalized = try_normalize(std::path::Path::new(&format!( + "{}/../{}", + script_path, import + ))); + if let Some(normalized) = normalized { + let normalized = normalized.to_str().unwrap().to_string(); + relative_imports.push(normalized); + } else { + tracing::error!("error canonicalizing path: {:?}", normalized); + } + } + } + + Ok(relative_imports) +} + +fn extract_relative_imports( + raw_code: &str, + script_path: &str, + language: &Option, +) -> Option> { + match language { + Some(ScriptLang::Python3) => parse_relative_imports(&raw_code, script_path).ok(), + Some(ScriptLang::Bun) => parse_bun_relative_imports(&raw_code, script_path).ok(), + _ => None, + } +} +#[tracing::instrument(level = "trace", skip_all)] +pub async fn handle_dependency_job( + job: &QueuedJob, + mem_peak: &mut i32, + canceled_by: &mut Option, + job_dir: &str, + db: &sqlx::Pool, + worker_name: &str, + worker_dir: &str, + base_internal_url: &str, + token: &str, + rsmq: Option, +) -> error::Result> { + let raw_code = match job.raw_code { + Some(ref code) => code.to_owned(), + None => sqlx::query_scalar!( + "SELECT content FROM script WHERE hash = $1 AND workspace_id = $2", + &job.script_hash.unwrap_or(ScriptHash(0)).0, + &job.workspace_id + ) + .fetch_optional(db) + .await? + .unwrap_or_else(|| "No script found at this hash".to_string()), + }; + + let script_path = job.script_path(); + let raw_deps = job + .args + .as_ref() + .map(|x| { + x.get("raw_deps") + .is_some_and(|y| y.to_string().as_str() == "true") + }) + .unwrap_or(false); + + let content = capture_dependency_job( + &job.id, + job.language.as_ref().map(|v| Ok(v)).unwrap_or_else(|| { + Err(Error::InternalErr( + "Job Language required for dependency jobs".to_owned(), + )) + })?, + &raw_code, + mem_peak, + canceled_by, + job_dir, + db, + worker_name, + &job.workspace_id, + worker_dir, + base_internal_url, + token, + script_path, + raw_deps, + ) + .await; + + match content { + Ok(content) => { + if job.script_hash.is_none() { + // it a one-off raw script dependency job, no need to update the db + return Ok(to_raw_value_owned( + json!({ "success": "Successful lock file generation", "lock": content }), + )); + } + + let hash = job.script_hash.unwrap_or(ScriptHash(0)); + let w_id = &job.workspace_id; + sqlx::query!( + "UPDATE script SET lock = $1 WHERE hash = $2 AND workspace_id = $3", + &content, + &hash.0, + w_id + ) + .execute(db) + .await?; + + let (deployment_message, parent_path) = + get_deployment_msg_and_parent_path_from_args(job.args.clone()); + + if let Err(e) = handle_deployment_metadata( + &job.email, + &job.created_by, + &db, + &w_id, + DeployedObject::Script { + hash, + path: script_path.to_string(), + parent_path: parent_path.clone(), + }, + deployment_message.clone(), + rsmq.clone(), + false, + ) + .await + { + tracing::error!(%e, "error handling deployment metadata"); + } + + let relative_imports = extract_relative_imports(&raw_code, script_path, &job.language); + if let Some(relative_imports) = relative_imports { + update_script_dependency_map( + &job.id, + db, + w_id, + &parent_path, + script_path, + relative_imports, + ) + .await?; + if let Err(e) = trigger_dependents_to_recompute_dependencies( + w_id, + script_path, + deployment_message, + parent_path, + &job.email, + &job.created_by, + &job.permissioned_as, + db, + rsmq, + ) + .await + { + tracing::error!(%e, "error triggering dependents to recompute dependencies"); + } + } + + Ok(to_raw_value_owned( + json!({ "success": "Successful lock file generation", "lock": content }), + )) + } + Err(error) => { + let logs2 = sqlx::query_scalar!( + "SELECT logs FROM job_logs WHERE job_id = $1 AND workspace_id = $2", + &job.id, + &job.workspace_id + ) + .fetch_optional(db) + .await? + .flatten() + .unwrap_or_else(|| "no logs".to_string()); + sqlx::query!( + "UPDATE script SET lock_error_logs = $1 WHERE hash = $2 AND workspace_id = $3", + &format!("{logs2}\n{error}"), + &job.script_hash.unwrap_or(ScriptHash(0)).0, + &job.workspace_id + ) + .execute(db) + .await?; + Err(Error::ExecutionErr(format!("Error locking file: {error}")))? + } + } +} + +async fn trigger_dependents_to_recompute_dependencies< + R: rsmq_async::RsmqConnection + Send + Sync + Clone, +>( + w_id: &str, + script_path: &str, + deployment_message: Option, + parent_path: Option, + email: &str, + created_by: &str, + permissioned_as: &str, + db: &sqlx::Pool, + rsmq: Option, +) -> error::Result<()> { + let script_importers = sqlx::query!( + "SELECT importer_path, importer_kind::text, array_agg(importer_node_id) as importer_node_ids FROM dependency_map + WHERE imported_path = $1 + AND workspace_id = $2 + GROUP BY importer_path, importer_kind", + script_path, + w_id + ) + .fetch_all(db) + .await?; + for s in script_importers.iter() { + let tx: PushIsolationLevel<'_, R> = + PushIsolationLevel::IsolatedRoot(db.clone(), rsmq.clone()); + let mut args: HashMap> = HashMap::new(); + if let Some(ref dm) = deployment_message { + args.insert("deployment_message".to_string(), to_raw_value(&dm)); + } + if let Some(ref p_path) = parent_path { + args.insert("common_dependency_path".to_string(), to_raw_value(&p_path)); + } + + let kind = s.importer_kind.clone().unwrap_or_default(); + let job_payload = if kind == "script" { + let r = get_latest_deployed_hash_for_path(db, w_id, s.importer_path.as_str()).await; + if let Ok(r) = r { + JobPayload::Dependencies { + path: s.importer_path.clone(), + hash: r.0, + language: r.6, + dedicated_worker: r.7, + } + } else { + tracing::error!( + "error getting latest deployed hash for path {path}: {err}", + path = s.importer_path, + err = r.unwrap_err() + ); + continue; + } + } else if kind == "flow" { + args.insert( + "nodes_to_relock".to_string(), + to_raw_value(&s.importer_node_ids), + ); + JobPayload::FlowDependencies { path: s.importer_path.clone(), dedicated_worker: None } + } else { + tracing::error!( + "unexpected importer kind: {kind} for path {path}", + kind = kind, + path = s.importer_path + ); + continue; + }; + + let (job_uuid, new_tx) = windmill_queue::push( + db, + tx, + &w_id, + job_payload, + windmill_queue::PushArgs { args, extra: HashMap::new() }, + &created_by, + email, + permissioned_as.to_string(), + None, + None, + None, + None, + None, + false, + false, + None, + true, + None, + None, + None, + None, + ) + .await?; + tracing::info!( + "pushed dependency job due to common python path: {job_uuid} for path {path}", + path = s.importer_path, + ); + new_tx.commit().await?; + } + Ok(()) +} + +pub async fn handle_flow_dependency_job( + job: &QueuedJob, + mem_peak: &mut i32, + canceled_by: &mut Option, + job_dir: &str, + db: &sqlx::Pool, + worker_name: &str, + worker_dir: &str, + base_internal_url: &str, + token: &str, + rsmq: Option, +) -> error::Result> { + let job_path = job.script_path.clone().ok_or_else(|| { + error::Error::InternalErr( + "Cannot resolve flow dependencies for flow without path".to_string(), + ) + })?; + let raw_flow = job.raw_flow.clone().map(|v| Ok(v)).unwrap_or_else(|| { + Err(Error::InternalErr( + "Flow Dependency requires raw flow".to_owned(), + )) + })?; + let (deployment_message, parent_path) = + get_deployment_msg_and_parent_path_from_args(job.args.clone()); + + let nodes_to_relock = job + .args + .as_ref() + .map(|x| { + x.get("nodes_to_relock") + .map(|v| serde_json::from_str::>(v.get()).ok()) + .flatten() + }) + .flatten(); + + let mut flow = serde_json::from_str::((*raw_flow.0).get()).map_err(to_anyhow)?; + + let mut tx = db.begin().await?; + + tx = clear_dependency_parent_path(&parent_path, &job_path, &job.workspace_id, "flow", tx) + .await?; + let modified_ids; + (flow.modules, tx, modified_ids) = lock_modules( + flow.modules, + job, + mem_peak, + canceled_by, + job_dir, + db, + tx, + worker_name, + worker_dir, + &job_path, + base_internal_url, + token, + &nodes_to_relock, + ) + .await?; + let new_flow_value = serde_json::to_value(flow).map_err(to_anyhow)?; + + // Re-check cancelation to ensure we don't accidentially override a flow. + if sqlx::query_scalar!("SELECT canceled FROM queue WHERE id = $1", job.id) + .fetch_optional(db) + .await + .map(|v| Some(true) == v) + .unwrap_or_else(|err| { + tracing::error!(%job.id, %err, "error checking cancelation for job {0}: {err}", job.id); + false + }) + { + return Ok(to_raw_value_owned(json!({ + "status": "Flow lock generation was canceled", + }))); + } + + sqlx::query!( + "UPDATE flow SET value = $1 WHERE path = $2 AND workspace_id = $3", + new_flow_value, + job_path, + job.workspace_id + ) + .execute(db) + .await?; + tx.commit().await?; + + if let Err(e) = handle_deployment_metadata( + &job.email, + &job.created_by, + &db, + &job.workspace_id, + DeployedObject::Flow { path: job_path, parent_path }, + deployment_message, + rsmq.clone(), + false, + ) + .await + { + tracing::error!(%e, "error handling deployment metadata"); + } + + Ok(to_raw_value_owned(json!({ + "status": "Successful lock file generation", + "modified_ids": modified_ids, + }))) +} + +fn get_deployment_msg_and_parent_path_from_args( + args: Option>>>, +) -> (Option, Option) { + let args_map = args.map(|json_hashmap| json_hashmap.0); + let deployment_message = args_map + .clone() + .map(|hashmap| { + hashmap + .get("deployment_message") + .map(|map_value| serde_json::from_str::(map_value.get()).ok()) + .flatten() + }) + .flatten(); + let parent_path = args_map + .clone() + .map(|hashmap| { + hashmap + .get("parent_path") + .map(|map_value| serde_json::from_str::(map_value.get()).ok()) + .flatten() + }) + .flatten(); + (deployment_message, parent_path) +} + +async fn lock_modules<'c>( + modules: Vec, + job: &QueuedJob, + mem_peak: &mut i32, + canceled_by: &mut Option, + job_dir: &str, + db: &sqlx::Pool, + mut tx: sqlx::Transaction<'c, sqlx::Postgres>, + worker_name: &str, + worker_dir: &str, + job_path: &str, + base_internal_url: &str, + token: &str, + locks_to_reload: &Option>, + // (modules to replace old seq (even unmmodified ones), new transaction, modified ids) ) +) -> Result<( + Vec, + sqlx::Transaction<'c, sqlx::Postgres>, + Vec, +)> { + let mut new_flow_modules = Vec::new(); + let mut modified_ids = Vec::new(); + for mut e in modules.into_iter() { + let mut nmodified_ids = Vec::new(); + let FlowModuleValue::RawScript { + lock, + path, + content, + language, + input_transforms, + tag, + custom_concurrency_key, + concurrent_limit, + concurrency_time_window_s, + } = e.get_value()? + else { + match e.get_value()? { + FlowModuleValue::ForloopFlow { + iterator, + modules, + skip_failures, + parallel, + parallelism, + } => { + let nmodules; + (nmodules, tx, nmodified_ids) = Box::pin(lock_modules( + modules, + job, + mem_peak, + canceled_by, + job_dir, + db, + tx, + worker_name, + worker_dir, + job_path, + base_internal_url, + token, + locks_to_reload, + )) + .await?; + e.value = FlowModuleValue::ForloopFlow { + iterator, + modules: nmodules, + skip_failures, + parallel, + parallelism, + } + .into() + } + FlowModuleValue::BranchAll { branches, parallel } => { + let mut nbranches = vec![]; + nmodified_ids = vec![]; + for mut b in branches { + let nmodules; + let inner_modified_ids; + (nmodules, tx, inner_modified_ids) = Box::pin(lock_modules( + b.modules, + job, + mem_peak, + canceled_by, + job_dir, + db, + tx, + worker_name, + worker_dir, + job_path, + base_internal_url, + token, + locks_to_reload, + )) + .await?; + nmodified_ids.extend(inner_modified_ids); + b.modules = nmodules; + nbranches.push(b) + } + e.value = FlowModuleValue::BranchAll { branches: nbranches, parallel }.into() + } + FlowModuleValue::WhileloopFlow { modules, skip_failures } => { + let nmodules; + (nmodules, tx, nmodified_ids) = Box::pin(lock_modules( + modules, + job, + mem_peak, + canceled_by, + job_dir, + db, + tx, + worker_name, + worker_dir, + job_path, + base_internal_url, + token, + locks_to_reload, + )) + .await?; + e.value = + FlowModuleValue::WhileloopFlow { modules: nmodules, skip_failures }.into() + } + FlowModuleValue::BranchOne { branches, default } => { + let mut nbranches = vec![]; + nmodified_ids = vec![]; + for mut b in branches { + let nmodules; + let inner_modified_ids; + + (nmodules, tx, inner_modified_ids) = Box::pin(lock_modules( + b.modules, + job, + mem_peak, + canceled_by, + job_dir, + db, + tx, + worker_name, + worker_dir, + job_path, + base_internal_url, + token, + locks_to_reload, + )) + .await?; + nmodified_ids.extend(inner_modified_ids); + b.modules = nmodules; + nbranches.push(b) + } + let ndefault; + (ndefault, tx, nmodified_ids) = Box::pin(lock_modules( + default, + job, + mem_peak, + canceled_by, + job_dir, + db, + tx, + worker_name, + worker_dir, + job_path, + base_internal_url, + token, + locks_to_reload, + )) + .await?; + e.value = FlowModuleValue::BranchOne { branches: nbranches, default: ndefault } + .into(); + } + _ => (), + }; + modified_ids.extend(nmodified_ids); + new_flow_modules.push(e); + continue; + }; + + if let Some(locks_to_reload) = locks_to_reload { + if !locks_to_reload.contains(&e.id) { + new_flow_modules.push(e); + continue; + } + } else { + if lock.as_ref().is_some_and(|x| !x.trim().is_empty()) { + new_flow_modules.push(e); + continue; + } + } + + modified_ids.push(e.id.clone()); + + let new_lock = capture_dependency_job( + &job.id, + &language, + &content, + mem_peak, + canceled_by, + job_dir, + db, + worker_name, + &job.workspace_id, + worker_dir, + base_internal_url, + token, + &path.clone().unwrap_or_else(|| job_path.to_string()), + false, + ) + .await; + // + match new_lock { + Ok(new_lock) => { + let dep_path = path.clone().unwrap_or_else(|| job_path.to_string()); + tx = clear_dependency_map_for_item( + &job_path, + &job.workspace_id, + "flow", + tx, + &Some(e.id.clone()), + ) + .await?; + let relative_imports = + extract_relative_imports(&content, &dep_path, &Some(language.clone())); + if let Some(relative_imports) = relative_imports { + let mut logs = "".to_string(); + logs.push_str(format!("\n\n--- RELATIVE IMPORTS of {} ---\n\n", e.id).as_str()); + + tx = add_relative_imports_to_dependency_map( + &dep_path, + &job.workspace_id, + relative_imports, + "flow", + tx, + &mut logs, + Some(e.id.clone()), + ) + .await?; + append_logs(&job.id, &job.workspace_id, logs, db).await; + } + + e.value = windmill_common::worker::to_raw_value(&FlowModuleValue::RawScript { + lock: Some(new_lock), + path, + input_transforms, + content, + language, + tag, + custom_concurrency_key, + concurrent_limit, + concurrency_time_window_s, + }); + new_flow_modules.push(e); + continue; + } + Err(error) => { + // TODO: Record flow raw script error lock logs + tracing::warn!( + path = path, + language = ?language, + error = ?error, + "Failed to generate flow lock for raw script" + ); + e.value = windmill_common::worker::to_raw_value(&FlowModuleValue::RawScript { + lock: None, + path, + input_transforms, + content, + language, + tag, + custom_concurrency_key, + concurrent_limit, + concurrency_time_window_s, + }); + new_flow_modules.push(e); + continue; + } + } + } + Ok((new_flow_modules, tx, modified_ids)) +} + +#[async_recursion] +async fn lock_modules_app( + value: Value, + job: &QueuedJob, + mem_peak: &mut i32, + canceled_by: &mut Option, + job_dir: &str, + db: &sqlx::Pool, + worker_name: &str, + worker_dir: &str, + job_path: &str, + base_internal_url: &str, + token: &str, +) -> Result { + match value { + Value::Object(mut m) => { + if m.contains_key("inlineScript") { + let v = m.get_mut("inlineScript").unwrap(); + if let Some(v) = v.as_object_mut() { + if v.contains_key("content") && v.contains_key("language") { + if let Ok(language) = + serde_json::from_value::(v.get("language").unwrap().clone()) + { + let content = v + .get("content") + .unwrap() + .as_str() + .unwrap_or_default() + .to_string(); + let mut logs = "".to_string(); + if v.get("lock") + .is_some_and(|x| !x.as_str().unwrap().trim().is_empty()) + { + logs.push_str( + "Found already locked inline script. Skipping lock...\n", + ); + return Ok(Value::Object(m.clone())); + } + logs.push_str("Found lockable inline script. Generating lock...\n"); + let new_lock = capture_dependency_job( + &job.id, + &language, + &content, + mem_peak, + canceled_by, + job_dir, + db, + worker_name, + &job.workspace_id, + worker_dir, + base_internal_url, + token, + job.script_path(), + false, + ) + .await; + match new_lock { + Ok(new_lock) => { + append_logs(&job.id, &job.workspace_id, logs, db).await; + v.insert( + "lock".to_string(), + serde_json::Value::String(new_lock), + ); + return Ok(Value::Object(m.clone())); + } + Err(e) => { + tracing::warn!( + language = ?language, + error = ?e, + logs = ?logs, + "Failed to generate flow lock for inline script" + ); + () + } + } + } + } + } + } + for (a, b) in m.clone().into_iter() { + m.insert( + a.clone(), + lock_modules_app( + b, + job, + mem_peak, + canceled_by, + job_dir, + db, + worker_name, + worker_dir, + job_path, + base_internal_url, + token, + ) + .await?, + ); + } + Ok(Value::Object(m)) + } + Value::Array(a) => { + let mut nv = vec![]; + for b in a.clone().into_iter() { + nv.push( + lock_modules_app( + b, + job, + mem_peak, + canceled_by, + job_dir, + db, + worker_name, + worker_dir, + job_path, + base_internal_url, + token, + ) + .await?, + ); + } + Ok(Value::Array(nv)) + } + a @ _ => Ok(a), + } +} + +pub async fn handle_app_dependency_job( + job: &QueuedJob, + mem_peak: &mut i32, + canceled_by: &mut Option, + job_dir: &str, + db: &sqlx::Pool, + worker_name: &str, + worker_dir: &str, + base_internal_url: &str, + token: &str, + rsmq: Option, +) -> error::Result<()> { + let job_path = job.script_path.clone().ok_or_else(|| { + error::Error::InternalErr( + "Cannot resolve flow dependencies for flow without path".to_string(), + ) + })?; + + let id = job + .script_hash + .clone() + .ok_or_else(|| Error::InternalErr("Flow Dependency requires script hash".to_owned()))? + .0; + let value = sqlx::query_scalar!("SELECT value FROM app_version WHERE id = $1", id) + .fetch_optional(db) + .await?; + + if let Some(value) = value { + let value = lock_modules_app( + value, + job, + mem_peak, + canceled_by, + job_dir, + db, + worker_name, + worker_dir, + &job_path, + base_internal_url, + token, + ) + .await?; + + // Re-check cancelation to ensure we don't accidentially override a flow. + if sqlx::query_scalar!("SELECT canceled FROM queue WHERE id = $1", job.id) + .fetch_optional(db) + .await + .map(|v| Some(true) == v) + .unwrap_or_else(|err| { + tracing::error!(%job.id, %err, "error checking cancelation for job {0}: {err}", job.id); + false + }) + { + return Ok(()); + } + + sqlx::query!("UPDATE app_version SET value = $1 WHERE id = $2", value, id,) + .execute(db) + .await?; + + let (deployment_message, parent_path) = + get_deployment_msg_and_parent_path_from_args(job.args.clone()); + + if let Err(e) = handle_deployment_metadata( + &job.email, + &job.created_by, + &db, + &job.workspace_id, + DeployedObject::App { path: job_path, version: id, parent_path }, + deployment_message, + rsmq.clone(), + false, + ) + .await + { + tracing::error!(%e, "error handling deployment metadata"); + } + + // tx = PushIsolationLevel::Transaction(new_tx); + // tx = handle_deployment_metadata( + // tx, + // &authed, + // &db, + // &w_id, + // DeployedObject::App { path: app.path.clone(), version: v_id }, + // app.deployment_message, + // ) + // .await?; + + // match tx { + // PushIsolationLevel::Transaction(tx) => tx.commit().await?, + // _ => { + // return Err(Error::InternalErr( + // "Expected a transaction here".to_string(), + // )); + // } + // } + + Ok(()) + } else { + Ok(()) + } +} + +async fn capture_dependency_job( + job_id: &Uuid, + job_language: &ScriptLang, + job_raw_code: &str, + mem_peak: &mut i32, + canceled_by: &mut Option, + job_dir: &str, + db: &sqlx::Pool, + worker_name: &str, + w_id: &str, + worker_dir: &str, + base_internal_url: &str, + token: &str, + script_path: &str, + raw_deps: bool, +) -> error::Result { + match job_language { + ScriptLang::Python3 => { + let reqs = if raw_deps { + job_raw_code.to_string() + } else { + let mut already_visited = vec![]; + + windmill_parser_py_imports::parse_python_imports( + job_raw_code, + &w_id, + script_path, + &db, + &mut already_visited, + ) + .await? + .join("\n") + }; + create_dependencies_dir(job_dir).await; + let req: std::result::Result = pip_compile( + job_id, + &reqs, + mem_peak, + canceled_by, + job_dir, + db, + worker_name, + w_id, + ) + .await; + // install the dependencies to pre-fill the cache + if let Ok(req) = req.as_ref() { + let r = handle_python_reqs( + req.split("\n").filter(|x| !x.starts_with("--")).collect(), + job_id, + w_id, + mem_peak, + canceled_by, + db, + worker_name, + job_dir, + worker_dir, + ) + .await; + + if let Err(e) = r { + tracing::error!( + "Failed to install python dependencies to prefill the cache: {:?} \n", + e + ); + } + } + req + } + ScriptLang::Go => { + if raw_deps { + return Err(Error::ExecutionErr( + "Raw dependencies not supported for go".to_string(), + )); + } + install_go_dependencies( + job_id, + job_raw_code, + mem_peak, + canceled_by, + job_dir, + db, + false, + false, + false, + worker_name, + w_id, + ) + .await + } + ScriptLang::Deno => { + if raw_deps { + return Err(Error::ExecutionErr( + "Raw dependencies not supported for deno".to_string(), + )); + } + generate_deno_lock( + job_id, + job_raw_code, + mem_peak, + canceled_by, + job_dir, + db, + w_id, + worker_name, + base_internal_url, + ) + .await + } + ScriptLang::Bun => { + let trusted_deps = if !raw_deps { + let _ = write_file(job_dir, "main.ts", job_raw_code).await?; + //TODO: remove once bun provides sane default fot it + get_trusted_deps(job_raw_code) + } else { + vec![] + }; + let req = gen_lockfile( + mem_peak, + canceled_by, + job_id, + w_id, + db, + token, + script_path, + job_dir, + base_internal_url, + worker_name, + true, + trusted_deps, + if raw_deps { + Some(job_raw_code.to_string()) + } else { + None + }, + false, + ) + .await?; + Ok(req.unwrap_or_else(String::new)) + } + ScriptLang::Php => { + let reqs = if raw_deps { + if job_raw_code.is_empty() { + return Ok("".to_string()); + } + job_raw_code.to_string() + } else { + match parse_php_imports(job_raw_code)? { + Some(reqs) => reqs, + None => { + return Ok("".to_string()); + } + } + }; + + composer_install( + mem_peak, + canceled_by, + job_id, + w_id, + db, + job_dir, + worker_name, + reqs, + None, + ) + .await + } + ScriptLang::Postgresql => Ok("".to_owned()), + ScriptLang::Mysql => Ok("".to_owned()), + ScriptLang::Bigquery => Ok("".to_owned()), + ScriptLang::Snowflake => Ok("".to_owned()), + ScriptLang::Mssql => Ok("".to_owned()), + ScriptLang::Graphql => Ok("".to_owned()), + ScriptLang::Bash => Ok("".to_owned()), + ScriptLang::Powershell => Ok("".to_owned()), + ScriptLang::Nativets => Ok("".to_owned()), + } +} diff --git a/frontend/src/lib/ata/index.ts b/frontend/src/lib/ata/index.ts index 308b37033d..4be56edb0f 100644 --- a/frontend/src/lib/ata/index.ts +++ b/frontend/src/lib/ata/index.ts @@ -117,6 +117,7 @@ export const setupTypeAcquisition = (config: ATABootstrapConfig) => { ? f.raw : '/' + config.scriptPath + (f.raw.startsWith('../') ? '/../' : '/.') + f.raw let url = config.root + path + // console.log('FOO', config.scriptPath, path, f.raw) console.log('fetching local file', url, f.raw) const res = await fetch(url) if (res.ok) { diff --git a/frontend/src/lib/components/Editor.svelte b/frontend/src/lib/components/Editor.svelte index e8652e0a21..fbf9f8d9a5 100644 --- a/frontend/src/lib/components/Editor.svelte +++ b/frontend/src/lib/components/Editor.svelte @@ -27,6 +27,8 @@ import 'monaco-editor/esm/vs/basic-languages/css/css.contribution' import libStdContent from '$lib/es6.d.ts.txt?raw' + import domContent from '$lib/dom.d.ts.txt?raw' + import denoFetchContent from '$lib/deno_fetch.d.ts.txt?raw' import processStdContent from '$lib/process.d.ts.txt?raw' import windmillFetchContent from '$lib/windmill_fetch.d.ts.txt?raw' @@ -1145,7 +1147,8 @@ const stdLib = { content: libStdContent, filePath: 'es6.d.ts' } if (scriptLang == 'bun') { const processLib = { content: processStdContent, filePath: 'process.d.ts' } - languages.typescript.typescriptDefaults.setExtraLibs([stdLib, processLib]) + const domLib = { content: domContent, filePath: 'dom.d.ts' } + languages.typescript.typescriptDefaults.setExtraLibs([stdLib, domLib, processLib]) } else { const denoFetch = { content: denoFetchContent, filePath: 'deno_fetch.d.ts' } languages.typescript.typescriptDefaults.setExtraLibs([stdLib, denoFetch]) diff --git a/frontend/src/lib/components/flows/content/FlowModuleComponent.svelte b/frontend/src/lib/components/flows/content/FlowModuleComponent.svelte index 343a613efb..f877ad5903 100644 --- a/frontend/src/lib/components/flows/content/FlowModuleComponent.svelte +++ b/frontend/src/lib/components/flows/content/FlowModuleComponent.svelte @@ -262,7 +262,7 @@ {#key flowModule.id}