diff --git a/backend/.sqlx/query-7fa8b53615a7cb678f9d5caedbc44cfa928469ce3c1a53ec273ecc997f6e61f8.json b/backend/.sqlx/query-7fa8b53615a7cb678f9d5caedbc44cfa928469ce3c1a53ec273ecc997f6e61f8.json new file mode 100644 index 0000000000..e171d44af5 --- /dev/null +++ b/backend/.sqlx/query-7fa8b53615a7cb678f9d5caedbc44cfa928469ce3c1a53ec273ecc997f6e61f8.json @@ -0,0 +1,23 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT hash FROM script WHERE path = $1 AND workspace_id = $2 AND archived = false ORDER BY created_at DESC LIMIT 1", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "hash", + "type_info": "Int8" + } + ], + "parameters": { + "Left": [ + "Text", + "Text" + ] + }, + "nullable": [ + false + ] + }, + "hash": "7fa8b53615a7cb678f9d5caedbc44cfa928469ce3c1a53ec273ecc997f6e61f8" +} diff --git a/backend/.sqlx/query-8fef781ecef8bb7e4f5172adbab1d1eefa5cf05ff6521ee605af7ac2b90fa60b.json b/backend/.sqlx/query-8fef781ecef8bb7e4f5172adbab1d1eefa5cf05ff6521ee605af7ac2b90fa60b.json deleted file mode 100644 index f65ea6d2d4..0000000000 --- a/backend/.sqlx/query-8fef781ecef8bb7e4f5172adbab1d1eefa5cf05ff6521ee605af7ac2b90fa60b.json +++ /dev/null @@ -1,23 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "SELECT created_at FROM script WHERE workspace_id = $1 AND hash = $2", - "describe": { - "columns": [ - { - "ordinal": 0, - "name": "created_at", - "type_info": "Timestamptz" - } - ], - "parameters": { - "Left": [ - "Text", - "Int8" - ] - }, - "nullable": [ - false - ] - }, - "hash": "8fef781ecef8bb7e4f5172adbab1d1eefa5cf05ff6521ee605af7ac2b90fa60b" -} diff --git a/backend/src/main.rs b/backend/src/main.rs index 94b8e2f4f6..0d7bef4d70 100644 --- a/backend/src/main.rs +++ b/backend/src/main.rs @@ -1708,6 +1708,10 @@ async fn process_notify_event( match *source_type { "script" => { windmill_common::DEPLOYED_SCRIPT_HASH_CACHE.remove(&key); + // Bundle-cache key resolution for imported scripts; evicted + // together with the content-side caches below so key and + // inlined content flip to the new version in the same window. + windmill_common::IMPORTED_SCRIPT_HASH_CACHE.remove(&key); // Evict the relative-import latest-hash cache so a redeployed // imported script flips the content cache to its new version // across all replicas within a poll interval (see #6769). Keyed diff --git a/backend/tests/bun_jobs.rs b/backend/tests/bun_jobs.rs index d51e51f58b..fc766253bc 100644 --- a/backend/tests/bun_jobs.rs +++ b/backend/tests/bun_jobs.rs @@ -1,3 +1,4 @@ +use futures::StreamExt; use sqlx::postgres::Postgres; use sqlx::Pool; use uuid::Uuid; @@ -818,6 +819,126 @@ export function main() { Ok(()) } +// ============================================================================ +// Bundle cache invalidation on transitive relative-import change +// ============================================================================ + +async fn insert_deployed_bun_script(db: &Pool, path: &str, hash: i64, content: &str) { + // What gen_bun_lockfile stores for a script with no npm dependencies; a + // bare '' lock fails split_lockfile when the script is run directly. + const EMPTY_BUN_LOCK: &str = "{\n \"dependencies\": {}\n}\n//bun.lock\n"; + // Runtime query to avoid touching the sqlx offline cache. + sqlx::query( + "INSERT INTO script (workspace_id, created_by, content, schema, summary, description, path, hash, language, lock) + VALUES ('test-workspace', 'test-user', $1, '{\"$schema\":\"https://json-schema.org/draft/2020-12/schema\",\"properties\":{},\"required\":[],\"type\":\"object\"}', '', '', $2, $3, 'bun', $4)", + ) + .bind(content) + .bind(path) + .bind(hash) + .bind(EMPTY_BUN_LOCK) + .execute(db) + .await + .unwrap(); +} + +fn run_main_script_job(hash: i64) -> RunJob { + RunJob::from(JobPayload::ScriptHash { + path: "f/stale_bundle/main_script".to_string(), + hash: windmill_common::scripts::ScriptHash(hash), + cache_ttl: None, + cache_ignore_s3_path: None, + dedicated_worker: None, + language: ScriptLang::Bun, + priority: None, + apply_preprocessor: false, + concurrency_settings: windmill_common::runnable_settings::ConcurrencySettings::default(), + debouncing_settings: windmill_common::runnable_settings::DebouncingSettings::default(), + labels: None, + }) +} + +/// Editing a script that a runnable imports only TRANSITIVELY (main -> mid -> +/// leaf) must invalidate the runnable's cached bundle: the leaf's code is +/// inlined in the bundle, so the cache key has to cover the whole closure, not +/// just direct imports. +#[sqlx::test(fixtures("base"))] +async fn test_bun_transitive_import_change_rebundles(db: Pool) -> anyhow::Result<()> { + initialize_tracing().await; + let server = ApiServer::start(db.clone()).await?; + let port = server.addr.port(); + + // Hashes/paths unique across this test binary: the script/hash caches are + // process-global while parallel tests each run in their own DB. + const LEAF_V1: i64 = 41230001; + const MID: i64 = 41230002; + const MAIN: i64 = 41230003; + const LEAF_V2: i64 = 41230004; + + insert_deployed_bun_script( + &db, + "f/stale_bundle/leaf", + LEAF_V1, + r#"export function leafValue() { return "V1_FROM_LEAF"; }"#, + ) + .await; + insert_deployed_bun_script( + &db, + "f/stale_bundle/mid", + MID, + r#"import { leafValue } from "./leaf"; +export function midValue() { return `M(${leafValue()})`; }"#, + ) + .await; + insert_deployed_bun_script( + &db, + "f/stale_bundle/main_script", + MAIN, + r#"import { midValue } from "./mid"; +export function main() { return midValue(); }"#, + ) + .await; + + let mut completed = listen_for_completed_jobs(&db).await; + let db2 = db.clone(); + in_test_worker( + &db, + async move { + let job = run_main_script_job(MAIN).push(&db2).await; + completed.next().await; + let result = completed_job(job, &db2).await.json_result().unwrap(); + assert_eq!(result, serde_json::json!("M(V1_FROM_LEAF)")); + + // Deploy a new version of ONLY the leaf; main_script and mid keep + // their hash, content, and lock byte-identical. + insert_deployed_bun_script( + &db2, + "f/stale_bundle/leaf", + LEAF_V2, + r#"export function leafValue() { return "V2_FROM_LEAF"; }"#, + ) + .await; + + // Tests don't run the notify_event poll loop, so replay what its + // `notify_runnable_version_change` handler (main.rs) does on deploy: + // evict the leaf's latest-hash cache entries. + windmill_common::IMPORTED_SCRIPT_HASH_CACHE.remove(&( + "test-workspace".to_string(), + "f/stale_bundle/leaf".to_string(), + )); + windmill_api_scripts::scripts::RAW_SCRIPT_LATEST_HASH_CACHE + .remove(&format!("test-workspace:f/stale_bundle/leaf")); + + let job = run_main_script_job(MAIN).push(&db2).await; + completed.next().await; + let result = completed_job(job, &db2).await.json_result().unwrap(); + assert_eq!(result, serde_json::json!("M(V2_FROM_LEAF)")); + }, + port, + ) + .await; + Ok(()) +} + #[sqlx::test(fixtures("base", "bun_edge_cases"))] async fn test_bun_shared_imports_both_styles(db: Pool) -> anyhow::Result<()> { initialize_tracing().await; diff --git a/backend/windmill-common/src/lib.rs b/backend/windmill-common/src/lib.rs index aaead77925..baa3e4b500 100644 --- a/backend/windmill-common/src/lib.rs +++ b/backend/windmill-common/src/lib.rs @@ -268,6 +268,10 @@ lazy_static::lazy_static! { pub static ref INSTANCE_NAME: String = rd_string(5); pub static ref DEPLOYED_SCRIPT_HASH_CACHE: Cache<(String, String), ExpiringLatestVersionId> = Cache::new(1000); + // Latest non-archived version per (workspace, path) for bundle cache keying — + // looser predicate than DEPLOYED_SCRIPT_HASH_CACHE (no lock requirement), so + // the two must not share entries. See get_latest_script_hash_for_import_cached. + pub static ref IMPORTED_SCRIPT_HASH_CACHE: Cache<(String, String), ExpiringLatestVersionId> = Cache::new(1000); pub static ref FLOW_VERSION_CACHE: Cache<(String, String), ExpiringLatestVersionId> = Cache::new(1000); pub static ref DYNAMIC_INPUT_CACHE: Cache> = Cache::new(1000); pub static ref DEPLOYED_SCRIPT_INFO_CACHE: Cache<(String, i64), ScriptHashInfo> = Cache::new(1000); @@ -1602,6 +1606,46 @@ pub async fn get_latest_script_hash<'e, E: sqlx::PgExecutor<'e>>( return Ok(hash); } +/// Latest non-archived hash for an imported `path`, for bundle cache keying. +/// MUST select the same row as the bundler's content endpoint +/// (`raw_script_by_path_internal`: `archived = false ORDER BY created_at DESC`, +/// no lock predicate) — a stricter filter here would let the key point at an +/// older version than the content that gets inlined. Cached with the same +/// freshness contract as that endpoint's `RAW_SCRIPT_LATEST_HASH_CACHE`: +/// evicted by `notify_runnable_version_change` events, 60s TTL fallback. +pub async fn get_latest_script_hash_for_import_cached( + db: &DB, + w_id: &str, + script_path: &str, +) -> error::Result> { + let use_cache = !DEPLOYED_SCRIPT_CACHE_DISABLED.load(std::sync::atomic::Ordering::Relaxed); + let cache_key = (w_id.to_string(), script_path.to_string()); + if use_cache { + if let Some(cached) = IMPORTED_SCRIPT_HASH_CACHE.get(&cache_key) { + if cached.expires_at > std::time::Instant::now() { + return Ok(Some(cached.id)); + } + } + } + let hash = sqlx::query_scalar!( + "SELECT hash FROM script WHERE path = $1 AND workspace_id = $2 AND archived = false ORDER BY created_at DESC LIMIT 1", + script_path, + w_id + ) + .fetch_optional(db) + .await?; + if let (true, Some(hash)) = (use_cache, hash) { + IMPORTED_SCRIPT_HASH_CACHE.insert( + cache_key, + ExpiringLatestVersionId { + id: hash, + expires_at: std::time::Instant::now() + LATEST_VERSION_ID_CACHE_TTL, + }, + ); + } + Ok(hash) +} + pub async fn get_script_info_for_hash<'e, E: sqlx::PgExecutor<'e>>( db_authed: Option>>, db: E, diff --git a/backend/windmill-worker/src/bun_executor.rs b/backend/windmill-worker/src/bun_executor.rs index 1e4f4aafd6..f6b093f766 100644 --- a/backend/windmill-worker/src/bun_executor.rs +++ b/backend/windmill-worker/src/bun_executor.rs @@ -1,6 +1,11 @@ #[cfg(feature = "deno_core")] use std::time::Instant; -use std::{collections::HashMap, fs, process::Stdio}; +use std::{ + collections::{HashMap, HashSet}, + fs, + process::Stdio, + sync::Arc, +}; use base64::Engine; use itertools::Itertools; @@ -27,9 +32,10 @@ use crate::{ NSJAIL_AVAILABLE, NSJAIL_PATH, PATH_ENV, PROXY_ENVS, TRACING_PROXY_CA_CERT_PATH, TZ_ENV, }; use windmill_common::{ + cache, client::AuthedClient, jobs::JobKind, - scripts::{id_to_codebase_info, CodebaseInfo, ScriptLang}, + scripts::{id_to_codebase_info, CodebaseInfo, ScriptHash, ScriptLang}, utils::WarnAfterExt, workspace_dependencies::WorkspaceDependenciesPrefetched, }; @@ -44,7 +50,6 @@ use tokio::io::AsyncReadExt; use windmill_common::{ error::{self, Result}, - get_latest_hash_for_path, worker::{write_file, Connection, DISABLE_BUNDLING}, DB, }; @@ -1242,16 +1247,101 @@ pub fn ensure_bundle_output_exists(bundle_path: &str) -> Result<()> { pub const BUN_BUNDLE_OBJECT_STORE_PREFIX: &str = "bun_bundle/"; -async fn get_script_import_updated_at(db: &DB, w_id: &str, script_path: &str) -> Result { - let script_hash = get_latest_hash_for_path(db, w_id, script_path, false).await?; - let last_updated_at = sqlx::query_scalar!( - "SELECT created_at FROM script WHERE workspace_id = $1 AND hash = $2", - w_id, - script_hash.0 .0 +// A script version's relative-import list never changes (content is immutable +// per hash), so parses are memoized without any invalidation. +lazy_static::lazy_static! { + static ref RELATIVE_IMPORTS_PER_HASH: quick_cache::sync::Cache>> = + quick_cache::sync::Cache::new(1000); +} + +const MAX_TRANSITIVE_IMPORT_PATHS: usize = 256; + +/// `(path, latest hash)` for the whole transitive closure of relative imports +/// of `inner_content` — the set of scripts whose code gets inlined into the +/// bundle, so all of them must key the bundle cache. Resolution goes through +/// `IMPORTED_SCRIPT_HASH_CACHE` (notify-evicted, 60s TTL fallback, same +/// version-selection predicate as the loader's content endpoint) and the +/// per-hash script/parse caches, so steady state costs no DB queries. +async fn collect_transitive_import_versions( + db: &DB, + w_id: &str, + script_path: &str, + inner_content: &str, +) -> Vec<(String, i64)> { + let conn = Connection::from(db.clone()); + let mut queue = crate::worker_lockfiles::extract_relative_imports( + inner_content, + script_path, + &Some(ScriptLang::Bun), ) - .fetch_one(db) - .await?; - Ok(last_updated_at.to_string()) + .unwrap_or_default(); + let mut visited: HashSet = HashSet::new(); + let mut versions: Vec<(String, i64)> = vec![]; + while let Some(path) = queue.pop() { + if !visited.insert(path.clone()) { + continue; + } + if visited.len() > MAX_TRANSITIVE_IMPORT_PATHS { + tracing::warn!( + "transitive relative-import closure of {script_path} exceeds \ + {MAX_TRANSITIVE_IMPORT_PATHS} scripts; bundle cache key covers only the first \ + {MAX_TRANSITIVE_IMPORT_PATHS}" + ); + break; + } + let hash = match windmill_common::get_latest_script_hash_for_import_cached(db, w_id, &path) + .await + { + Ok(Some(hash)) => hash, + // Not a deployed script at this path (deleted, or not a script): + // excluded from the key, matching what the bundler can inline. + Ok(None) => continue, + Err(e) => { + tracing::warn!( + "could not resolve import {path} while computing bundle cache key for \ + {script_path}: {e:#}" + ); + continue; + } + }; + versions.push((path.clone(), hash)); + let imports = match RELATIVE_IMPORTS_PER_HASH.get(&hash) { + Some(imports) => imports, + None => match cache::script::fetch(&conn, ScriptHash(hash)).await { + Ok((data, meta)) => { + let imports = Arc::new(match meta.language { + Some(ScriptLang::Bun) + | Some(ScriptLang::Bunnative) + | Some(ScriptLang::Deno) => { + crate::worker_lockfiles::extract_relative_imports( + &data.code, + &path, + &meta.language, + ) + .unwrap_or_default() + } + _ => vec![], + }); + RELATIVE_IMPORTS_PER_HASH.insert(hash, imports.clone()); + imports + } + // A fetch error is transient, not a property of the (immutable) + // content — memoizing it would drop this subtree from the key + // until worker restart. Skip caching and retry next run. + Err(e) => { + tracing::warn!( + "could not fetch import {path} (hash {hash}) while computing bundle \ + cache key for {script_path}: {e:#}" + ); + Arc::new(vec![]) + } + }, + }; + queue.extend(imports.iter().cloned()); + } + // deterministic key regardless of traversal order + versions.sort(); + versions } pub async fn compute_bundle_local_and_remote_path( @@ -1265,16 +1355,13 @@ pub async fn compute_bundle_local_and_remote_path( let mut input_src = format!("{inner_content}{lock}",); if let Some(db) = db { - let relative_imports = crate::worker_lockfiles::extract_relative_imports( - &inner_content, - script_path, - &Some(ScriptLang::Bun), - ); - for path in relative_imports.unwrap_or_default() { - if let Ok(updated_at) = get_script_import_updated_at(&db, w_id, &path).await { - input_src.push_str(&path); - input_src.push_str(&updated_at.to_string()); - } + // The bundle inlines the whole transitive relative-import closure, so a + // new deployed version of ANY script in it must change the key. + for (path, hash) in + collect_transitive_import_versions(db, w_id, script_path, inner_content).await + { + input_src.push_str(&path); + input_src.push_str(&hash.to_string()); } };