From 31c43255fdcc827c3fdf65e40238f8f3a83201cd Mon Sep 17 00:00:00 2001 From: Ruben Fiszel Date: Tue, 15 Sep 2026 13:02:44 +0200 Subject: [PATCH] fix(worker): bound cache transfers and import fetches in bun jobs (#11138) * fix(worker): bound object-store cache transfers and relative import fetches Co-Authored-By: Claude Fable 5.1 * fix(worker): bound the codebase download and label slow-step warnings Co-Authored-By: Claude Fable 5.1 * fix(worker): log a stalled cache transfer once and ignore a zero cache timeout Co-Authored-By: Claude Fable 5.1 --------- Co-authored-by: Claude Fable 5.1 --- backend/windmill-common/src/utils.rs | 66 ++++++-- backend/windmill-worker/loader.bun.js | 47 ++++-- backend/windmill-worker/loader.bun.windows.js | 50 ++++-- backend/windmill-worker/src/bun_executor.rs | 27 ++- backend/windmill-worker/src/global_cache.rs | 154 ++++++++++++++---- 5 files changed, 268 insertions(+), 76 deletions(-) diff --git a/backend/windmill-common/src/utils.rs b/backend/windmill-common/src/utils.rs index 62d47a34d7..4b93bb89ff 100644 --- a/backend/windmill-common/src/utils.rs +++ b/backend/windmill-common/src/utils.rs @@ -1125,19 +1125,35 @@ use tokio::time::{self, Duration, Sleep}; use pin_project_lite::pin_project; +/// What a [`WarnAfterFuture`] is timing, which decides how its warning reads. +pub enum WarnSubject { + /// A database query, with the SQL when the caller has it. + Query(Option), + /// Anything else (a child process, a cache transfer), named for the log line. + Step(String), +} + pub trait WarnAfterExt: Future + Sized { /// Warns if the future takes longer than the specified number of seconds to complete. #[track_caller] fn warn_after_seconds(self, seconds: u8) -> WarnAfterFuture { let caller = Location::caller(); - self.build_from_caller(seconds, caller, None) + self.build_from_caller(seconds, caller, WarnSubject::Query(None)) + } + + /// Same, for a step that is not a database query (a child process, a cache transfer): + /// the warning names `step` instead of reporting a slow query. + #[track_caller] + fn warn_after_seconds_for(self, seconds: u8, step: &str) -> WarnAfterFuture { + let caller = Location::caller(); + self.build_from_caller(seconds, caller, WarnSubject::Step(step.to_string())) } fn build_from_caller( self, seconds: u8, caller: &Location, - sql: Option, + subject: WarnSubject, ) -> WarnAfterFuture { let location = format!("{}:{}", caller.file(), caller.line()); WarnAfterFuture { @@ -1147,13 +1163,13 @@ pub trait WarnAfterExt: Future + Sized { start_time: std::time::Instant::now(), location, seconds, - sql, + subject, } } #[track_caller] fn warn_after_seconds_with_sql(self, seconds: u8, sql: String) -> WarnAfterFuture { let caller = Location::caller(); - self.build_from_caller(seconds, caller, Some(sql)) + self.build_from_caller(seconds, caller, WarnSubject::Query(Some(sql))) } } @@ -1171,7 +1187,7 @@ pin_project! { location: String, start_time: std::time::Instant, seconds: u8, - sql: Option, + subject: WarnSubject, } } @@ -1191,12 +1207,20 @@ impl Future for WarnAfterFuture { // Poll the timeout future to check if it has elapsed. if !*this.warned { if this.timeout.poll(cx).is_ready() { - tracing::warn!( - location = this.location, - "SLOW_QUERY: query {} to db taking longer than expected (> {} seconds)", - build_query_string(&this.location, this.sql.as_deref()), - this.seconds, - ); + match &*this.subject { + WarnSubject::Step(step) => tracing::warn!( + location = this.location, + "SLOW_STEP: {step} at {} taking longer than expected (> {} seconds)", + this.location, + this.seconds, + ), + WarnSubject::Query(sql) => tracing::warn!( + location = this.location, + "SLOW_QUERY: query {} to db taking longer than expected (> {} seconds)", + build_query_string(&this.location, sql.as_deref()), + this.seconds, + ), + } *this.warned = true; } } @@ -1206,12 +1230,20 @@ impl Future for WarnAfterFuture { Poll::Ready(output) => { if *this.warned { let elapsed = this.start_time.elapsed(); - tracing::warn!( - location = this.location, - "SLOW_QUERY: completed query {} with total duration: {:.2?}", - build_query_string(&this.location, this.sql.as_deref()), - elapsed - ); + match &*this.subject { + WarnSubject::Step(step) => tracing::warn!( + location = this.location, + "SLOW_STEP: {step} at {} completed with total duration: {:.2?}", + this.location, + elapsed + ), + WarnSubject::Query(sql) => tracing::warn!( + location = this.location, + "SLOW_QUERY: completed query {} with total duration: {:.2?}", + build_query_string(&this.location, sql.as_deref()), + elapsed + ), + } } Poll::Ready(output) } diff --git a/backend/windmill-worker/loader.bun.js b/backend/windmill-worker/loader.bun.js index 7a38369dd7..edfd3375ab 100644 --- a/backend/windmill-worker/loader.bun.js +++ b/backend/windmill-worker/loader.bun.js @@ -55,21 +55,46 @@ const p = { return replaceRelativeImports(code); }); - build.onLoad({ filter: /.*\.url$/ }, async (args) => { - const url = readFileSync(args.path, "utf8"); - const req = await fetch(url, { - method: "GET", - headers: { - Authorization: "Bearer " + token, - }, - }); + // A stalled fetch would otherwise hold the whole build for bun's own 5-minute + // default, with nothing naming the script it was waiting on. + const RELATIVE_IMPORT_FETCH_TIMEOUT_MS = 120000; + + function relativeImportFetchError(url, e) { + const reason = + e?.name === "TimeoutError" + ? `no response within ${RELATIVE_IMPORT_FETCH_TIMEOUT_MS / 1000}s` + : String(e?.message ?? e); + return new Error(`Failed to fetch relative import at ${url}: ${reason}`); + } + + async function fetchRelativeImport(url) { + let req; + try { + req = await fetch(url, { + method: "GET", + headers: { + Authorization: "Bearer " + token, + }, + signal: AbortSignal.timeout(RELATIVE_IMPORT_FETCH_TIMEOUT_MS), + }); + } catch (e) { + throw relativeImportFetchError(url, e); + } if (!req.ok) { throw new Error( - `Failed to find relative import at ${url}`, - req.statusText + `Failed to find relative import at ${url} (status ${req.status} ${req.statusText})` ); } - const contents = await req.text(); + try { + return await req.text(); + } catch (e) { + throw relativeImportFetchError(url, e); + } + } + + build.onLoad({ filter: /.*\.url$/ }, async (args) => { + const url = readFileSync(args.path, "utf8"); + const contents = await fetchRelativeImport(url); return { contents: replaceRelativeImports(contents).contents, loader: "tsx", diff --git a/backend/windmill-worker/loader.bun.windows.js b/backend/windmill-worker/loader.bun.windows.js index a877266662..e9bb36c21e 100644 --- a/backend/windmill-worker/loader.bun.windows.js +++ b/backend/windmill-worker/loader.bun.windows.js @@ -101,6 +101,43 @@ const p = { return replaceRelativeImports(code); }); + // A stalled fetch would otherwise hold the whole build for bun's own 5-minute + // default, with nothing naming the script it was waiting on. + const RELATIVE_IMPORT_FETCH_TIMEOUT_MS = 120000; + + function relativeImportFetchError(url, e) { + const reason = + e?.name === "TimeoutError" + ? `no response within ${RELATIVE_IMPORT_FETCH_TIMEOUT_MS / 1000}s` + : String(e?.message ?? e); + return new Error(`Failed to fetch relative import at ${url}: ${reason}`); + } + + async function fetchRelativeImport(url) { + let req; + try { + req = await fetch(url, { + method: "GET", + headers: { + Authorization: "Bearer " + token, + }, + signal: AbortSignal.timeout(RELATIVE_IMPORT_FETCH_TIMEOUT_MS), + }); + } catch (e) { + throw relativeImportFetchError(url, e); + } + if (!req.ok) { + throw new Error( + `Failed to find relative import at ${url} (status ${req.status} ${req.statusText})` + ); + } + try { + return await req.text(); + } catch (e) { + throw relativeImportFetchError(url, e); + } + } + // Load windmill scripts by fetching from the API build.onLoad({ filter: /.*/, namespace: "windmill-url" }, async (args) => { // Extract temp_script_hash if embedded in the path by resolveWindmillImport @@ -110,18 +147,7 @@ const p = { : undefined; const url = `${base_internal_url}/api/w/${w_id}/scripts/RAW_GET_ENDPOINT/p/${scriptPath}` + (hashParam ? `?temp_script_hash=${hashParam}` : ""); - const req = await fetch(url, { - method: "GET", - headers: { - Authorization: "Bearer " + token, - }, - }); - if (!req.ok) { - throw new Error( - `Failed to find relative import at ${url} (status ${req.status})` - ); - } - const contents = await req.text(); + const contents = await fetchRelativeImport(url); return { contents: replaceRelativeImports(contents).contents, loader: "tsx", diff --git a/backend/windmill-worker/src/bun_executor.rs b/backend/windmill-worker/src/bun_executor.rs index 4c19d0510d..7d45cab178 100644 --- a/backend/windmill-worker/src/bun_executor.rs +++ b/backend/windmill-worker/src/bun_executor.rs @@ -868,7 +868,7 @@ pub async fn install_bun_lockfile( if quiet { Some(&mut quiet_buf) } else { None }, None, ) - .warn_after_seconds(10) + .warn_after_seconds_for(10, "bun install") .await; if quiet && result.is_err() { // On failure, flush suppressed install output so the user can diagnose @@ -1131,9 +1131,12 @@ pub async fn generate_bun_bundle( None, None, ) + .warn_after_seconds_for(60, "bun build") .await?; } else { - let output = Box::into_pin(child_process.wait_with_output()).await?; + let output = Box::into_pin(child_process.wait_with_output()) + .warn_after_seconds_for(60, "bun build") + .await?; if !output.status.success() { let stdout = String::from_utf8_lossy(&output.stdout); let stderr = String::from_utf8_lossy(&output.stderr); @@ -1278,7 +1281,12 @@ async fn pull_codebase(w_id: &str, id: &str, job_dir: &str) -> Result { let em = format!("could not save {local_path} to bundle cache: {e:?}"); tracing::error!(em) diff --git a/backend/windmill-worker/src/global_cache.rs b/backend/windmill-worker/src/global_cache.rs index 3f6205f100..1c3bbf870f 100644 --- a/backend/windmill-worker/src/global_cache.rs +++ b/backend/windmill-worker/src/global_cache.rs @@ -9,6 +9,73 @@ use std::sync::Arc; pub const TARGET: &str = const_format::concatcp!(std::env::consts::OS, "_", std::env::consts::ARCH); +#[cfg(all(feature = "enterprise", feature = "parquet"))] +lazy_static::lazy_static! { + /// Object-store clients are built with their request timeout disabled so a large job + /// payload can stream for as long as it needs. A cache transfer must not inherit that: + /// a put or get that stalls after connecting would otherwise hold the job for its whole + /// duration limit, with nothing in the job log saying why. + pub(crate) static ref OBJECT_STORE_CACHE_IO_TIMEOUT: std::time::Duration = std::time::Duration::from_secs( + std::env::var("OBJECT_STORE_CACHE_IO_TIMEOUT_SECS") + .ok() + .and_then(|x| x.parse::().ok()) + .filter(|secs| *secs > 0) + .unwrap_or(30 * 60), + ); +} + +/// A cache transfer that did not complete: the store answered with an error, or +/// [`OBJECT_STORE_CACHE_IO_TIMEOUT`] ran out first. Nothing is logged on the way out: a failed +/// download is usually an ordinary miss, so each call site decides which outcome gets a line, +/// and logs it once. +#[cfg(all(feature = "enterprise", feature = "parquet"))] +pub(crate) enum CacheIoError { + TimedOut { what: String, path: String, secs: u64 }, + Failed(error::Error), +} + +#[cfg(all(feature = "enterprise", feature = "parquet"))] +impl std::fmt::Display for CacheIoError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + CacheIoError::TimedOut { what, path, secs } => write!( + f, + "{what} {path} in the object store cache timed out after {secs}s \ + (OBJECT_STORE_CACHE_IO_TIMEOUT_SECS)" + ), + CacheIoError::Failed(e) => write!(f, "{e:#}"), + } + } +} + +#[cfg(all(feature = "enterprise", feature = "parquet"))] +impl From for error::Error { + fn from(e: CacheIoError) -> Self { + match e { + CacheIoError::Failed(e) => e, + timed_out => error::Error::ExecutionErr(timed_out.to_string()), + } + } +} + +/// Runs one object-store cache transfer under [`OBJECT_STORE_CACHE_IO_TIMEOUT`]. +#[cfg(all(feature = "enterprise", feature = "parquet"))] +pub(crate) async fn bounded_cache_io( + what: &str, + path: &str, + io: impl std::future::Future>, +) -> Result { + let timeout = *OBJECT_STORE_CACHE_IO_TIMEOUT; + match tokio::time::timeout(timeout, io).await { + Ok(result) => result.map_err(CacheIoError::Failed), + Err(_) => Err(CacheIoError::TimedOut { + what: what.to_string(), + path: path.to_string(), + secs: timeout.as_secs(), + }), + } +} + #[cfg(all(feature = "enterprise", feature = "parquet"))] pub async fn build_tar_and_push( s3_client: Arc, @@ -56,17 +123,20 @@ pub async fn build_tar_and_push( // let s3_client = s3_settings.as_ref().ok_or_else(|| { // error::Error::ExecutionErr("Failed to read s3 cache settings".to_string()) // })?; - if let Err(e) = s3_client - .put( - &Path::from(format!( - "/tar/{}/{lang}/{folder_name}.tar", - if platform_agnostic { "" } else { TARGET } - )), - std::fs::read(&tar_path)?.into(), - ) - .await - { - tracing::info!("Failed to put tar to s3: {tar_path}. Error: {:?}", e); + let remote_path = format!( + "/tar/{}/{lang}/{folder_name}.tar", + if platform_agnostic { "" } else { TARGET } + ); + let tar_bytes = std::fs::read(&tar_path)?; + let put = bounded_cache_io("uploading", &remote_path, async { + s3_client + .put(&Path::from(remote_path.as_str()), tar_bytes.into()) + .await + .map_err(|e| error::Error::ExecutionErr(format!("{e:?}"))) + }) + .await; + if let Err(e) = put { + tracing::info!("Failed to put tar to s3: {tar_path}. Error: {e}"); return Err(error::Error::ExecutionErr(format!( "Failed to put tar to s3: {tar_path}" ))); @@ -109,7 +179,12 @@ pub async fn pull_from_tar( "tar/{}/{lang}/{folder_name}.tar", if platform_agnostic { "" } else { TARGET } ); - let bytes = attempt_fetch_bytes(client, &tar_path).await?; + let bytes = bounded_cache_io( + "downloading", + &tar_path, + attempt_fetch_bytes(client, &tar_path), + ) + .await?; extract_tar(bytes, &folder).map_err(|e| { tracing::error!("Failed to extract piptar {folder_name}. Error: {:?}", e); @@ -160,7 +235,16 @@ pub async fn load_cache(bin_path: &str, _remote_path: &str, is_dir: bool) -> (bo if let Some(os) = windmill_object_store::get_cache_object_store().await { let started = std::time::Instant::now(); - if let Ok(mut x) = windmill_object_store::attempt_fetch_bytes(os, _remote_path).await { + let fetched = bounded_cache_io( + "downloading", + _remote_path, + windmill_object_store::attempt_fetch_bytes(os, _remote_path), + ) + .await; + if let Err(e @ CacheIoError::TimedOut { .. }) = &fetched { + tracing::error!("{e}"); + } + if let Ok(mut x) = fetched { if is_dir { // Extract into a sibling temp dir then atomically publish it, // so a concurrent cold-load gating on metadata(bin_path) never @@ -227,12 +311,18 @@ pub async fn object_store_available() -> bool { pub async fn exists_in_object_store(_remote_path: &str) -> bool { #[cfg(all(feature = "enterprise", feature = "parquet"))] if let Some(os) = windmill_object_store::get_cache_object_store().await { - return os - .head(&windmill_object_store::object_store_reexports::Path::from( + let head = bounded_cache_io("checking", _remote_path, async { + os.head(&windmill_object_store::object_store_reexports::Path::from( _remote_path, )) .await - .is_ok(); + .map_err(|e| error::Error::ExecutionErr(format!("{e:?}"))) + }) + .await; + if let Err(e @ CacheIoError::TimedOut { .. }) = &head { + tracing::error!("{e}"); + } + return head.is_ok(); } false } @@ -258,12 +348,18 @@ pub async fn exists_in_cache(bin_path: &str, _remote_path: &str) -> bool { } else { #[cfg(all(feature = "enterprise", feature = "parquet"))] if let Some(os) = windmill_object_store::get_cache_object_store().await { - return os - .get(&windmill_object_store::object_store_reexports::Path::from( + let get = bounded_cache_io("checking", _remote_path, async { + os.get(&windmill_object_store::object_store_reexports::Path::from( _remote_path, )) .await - .is_ok(); + .map_err(|e| error::Error::ExecutionErr(format!("{e:?}"))) + }) + .await; + if let Err(e @ CacheIoError::TimedOut { .. }) = &get { + tracing::error!("{e}"); + } + return get.is_ok(); } return false; } @@ -307,17 +403,15 @@ pub async fn save_cache( origin.to_owned() }; - if let Err(e) = os - .put( - &Path::from(_remote_cache_path), - std::fs::read(&file_to_cache)?.into(), - ) - .await - { - tracing::error!( - "Failed to put bin to object store: {_remote_cache_path}. Error: {:?}", - e - ); + let bytes = std::fs::read(&file_to_cache)?; + let put = bounded_cache_io("uploading", _remote_cache_path, async { + os.put(&Path::from(_remote_cache_path), bytes.into()) + .await + .map_err(|e| error::Error::ExecutionErr(format!("{e:?}"))) + }) + .await; + if let Err(e) = put { + tracing::error!("Failed to put bin to object store: {_remote_cache_path}. Error: {e}"); } else { _cached_to_s3 = true; if is_dir {