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 <noreply@anthropic.com>

* fix(worker): bound the codebase download and label slow-step warnings

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

* fix(worker): log a stalled cache transfer once and ignore a zero cache timeout

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

---------

Co-authored-by: Claude Fable 5.1 <noreply@anthropic.com>

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<String>),
+    /// 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<Self> {
         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<Self> {
+        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<String>,
+        subject: WarnSubject,
     ) -> WarnAfterFuture<Self> {
         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<Self> {
         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<String>,
+        subject: WarnSubject,
     }
 }
 
@@ -1191,12 +1207,20 @@ impl<F: Future> Future for WarnAfterFuture<F> {
         // 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<F: Future> Future for WarnAfterFuture<F> {
             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)
             }

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",

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",

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<PulledCode
                 let dirs_splitted = bun_cache_path.split("/").collect_vec();
                 std::fs::create_dir_all(dirs_splitted[..dirs_splitted.len() - 1].join("/"))?;
 
-                let bytes = attempt_fetch_bytes(os, &path).await?;
+                let bytes = crate::global_cache::bounded_cache_io(
+                    "downloading",
+                    &path,
+                    attempt_fetch_bytes(os, &path),
+                )
+                .await?;
                 tracing::info!("loading {bun_cache_path} from object store");
 
                 windmill_common::worker::atomic_write_file_bytes(
@@ -1396,7 +1404,9 @@ pub async fn prebundle_bun_script(
 
     ensure_bundle_output_exists(&origin)?;
 
-    save_cache(&local_path, &remote_path, &origin, false).await?;
+    save_cache(&local_path, &remote_path, &origin, false)
+        .warn_after_seconds_for(60, "bundle cache save")
+        .await?;
 
     Ok(())
 }
@@ -1667,7 +1677,9 @@ pub async fn handle_bun_job(
             }
         };
 
-        let (cache, logs) = crate::global_cache::load_cache(&local_path, &remote_path, false).await;
+        let (cache, logs) = crate::global_cache::load_cache(&local_path, &remote_path, false)
+            .warn_after_seconds_for(60, "bundle cache load")
+            .await;
         (cache, logs, local_path, remote_path)
     } else {
         (false, "".to_string(), "".to_string(), "".to_string())
@@ -2293,7 +2305,10 @@ try {{
             let bundle_path = format!("{job_dir}/main.js");
             ensure_bundle_output_exists(&bundle_path)?;
             if !local_path.is_empty() {
-                match save_cache(&local_path, &remote_path, &bundle_path, false).await {
+                match save_cache(&local_path, &remote_path, &bundle_path, false)
+                    .warn_after_seconds_for(60, "bundle cache save")
+                    .await
+                {
                     Err(e) => {
                         let em = format!("could not save {local_path} to bundle cache: {e:?}");
                         tracing::error!(em)

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::<u64>().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<CacheIoError> 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<T>(
+    what: &str,
+    path: &str,
+    io: impl std::future::Future<Output = error::Result<T>>,
+) -> Result<T, CacheIoError> {
+    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<dyn ObjectStore>,
@@ -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 {