feat: accelerate bun through caches (#3909)

* all

* all

* all

* all

* all

* simplify extract

backend/src/main.rs

@@ -53,9 +53,9 @@ use windmill_common::METRICS_ADDR;
 use windmill_common::global_settings::OBJECT_STORE_CACHE_CONFIG_SETTING;
 
 use windmill_worker::{
-    BUN_CACHE_DIR, DENO_CACHE_DIR, DENO_CACHE_DIR_DEPS, DENO_CACHE_DIR_NPM, GO_BIN_CACHE_DIR,
-    GO_CACHE_DIR, HUB_CACHE_DIR, LOCK_CACHE_DIR, PIP_CACHE_DIR, POWERSHELL_CACHE_DIR,
-    TAR_PIP_CACHE_DIR, TMP_LOGS_DIR,
+    BUN_CACHE_DIR, BUN_TAR_CACHE_DIR, DENO_CACHE_DIR, DENO_CACHE_DIR_DEPS, DENO_CACHE_DIR_NPM,
+    GO_BIN_CACHE_DIR, GO_CACHE_DIR, HUB_CACHE_DIR, LOCK_CACHE_DIR, PIP_CACHE_DIR,
+    POWERSHELL_CACHE_DIR, TAR_PIP_CACHE_DIR, TMP_LOGS_DIR,
 };
 
 use crate::monitor::{
@@ -709,6 +709,7 @@ pub async fn run_workers<R: rsmq_async::RsmqConnection + Send + Sync + Clone + '
         DENO_CACHE_DIR_DEPS,
         DENO_CACHE_DIR_NPM,
         BUN_CACHE_DIR,
+        BUN_TAR_CACHE_DIR,
         GO_CACHE_DIR,
         GO_BIN_CACHE_DIR,
         HUB_CACHE_DIR,

backend/windmill-api/src/jobs.rs

@@ -28,8 +28,8 @@ use windmill_common::worker::TMP_DIR;
 use windmill_common::scripts::PREVIEW_IS_CODEBASE_HASH;
 use windmill_common::variables::get_workspace_key;
 
-use crate::concurrency_groups::join_concurrency_key;
 use crate::add_webhook_allowed_origin;
+use crate::concurrency_groups::join_concurrency_key;
 use crate::db::ApiAuthed;
 
 use crate::{
@@ -1327,9 +1327,7 @@ async fn list_filtered_uuids(
 ) -> error::JsonResult<Vec<Uuid>> {
     require_admin(authed.is_admin, &authed.username)?;
 
-    let mut sqlb = SqlBuilder::select_from("queue")
-        .fields(&["id"])
-        .clone();
+    let mut sqlb = SqlBuilder::select_from("queue").fields(&["id"]).clone();
 
     sqlb = join_concurrency_key(lq.concurrency_key.as_ref(), sqlb);
 
@@ -3935,13 +3933,16 @@ async fn add_batch_jobs(
         )
         .fetch_all(&db)
         .await?;
-    sqlx::query!(
-        "INSERT INTO concurrency_key (job_id, key) SELECT id, $1 FROM unnest($2::uuid[]) as id",
-        custom_concurrency_key,
-        &uuids
-    )
-    .execute(&db)
-    .await?;
+
+    if let Some(custom_concurrency_key) = custom_concurrency_key {
+        sqlx::query!(
+            "INSERT INTO concurrency_key (job_id, key) SELECT id, $1 FROM unnest($2::uuid[]) as id",
+            custom_concurrency_key,
+            &uuids
+        )
+        .execute(&db)
+        .await?;
+    }
 
     Ok(Json(uuids))
 }

backend/windmill-worker/Cargo.toml

@@ -11,10 +11,10 @@ path = "src/lib.rs"
 [features]
 default = []
 prometheus = ["dep:prometheus", "windmill-common/prometheus"]
-enterprise = ["windmill-queue/enterprise", "windmill-git-sync/enterprise", "windmill-common/enterprise", "dep:gcp_auth", "dep:jsonwebtoken", "dep:pem", "dep:sha2", "dep:tiberius", "dep:tokio-util", "dep:openidconnect"]
+enterprise = ["windmill-queue/enterprise", "windmill-git-sync/enterprise", "windmill-common/enterprise", "dep:gcp_auth", "dep:jsonwebtoken", "dep:pem", "dep:tiberius", "dep:tokio-util", "dep:openidconnect"]
 benchmark = ["windmill-queue/benchmark"]
 flamegraph = []
-parquet = ["windmill-common/parquet", "dep:object_store", "dep:tar"]
+parquet = ["windmill-common/parquet", "dep:object_store"]
 flow_testing = []
 cloud = []
 
@@ -71,7 +71,7 @@ base64.workspace = true
 gcp_auth = { workspace = true, optional = true }
 rust_decimal.workspace = true
 jsonwebtoken = { workspace = true, optional = true }
-sha2 = { workspace = true, optional = true }
+sha2.workspace = true
 pem = { workspace = true, optional = true }
 urlencoding.workspace = true
 nix.workspace = true
@@ -81,7 +81,7 @@ hex.workspace = true
 tiberius = { workspace = true, optional = true }
 tokio-util = { workspace = true, optional = true }
 openidconnect = { workspace = true, optional = true}
-tar = { workspace = true, optional = true}
+tar.workspace = true
 object_store = { workspace = true, optional = true}
 convert_case.workspace = true
 

backend/windmill-worker/src/bun_executor.rs

@@ -2,8 +2,8 @@ use std::{collections::HashMap, process::Stdio};
 
 use base64::Engine;
 use itertools::Itertools;
-use regex::Regex;
 use serde_json::value::RawValue;
+use sha2::Digest;
 use uuid::Uuid;
 use windmill_parser_ts::remove_pinned_imports;
 use windmill_queue::{append_logs, CanceledBy};
@@ -16,9 +16,9 @@ use crate::{
         create_args_and_out_file, get_main_override, get_reserved_variables, handle_child,
         parse_npm_config, read_result, start_child_process, write_file, write_file_binary,
     },
-    AuthedClientBackgroundTask, BUNFIG_INSTALL_SCOPES, BUN_CACHE_DIR, BUN_PATH, DISABLE_NSJAIL,
-    DISABLE_NUSER, HOME_ENV, NODE_PATH, NPM_CONFIG_REGISTRY, NPM_PATH, NSJAIL_PATH, PATH_ENV,
-    TZ_ENV,
+    AuthedClientBackgroundTask, BUNFIG_INSTALL_SCOPES, BUN_CACHE_DIR, BUN_PATH, BUN_TAR_CACHE_DIR,
+    DISABLE_NSJAIL, DISABLE_NUSER, HOME_ENV, NODE_PATH, NPM_CONFIG_REGISTRY, NPM_PATH, NSJAIL_PATH,
+    PATH_ENV, TZ_ENV,
 };
 
 use tokio::{fs::File, process::Command};
@@ -32,7 +32,7 @@ use tokio::sync::mpsc::Receiver;
 use windmill_common::variables;
 
 use windmill_common::{
-    error::{self, to_anyhow, Result},
+    error::{self, Result},
     jobs::QueuedJob,
 };
 
@@ -50,11 +50,6 @@ const NSJAIL_CONFIG_RUN_BUN_CONTENT: &str = include_str!("../nsjail/run.bun.conf
 pub const BUN_LOCKB_SPLIT: &str = "\n//bun.lockb\n";
 pub const EMPTY_FILE: &str = "<empty>";
 
-lazy_static::lazy_static! {
-    pub static ref TRUSTED_DEP: Regex = Regex::new(r"//\s?trustedDependencies:(.*)\n").unwrap();
-
-}
-
 pub async fn gen_lockfile(
     mem_peak: &mut i32,
     canceled_by: &mut Option<CanceledBy>,
@@ -67,7 +62,6 @@ pub async fn gen_lockfile(
     base_internal_url: &str,
     worker_name: &str,
     export_pkg: bool,
-    trusted_deps: Vec<String>,
     raw_deps: Option<String>,
     npm_mode: bool,
 ) -> Result<Option<String>> {
@@ -123,33 +117,6 @@ pub async fn gen_lockfile(
             false,
         )
         .await?;
-
-        if trusted_deps.len() > 0 {
-            let logs1 = format!(
-                "\ndetected trustedDependencies: {}\n",
-                trusted_deps.join(", ")
-            );
-            append_logs(&job_id, w_id, logs1, db).await;
-
-            let mut content = "".to_string();
-            {
-                let mut file = File::open(format!("{job_dir}/package.json")).await?;
-                file.read_to_string(&mut content).await?;
-            }
-            let mut value =
-                serde_json::from_str::<serde_json::Map<String, serde_json::Value>>(&content)
-                    .map_err(to_anyhow)?;
-            value.insert(
-                "trustedDependencies".to_string(),
-                serde_json::json!(trusted_deps),
-            );
-            write_file(
-                job_dir,
-                "package.json",
-                &serde_json::to_string(&value).map_err(to_anyhow)?,
-            )
-            .await?;
-        }
     }
 
     install_lockfile(
@@ -309,23 +276,6 @@ pub async fn install_lockfile(
     Ok(())
 }
 
-pub fn get_trusted_deps(code: &str) -> Vec<String> {
-    // postinstall not allowed with nsjail
-    if !*DISABLE_NSJAIL {
-        return vec![];
-    }
-    TRUSTED_DEP
-        .captures(code)
-        .map(|x| {
-            x[1].to_string()
-                .trim()
-                .split(',')
-                .map(|y| y.trim().to_string())
-                .collect_vec()
-        })
-        .unwrap_or_default()
-}
-
 struct Annotations {
     npm_mode: bool,
     nodejs_mode: bool,
@@ -491,6 +441,18 @@ pub async fn pull_codebase(_w_id: &str, _id: &str, _job_dir: &str) -> Result<()>
     ));
 }
 
+fn untar_file(file_path: &str, output_dir: &str) -> anyhow::Result<()> {
+    // Open the tar file
+    let file = std::fs::File::open(file_path)?;
+    let file = std::io::BufReader::new(file);
+
+    // For a plain tar file, use it directly
+    let mut archive = tar::Archive::new(file);
+    archive.unpack(output_dir)?;
+
+    Ok(())
+}
+
 #[tracing::instrument(level = "trace", skip_all)]
 pub async fn handle_bun_job(
     requirements_o: Option<String>,
@@ -511,7 +473,7 @@ pub async fn handle_bun_job(
         let _ = write_file(job_dir, "main.ts", inner_content).await?;
     } else {
         let _ = write_file(job_dir, "package.json", r#"{ "type": "module" }"#).await?;
-    }
+    };
 
     let common_bun_proc_envs: HashMap<String, String> =
         get_common_bun_proc_envs(&base_internal_url).await;
@@ -530,6 +492,7 @@ pub async fn handle_bun_job(
         ));
     }
 
+    let mut gbuntar_name = None;
     if let Some(codebase) = codebase.as_ref() {
         pull_codebase(&job.workspace_id, codebase, job_dir).await?;
     } else if let Some(reqs) = requirements_o {
@@ -542,38 +505,58 @@ pub async fn handle_bun_job(
         let _ = write_file(job_dir, "package.json", &splitted[0]).await?;
         let lockb = splitted[1];
         if lockb != EMPTY_FILE {
-            let has_trusted_deps = &splitted[0].contains("trustedDependencies");
-
-            if !has_trusted_deps {
-                let _ = write_file_binary(
-                    job_dir,
-                    "bun.lockb",
-                    &base64::engine::general_purpose::STANDARD
-                        .decode(&splitted[1])
-                        .map_err(|_| {
-                            error::Error::InternalErr("Could not decode bun.lockb".to_string())
-                        })?,
-                )
-                .await?;
-            }
-
-            install_lockfile(
-                mem_peak,
-                canceled_by,
-                &job.id,
-                &job.workspace_id,
-                db,
+            let _ = write_file_binary(
                 job_dir,
-                worker_name,
-                common_bun_proc_envs.clone(),
-                annotation.npm_mode,
+                "bun.lockb",
+                &base64::engine::general_purpose::STANDARD
+                    .decode(&splitted[1])
+                    .map_err(|_| {
+                        error::Error::InternalErr("Could not decode bun.lockb".to_string())
+                    })?,
             )
             .await?;
+
+            let mut sha_path = sha2::Sha256::new();
+            sha_path.update(lockb.as_bytes());
+
+            let buntar_name = base64::engine::general_purpose::STANDARD.encode(sha_path.finalize());
+            let buntar_path = format!("{BUN_TAR_CACHE_DIR}/{buntar_name}.tar");
+
+            let mut skip_install = false;
+            let mut create_buntar = false;
+            if tokio::fs::metadata(&buntar_path).await.is_ok() {
+                if let Err(e) = untar_file(&buntar_path, job_dir) {
+                    tracing::error!("Could not untar buntar: {e}");
+                } else {
+                    gbuntar_name = Some(buntar_name.clone());
+                    skip_install = true;
+                }
+            } else {
+                create_buntar = true;
+            }
+            if !skip_install {
+                install_lockfile(
+                    mem_peak,
+                    canceled_by,
+                    &job.id,
+                    &job.workspace_id,
+                    db,
+                    job_dir,
+                    worker_name,
+                    common_bun_proc_envs.clone(),
+                    annotation.npm_mode,
+                )
+                .await?;
+                if create_buntar {
+                    if let Err(e) = tar::Builder::new(std::fs::File::create(&buntar_path)?)
+                        .append_dir_all(".", job_dir)
+                    {
+                        tracing::error!("Could not create buntar: {e}");
+                    }
+                }
+            }
         }
     } else {
-        // TODO: remove once bun implement a reasonable set of trusted deps
-        let trusted_deps = get_trusted_deps(inner_content);
-
         // 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, logs1, db).await;
@@ -590,17 +573,17 @@ pub async fn handle_bun_job(
             base_internal_url,
             worker_name,
             false,
-            trusted_deps,
             None,
             annotation.npm_mode,
         )
         .await?;
+
         // }
     }
 
     let _ = write_file(job_dir, "main.ts", &remove_pinned_imports(inner_content)?).await?;
 
-    let init_logs = if codebase.is_some() {
+    let mut init_logs = if codebase.is_some() {
         "\n\n--- NODE SNAPSHOT EXECUTION ---\n".to_string()
     } else if annotation.nodejs_mode {
         "\n\n--- NODE CODE EXECUTION ---\n".to_string()
@@ -608,6 +591,13 @@ pub async fn handle_bun_job(
         "\n\n--- BUN CODE EXECUTION ---\n".to_string()
     };
 
+    if let Some(gbuntar_name) = gbuntar_name {
+        init_logs = format!(
+            "\nskipping install, using cached buntar based on lockfile hash: {gbuntar_name}{}",
+            init_logs
+        );
+    }
+
     append_logs(&job.id, &job.workspace_id, init_logs, db).await;
 
     let write_wrapper_f = async {
@@ -963,20 +953,16 @@ pub async fn start_worker(
         let _ = write_file(job_dir, "package.json", &splitted[0]).await?;
         let lockb = splitted[1];
         if lockb != EMPTY_FILE {
-            let has_trusted_deps = &splitted[0].contains("trustedDependencies");
-
-            if !has_trusted_deps {
-                let _ = write_file_binary(
-                    job_dir,
-                    "bun.lockb",
-                    &base64::engine::general_purpose::STANDARD
-                        .decode(&splitted[1])
-                        .map_err(|_| {
-                            error::Error::InternalErr("Could not decode bun.lockb".to_string())
-                        })?,
-                )
-                .await?;
-            }
+            let _ = write_file_binary(
+                job_dir,
+                "bun.lockb",
+                &base64::engine::general_purpose::STANDARD
+                    .decode(&splitted[1])
+                    .map_err(|_| {
+                        error::Error::InternalErr("Could not decode bun.lockb".to_string())
+                    })?,
+            )
+            .await?;
 
             install_lockfile(
                 &mut mem_peak,
@@ -993,7 +979,6 @@ pub async fn start_worker(
             tracing::info!("dedicated worker requirements installed: {reqs}");
         }
     } else if !*DISABLE_NSJAIL {
-        let trusted_deps = get_trusted_deps(inner_content);
         logs.push_str("\n\n--- BUN INSTALL ---\n");
         let _ = gen_lockfile(
             &mut mem_peak,
@@ -1007,7 +992,6 @@ pub async fn start_worker(
             base_internal_url,
             worker_name,
             false,
-            trusted_deps,
             None,
             annotation.npm_mode,
         )

backend/windmill-worker/src/worker.rs

@@ -204,6 +204,7 @@ pub const DENO_CACHE_DIR_NPM: &str = concatcp!(ROOT_CACHE_DIR, "deno/npm");
 
 pub const GO_CACHE_DIR: &str = concatcp!(ROOT_CACHE_DIR, "go");
 pub const BUN_CACHE_DIR: &str = concatcp!(ROOT_CACHE_DIR, "bun");
+pub const BUN_TAR_CACHE_DIR: &str = concatcp!(ROOT_CACHE_DIR, "buntar");
 
 pub const HUB_CACHE_DIR: &str = concatcp!(ROOT_CACHE_DIR, "hub");
 pub const GO_BIN_CACHE_DIR: &str = concatcp!(ROOT_CACHE_DIR, "gobin");
@@ -723,12 +724,14 @@ pub async fn run_worker<R: rsmq_async::RsmqConnection + Send + Sync + Clone + 's
         .await
         .expect("could not create initial worker dir");
 
-    let _ = write_file(
-        &worker_dir,
-        "download_deps.py.sh",
-        INCLUDE_DEPS_PY_SH_CONTENT,
-    )
-    .await;
+    if !*DISABLE_NSJAIL {
+        let _ = write_file(
+            &worker_dir,
+            "download_deps.py.sh",
+            INCLUDE_DEPS_PY_SH_CONTENT,
+        )
+        .await;
+    }
 
     let mut last_ping = Instant::now() - Duration::from_secs(NUM_SECS_PING + 1);
 

backend/windmill-worker/src/worker_lockfiles.rs

@@ -27,7 +27,7 @@ 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},
+    bun_executor::gen_lockfile,
     common::write_file,
     deno_executor::generate_deno_lock,
     go_executor::install_go_dependencies,
@@ -1200,13 +1200,9 @@ async fn capture_dependency_job(
             .await
         }
         ScriptLang::Bun => {
-            let trusted_deps = if !raw_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,
@@ -1219,7 +1215,6 @@ async fn capture_dependency_job(
                 base_internal_url,
                 worker_name,
                 true,
-                trusted_deps,
                 if raw_deps {
                     Some(job_raw_code.to_string())
                 } else {

benchmarks/benchmark_oneoff.ts

@@ -119,7 +119,11 @@ export async function main({
     return completedJobs - pastJobs;
   }
 
-  if (["deno", "python", "go", "bash", "dedicated", "bun", "nativets"].includes(kind)) {
+  if (
+    ["deno", "python", "go", "bash", "dedicated", "bun", "nativets"].includes(
+      kind
+    )
+  ) {
     await createBenchScript(kind, workspace);
   }
 
@@ -135,7 +139,9 @@ export async function main({
       kind: "noop",
     });
   } else if (
-    ["deno", "python", "go", "bash", "dedicated", "bun", "nativets"].includes(kind)
+    ["deno", "python", "go", "bash", "dedicated", "bun", "nativets"].includes(
+      kind
+    )
   ) {
     body = JSON.stringify({
       kind: "script",
@@ -148,22 +154,21 @@ export async function main({
       flow_value: payload.value,
     });
   } else if (kind.startsWith("flow:")) {
-    console.log("Detected custom flow ")
+    console.log("Detected custom flow ");
     body = JSON.stringify({
       kind: "flow",
-      path: kind.substr(5)
+      path: kind.substr(5),
     });
   } else if (kind.startsWith("script:")) {
-    console.log("Detected custom script")
+    console.log("Detected custom script");
     body = JSON.stringify({
       kind: "script",
-      path: kind.substr(7)
+      path: kind.substr(7),
     });
   } else {
     throw new Error("Unknown script pattern " + kind);
   }
 
-
   const response = await fetch(
     config.server +
       "/api/w/" +
@@ -179,7 +184,12 @@ export async function main({
     }
   );
   if (!response.ok) {
-    throw new Error("Failed to create jobs: " + response.statusText);
+    throw new Error(
+      "Failed to create jobs: " +
+        response.statusText +
+        " " +
+        (await response.text())
+    );
   }
   const uuids = await response.json();
   const end_create = Date.now();
@@ -247,7 +257,13 @@ export async function main({
   console.log("completed jobs", completedJobs);
   console.log("queue length:", await getQueueCount());
 
-  if (!noVerify && kind !== "noop" && kind !== 'nativets' && !kind.startsWith("flow:") && !kind.startsWith("script:")) {
+  if (
+    !noVerify &&
+    kind !== "noop" &&
+    kind !== "nativets" &&
+    !kind.startsWith("flow:") &&
+    !kind.startsWith("script:")
+  ) {
     await verifyOutputs(uuids, config.workspace_id);
   }