diff --git a/backend/tests/nativets_dedicated.rs b/backend/tests/nativets_dedicated.rs index 1c6987a480..1e9aef3228 100644 --- a/backend/tests/nativets_dedicated.rs +++ b/backend/tests/nativets_dedicated.rs @@ -12,7 +12,7 @@ mod prewarmed_isolate_tests { use windmill_worker::{build_loader, LoaderMode, BUN_PATH}; fn default_annotation() -> NativeAnnotation { - NativeAnnotation { useragent: None, proxy: None } + NativeAnnotation::default() } /// Bundle a TypeScript script into JS suitable for `PrewarmedIsolate`. diff --git a/backend/windmill-runtime-nativets/src/lib.rs b/backend/windmill-runtime-nativets/src/lib.rs index 6e80b1405e..d44213d3bc 100644 --- a/backend/windmill-runtime-nativets/src/lib.rs +++ b/backend/windmill-runtime-nativets/src/lib.rs @@ -26,7 +26,7 @@ use std::{ cell::RefCell, path::PathBuf, rc::Rc, - sync::{Arc, Mutex}, + sync::{Arc, LazyLock, Mutex}, }; // Re-export deno_telemetry for use by windmill-worker's otel proxy @@ -182,10 +182,33 @@ struct LogString { pub s: mpsc::UnboundedSender, } -#[derive(Clone)] +#[derive(Clone, Default)] pub struct NativeAnnotation { pub useragent: Option, pub proxy: Option<(String, Option<(String, String)>)>, + /// `//fetch_response_timeout `: per-script override of + /// [`default_fetch_response_timeout_secs`]. `Some(0)` disables it for this + /// script; `None` leaves the default in force. + pub fetch_response_timeout_secs: Option, +} + +/// How long `fetch()` waits for a response to begin, in seconds; `0` disables. +/// +/// Covers everything up to the response headers and stops there, so a body may +/// then stream for any length of time. `src/runtime.js` holds the semantics. +/// +/// Must exceed `TIMEOUT_WAIT_RESULT` (default 600), which holds synchronous job +/// calls open without headers. Raising that hot-reloaded instance setting may +/// also require raising `WINDMILL_FETCH_RESPONSE_TIMEOUT_SECS` in the deployment +/// and restarting workers: this environment value is cached for the process. +pub fn default_fetch_response_timeout_secs() -> u64 { + static SECS: LazyLock = LazyLock::new(|| { + std::env::var("WINDMILL_FETCH_RESPONSE_TIMEOUT_SECS") + .ok() + .and_then(|x| x.trim().parse::().ok()) + .unwrap_or(900) + }); + *SECS } /// Serializes V8 isolate creation as defense-in-depth against concurrent @@ -401,7 +424,7 @@ pub fn transpile_ts(expr: String) -> anyhow::Result { } pub fn get_annotation(inner_content: &str) -> NativeAnnotation { - let mut res = NativeAnnotation { useragent: None, proxy: None }; + let mut res = NativeAnnotation::default(); let anns = inner_content .lines() @@ -414,6 +437,13 @@ pub fn get_annotation(inner_content: &str) -> NativeAnnotation { res.useragent = Some(ann.trim_start_matches("useragent").trim().to_string()); } else if ann.starts_with("proxy") { res.proxy = capture_proxy(ann.trim_start_matches("proxy").trim()); + } else if ann.starts_with("fetch_response_timeout") { + // A typo falls back to the default, never to "no timeout". + res.fetch_response_timeout_secs = ann + .trim_start_matches("fetch_response_timeout") + .trim() + .parse::() + .ok(); } } res @@ -554,6 +584,16 @@ pub(crate) fn create_nativets_runtime( let ops = vec![op_get_static_args(), op_log()]; let ext = Extension { name: "windmill", ops: ops.into(), ..Default::default() }; + // deno_web's setTimeout puts its delay through `webidl.converters.long`, + // which wraps at 32 bits: past i32::MAX ms (~24.8 days) the delay comes out + // negative and fires immediately, so an over-generous setting would abort + // every fetch on the spot. Cap rather than wrap. + let fetch_response_timeout_ms = ann + .fetch_response_timeout_secs + .unwrap_or_else(default_fetch_response_timeout_secs) + .saturating_mul(1000) + .min(i32::MAX as u64); + let fetch_options = deno_fetch::Options { root_cert_store_provider: NATIVE_ROOT_CERT_STORE_PROVIDER.clone(), user_agent: ann.useragent.unwrap_or_else(|| "windmill/beta".to_string()), @@ -625,10 +665,15 @@ pub(crate) fn create_nativets_runtime( } // Per-isolate JS init that can't run in the snapshot (runtime.js executes at - // snapshot-build time): currently seeds performance.timeOrigin via - // setTimeOrigin(), which must read this isolate's wall clock. + // snapshot-build time): the wall clock behind performance.timeOrigin and the + // fetch response timeout are both per-isolate values. js_runtime - .execute_script("", "globalThis.__wmInitPerIsolate()") + .execute_script( + "", + format!( + "globalThis.__wmInitPerIsolate({{ fetchResponseTimeoutMs: {fetch_response_timeout_ms} }})" + ), + ) .map_err(windmill_common::error::to_anyhow)?; Ok(CreatedRuntime { js_runtime, log_receiver, memory_limit_rx }) diff --git a/backend/windmill-runtime-nativets/src/runtime.js b/backend/windmill-runtime-nativets/src/runtime.js index 01bec09983..88e10f765a 100644 --- a/backend/windmill-runtime-nativets/src/runtime.js +++ b/backend/windmill-runtime-nativets/src/runtime.js @@ -30,9 +30,115 @@ import * as performance from "ext:deno_web/15_performance.js"; import "ext:deno_web/16_image_data.js"; import "ext:deno_fetch/27_eventsource.js"; +// deno_fetch applies no deadline, so a peer that accepts a request and then +// never answers leaves `await fetch(...)` pending until the job timeout, which +// self-hosted defaults to 7 days. +const ORIGINAL_FETCH = fetch.fetch; + +// deno_web's timers reject any `this` other than undefined/globalThis, so +// `timers.setTimeout(...)` passes the module namespace and throws "Illegal +// invocation". +const setTimeoutUnbound = timers.setTimeout; +const clearTimeoutUnbound = timers.clearTimeout; +// Captured before user code shares the isolate and could redefine them, the +// way deno's own modules reach intrinsics through primordials. +const PromiseReject = Promise.reject.bind(Promise); +const promiseThen = Function.prototype.call.bind(Promise.prototype.then); +const abortSignalAny = abortSignal.AbortSignal.any.bind(abortSignal.AbortSignal); +const abortControllerAbort = Function.prototype.call.bind( + abortSignal.AbortController.prototype.abort, +); +const ReflectApply = Reflect.apply; + +// Installed per isolate by __wmInitPerIsolate; 0 disables. Only a backstop +// for the impossible case of fetch running before that init. +let fetchResponseTimeoutMs = 900_000; + +function fetchResponseTimeoutError(requestUrl, timeoutMs) { + let target; + try { + const parsed = new url.URL(requestUrl); + // Query and fragment routinely carry tokens, and this reaches a job log. + target = parsed.origin + parsed.pathname; + } catch { + target = "the request target"; + } + return new domException.DOMException( + `fetch to ${target} timed out: no response headers arrived within ` + + `${Math.round(timeoutMs / 1000)}s (this covers connect, request upload ` + + `and the wait for the server to start replying; once a response begins ` + + `it is never interrupted). Change it per script with ` + + `"//fetch_response_timeout " (0 disables), or instance-wide ` + + `with WINDMILL_FETCH_RESPONSE_TIMEOUT_SECS.`, + "TimeoutError", + ); +} + globalThis.atob = base64.atob; globalThis.btoa = base64.btoa; -globalThis.fetch = fetch.fetch; +// Not `async`, for the same reason deno_fetch's own outer fetch isn't: WPT +// pins that an aborted fetch settles in the same tick, which adopting its +// promise through another one would break. Construction still has to reject +// rather than throw, so it is caught and handed back as a rejection. +globalThis.fetch = function fetch(input, init = undefined) { + const timeoutMs = fetchResponseTimeoutMs; + // Forwarded with the original argument count, so deno still sees an empty + // call as empty and raises its own "1 argument required". The default on + // `init` is what keeps `fetch.length` at 1, as the standard has it. + if (!(timeoutMs > 0) || arguments.length < 1) { + return ReflectApply(ORIGINAL_FETCH, undefined, arguments); + } + + let req; + let controller; + let signal; + try { + // RequestInit is a WebIDL dictionary: copying it drops inherited and + // non-enumerable members, and inheriting from it runs accessors against the + // wrong receiver. Hand it to the same Request constructor fetch() would, + // and carry our own signal in an init of our own. + req = new request.Request(input, init); + + // `req.signal` is deno's own resolution of init.signal over an input + // Request's signal, so combining with it preserves the caller's abort and + // reason while ours only adds a ceiling. + controller = new abortSignal.AbortController(); + signal = abortSignalAny([req.signal, controller.signal]); + } catch (e) { + return PromiseReject(e); + } + + // Already aborted: hand back deno's own settled rejection untouched, and arm + // nothing -- there is no response to wait for. + if (signal.aborted) { + return ORIGINAL_FETCH(req, { signal }); + } + + let timer = setTimeoutUnbound(() => { + timer = undefined; + abortControllerAbort(controller, fetchResponseTimeoutError(req.url, timeoutMs)); + }, timeoutMs); + // Disarmed on headers, never on body completion: a response that has begun + // arriving must be free to stream for as long as it needs. + const disarm = () => { + if (timer !== undefined) { + clearTimeoutUnbound(timer); + timer = undefined; + } + }; + + return promiseThen( + ORIGINAL_FETCH(req, { signal }), + (res) => { + disarm(); + return res; + }, + (e) => { + disarm(); + throw e; + }, + ); +}; globalThis.Request = request.Request; globalThis.Response = response.Response; globalThis.Blob = file.Blob; @@ -123,7 +229,11 @@ Object.assign(globalThis, { // Per-isolate init, invoked from Rust after the snapshot is restored (this // module body runs at snapshot-build time, not per isolate). -globalThis.__wmInitPerIsolate = () => { +globalThis.__wmInitPerIsolate = (config) => { + if (config != null && typeof config.fetchResponseTimeoutMs === "number") { + fetchResponseTimeoutMs = config.fetchResponseTimeoutMs; + } + // setTimeOrigin() seeds performance.timeOrigin from the isolate's wall clock; // without it timeOrigin is undefined and `timeOrigin + performance.now()` is NaN. performance.setTimeOrigin(); diff --git a/backend/windmill-runtime-nativets/src/smoke_tests.rs b/backend/windmill-runtime-nativets/src/smoke_tests.rs index b581a9f1fe..f3da460227 100644 --- a/backend/windmill-runtime-nativets/src/smoke_tests.rs +++ b/backend/windmill-runtime-nativets/src/smoke_tests.rs @@ -24,7 +24,7 @@ use crate::{transpile_ts, NativeAnnotation, PrewarmedIsolate, PrewarmedResult}; /// positional args, and return the isolate's result + captured logs. async fn run_ts(ts: &str, arg_names: &[&str], args: serde_json::Value) -> PrewarmedResult { let js = transpile_ts(ts.to_string()).expect("transpile_ts failed"); - let ann = NativeAnnotation { useragent: None, proxy: None }; + let ann = NativeAnnotation::default(); let arg_names: Vec = arg_names.iter().map(|s| s.to_string()).collect(); let mut iso = PrewarmedIsolate::spawn(String::new(), js, ann, arg_names, None); iso.wait_ready().await.expect("isolate failed to pre-warm"); @@ -232,7 +232,7 @@ export async function main(i: number): Promise { for i in 0..N { let js = js.clone(); let h = tokio::spawn(async move { - let ann = NativeAnnotation { useragent: None, proxy: None }; + let ann = NativeAnnotation::default(); let mut iso = PrewarmedIsolate::spawn(String::new(), js, ann, vec!["i".to_string()], None); iso.wait_ready().await.expect("pre-warm failed"); diff --git a/backend/windmill-runtime-nativets/tests/annotation.rs b/backend/windmill-runtime-nativets/tests/annotation.rs new file mode 100644 index 0000000000..d8319a87a5 --- /dev/null +++ b/backend/windmill-runtime-nativets/tests/annotation.rs @@ -0,0 +1,37 @@ +//! `//fetch_response_timeout` parsing. Cheap (no V8), so it stays out of the +//! e2e file. + +use windmill_runtime_nativets::get_annotation; + +#[test] +fn a_value_is_parsed_and_zero_stays_distinct_from_absent() { + // Collapsing `Some(0)` to `None` would silently reinstate the default on a + // script that explicitly asked for no limit. + assert_eq!( + get_annotation("//native\n//fetch_response_timeout 30\n").fetch_response_timeout_secs, + Some(30) + ); + assert_eq!( + get_annotation("//fetch_response_timeout 0\n").fetch_response_timeout_secs, + Some(0) + ); + assert_eq!( + get_annotation("//native\n").fetch_response_timeout_secs, + None + ); +} + +#[test] +fn a_malformed_value_falls_back_to_the_default_not_to_no_timeout() { + for src in [ + "//fetch_response_timeout abc\n", + "//fetch_response_timeout\n", + "//fetch_response_timeout -5\n", + ] { + assert_eq!( + get_annotation(src).fetch_response_timeout_secs, + None, + "{src:?} should leave the default in force" + ); + } +} diff --git a/backend/windmill-runtime-nativets/tests/fetch_response_timeout.rs b/backend/windmill-runtime-nativets/tests/fetch_response_timeout.rs new file mode 100644 index 0000000000..5118b5f93f --- /dev/null +++ b/backend/windmill-runtime-nativets/tests/fetch_response_timeout.rs @@ -0,0 +1,593 @@ +//! The nativets `fetch()` response timeout. +//! +//! Two halves of one contract, and a fix that satisfies only the first is worse +//! than no fix: +//! +//! 1. a peer that accepts a request and never answers is given up on +//! 2. a response that has begun arriving is never cut off, however long it +//! takes in total +//! +//! (2) rules out the obvious implementation — `AbortSignal.timeout(N)` around +//! every fetch would satisfy (1) and break every streaming response and long +//! download. +//! +//! Hermetic: loopback listeners, no egress. + +use std::sync::Arc; +use std::time::{Duration, Instant}; + +use tokio::io::{AsyncReadExt, AsyncWriteExt}; +use tokio::net::TcpListener; +use tokio::sync::Mutex; + +use windmill_runtime_nativets::{transpile_ts, NativeAnnotation, PrewarmedIsolate}; + +/// The annotation is in whole seconds, so the tests scale around this. +const TIMEOUT_SECS: u64 = 2; + +async fn run_with_timeout_secs(ts: &str, secs: u64) -> Result { + let js = transpile_ts(ts.to_string()).expect("transpile_ts failed"); + let ann = + NativeAnnotation { fetch_response_timeout_secs: Some(secs), ..NativeAnnotation::default() }; + let mut iso = PrewarmedIsolate::spawn(String::new(), js, ann, vec![], None); + iso.wait_ready().await.expect("isolate failed to pre-warm"); + let res = iso + .start_execution("{}".to_string()) + .wait() + .await + .expect("isolate panicked"); + res.result.map(|raw| raw.get().to_string()) +} + +/// A peer that reads the request and then answers nothing, closing only after +/// `close_after` so no test leaves an isolate wedged on a pending fetch. +/// +/// The socket is held rather than dropped on accept: dropping it sends a FIN, +/// which surfaces as a connection error — the easy failure, not this one. +async fn spawn_silent_peer(seen: Arc>>, close_after: Duration) -> u16 { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let port = listener.local_addr().unwrap().port(); + tokio::spawn(async move { + while let Ok((mut sock, _)) = listener.accept().await { + let seen = seen.clone(); + tokio::spawn(async move { + let mut buf = [0u8; 8192]; + if let Ok(n) = sock.read(&mut buf).await { + seen.lock().await.extend_from_slice(&buf[..n]); + } + tokio::time::sleep(close_after).await; + drop(sock); + }); + } + }); + port +} + +/// A peer that records the request it received and answers 200 immediately. +async fn spawn_echo_peer(seen: Arc>>) -> u16 { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let port = listener.local_addr().unwrap().port(); + tokio::spawn(async move { + while let Ok((mut sock, _)) = listener.accept().await { + let seen = seen.clone(); + tokio::spawn(async move { + let mut buf = [0u8; 8192]; + if let Ok(n) = sock.read(&mut buf).await { + seen.lock().await.extend_from_slice(&buf[..n]); + } + let _ = sock + .write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 0\r\n\r\n") + .await; + }); + } + }); + port +} + +/// A peer that responds after `headers_after`, then dribbles a chunked body out +/// over `chunks * chunk_every`. +async fn spawn_streaming_peer( + headers_after: Duration, + chunks: usize, + chunk_every: Duration, +) -> u16 { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let port = listener.local_addr().unwrap().port(); + tokio::spawn(async move { + while let Ok((mut sock, _)) = listener.accept().await { + tokio::spawn(async move { + let mut buf = [0u8; 8192]; + let _ = sock.read(&mut buf).await; + tokio::time::sleep(headers_after).await; + if sock + .write_all( + b"HTTP/1.1 200 OK\r\nContent-Type: text/plain\r\nTransfer-Encoding: chunked\r\n\r\n", + ) + .await + .is_err() + { + return; + } + for _ in 0..chunks { + tokio::time::sleep(chunk_every).await; + if sock.write_all(b"1\r\nx\r\n").await.is_err() { + return; + } + } + let _ = sock.write_all(b"0\r\n\r\n").await; + }); + } + }); + port +} + +const POST_TO_SILENT_PEER: &str = r#" +export async function main(): Promise { + const res = await fetch("http://127.0.0.1:{port}/orders", { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ id: 1 }), + }); + return res.status; +} +"#; + +fn post_script(port: u16) -> String { + POST_TO_SILENT_PEER.replace("{port}", &port.to_string()) +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn a_peer_that_never_answers_is_given_up_on() { + let seen = Arc::new(Mutex::new(Vec::new())); + let port = spawn_silent_peer(seen.clone(), Duration::from_secs(60)).await; + + let started = Instant::now(); + let err = run_with_timeout_secs(&post_script(port), TIMEOUT_SECS) + .await + .expect_err("fetch against a peer that never answers must not resolve"); + let elapsed = started.elapsed(); + + assert!( + elapsed >= Duration::from_secs(TIMEOUT_SECS), + "gave up after {elapsed:?}, before the configured {TIMEOUT_SECS}s -- \ + the timeout is firing on something other than the wait for a response", + ); + assert!( + elapsed < Duration::from_secs(TIMEOUT_SECS + 15), + "took {elapsed:?} to give up", + ); + + // The message has to stand on its own in a job log: a hung request is + // otherwise indistinguishable from a slow one. + assert!( + err.contains("no response headers arrived within"), + "error should explain what timed out, got: {err}", + ); + assert!( + err.contains("/orders") && err.contains("fetch_response_timeout"), + "error should name the target and how to change the limit, got: {err}", + ); + + // Without this the test would also pass if the request never went out. + let body = String::from_utf8_lossy(&seen.lock().await.clone()).to_string(); + assert!( + body.contains("POST /orders"), + "peer should have received the request, got: {body}", + ); +} + +/// The counterfactual for the test above: with the timeout disabled, the same +/// script against the same peer is still running well past the point the +/// timeout would have fired. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn without_a_timeout_the_same_request_keeps_running() { + let seen = Arc::new(Mutex::new(Vec::new())); + // Closes eventually, so the isolate unwinds instead of pinning a blocking + // task for the rest of the test binary's life. + let port = spawn_silent_peer(seen, Duration::from_secs(TIMEOUT_SECS * 3)).await; + + let still_running = tokio::time::timeout( + Duration::from_secs(TIMEOUT_SECS * 2), + run_with_timeout_secs(&post_script(port), 0), + ) + .await + .is_err(); + + assert!( + still_running, + "with the timeout disabled the request should still have been pending \ + at {}s -- if it ends on its own, the test above proves nothing", + TIMEOUT_SECS * 2, + ); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn a_streaming_body_outliving_the_timeout_is_not_cut_off() { + // Headers land fast, then the body trickles well past the timeout. A + // total-duration timeout fails here; that is the point of the test. + let port = + spawn_streaming_peer(Duration::from_millis(200), 10, Duration::from_millis(500)).await; + + let ts = format!( + r#" +export async function main(): Promise {{ + const res = await fetch("http://127.0.0.1:{port}/stream"); + return `${{res.status}}:${{(await res.text()).length}}`; +}} +"# + ); + + let started = Instant::now(); + let out = run_with_timeout_secs(&ts, TIMEOUT_SECS) + .await + .expect("a streaming response must not be interrupted"); + let elapsed = started.elapsed(); + + assert_eq!(out, "\"200:10\"", "full body should arrive intact"); + assert!( + elapsed > Duration::from_secs(TIMEOUT_SECS), + "the transfer ({elapsed:?}) has to outlast the {TIMEOUT_SECS}s timeout \ + for this to be exercising anything", + ); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn a_slow_but_answering_peer_is_not_cut_off() { + // A quarter of the window rather than half: CI runs this at + // --test-threads=10 alongside other V8 isolates, and this margin is what + // absorbs executor starvation. + let port = spawn_streaming_peer( + Duration::from_millis(1_000 * TIMEOUT_SECS / 4), + 1, + Duration::from_millis(10), + ) + .await; + + let ts = format!( + r#" +export async function main(): Promise {{ + const res = await fetch("http://127.0.0.1:{port}/slow"); + await res.text(); + return res.status; +}} +"# + ); + + let out = run_with_timeout_secs(&ts, TIMEOUT_SECS) + .await + .expect("a slow but answering peer must not be cut off"); + assert_eq!(out, "200"); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn an_over_large_timeout_does_not_wrap_into_an_instant_one() { + // deno_web's setTimeout puts its delay through `webidl.converters.long`, + // which wraps at 32 bits. Unclamped, this ~46-day setting wraps negative and + // aborts immediately -- asking for a longer leash would kill every fetch. + let port = spawn_streaming_peer(Duration::from_millis(50), 1, Duration::from_millis(10)).await; + + let ts = format!( + r#" +export async function main(): Promise {{ + const res = await fetch("http://127.0.0.1:{port}/ok"); + await res.text(); + return res.status; +}} +"# + ); + + let out = run_with_timeout_secs(&ts, 4_000_000) + .await + .expect("an over-large timeout must not abort the request"); + assert_eq!(out, "200"); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn a_caller_abort_still_wins_with_its_own_reason() { + let seen = Arc::new(Mutex::new(Vec::new())); + let port = spawn_silent_peer(seen, Duration::from_secs(60)).await; + + let ts = format!( + r#" +export async function main(): Promise {{ + const ac = new AbortController(); + setTimeout(() => ac.abort(new Error("caller_abort_marker")), 200); + try {{ + await fetch("http://127.0.0.1:{port}/probe", {{ signal: ac.signal }}); + return "unexpectedly resolved"; + }} catch (e) {{ + return String((e as Error).message); + }} +}} +"# + ); + + let out = run_with_timeout_secs(&ts, TIMEOUT_SECS) + .await + .expect("script should catch its own abort"); + assert!( + out.contains("caller_abort_marker"), + "caller's abort reason should survive being combined with ours, got: {out}", + ); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn the_timeout_is_catchable_as_a_timeout_error() { + // Scripts that retry on transient failures need to recognise this one; + // `TimeoutError` matches AbortSignal.timeout()'s reason. + let seen = Arc::new(Mutex::new(Vec::new())); + let port = spawn_silent_peer(seen, Duration::from_secs(60)).await; + + let ts = format!( + r#" +export async function main(): Promise {{ + try {{ + await fetch("http://127.0.0.1:{port}/probe"); + return "unexpectedly resolved"; + }} catch (e) {{ + return (e as Error).name; + }} +}} +"# + ); + + let out = run_with_timeout_secs(&ts, TIMEOUT_SECS) + .await + .expect("script should catch the timeout"); + assert_eq!(out, "\"TimeoutError\""); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn an_init_whose_members_are_inherited_is_not_flattened() { + // RequestInit is a WebIDL dictionary and deno_fetch reads its members with + // plain property gets, so they may sit on the prototype chain or be + // non-enumerable. Object spread copies neither, which would silently + // downgrade this POST to a GET and drop the header. + let seen = Arc::new(Mutex::new(Vec::new())); + let port = spawn_echo_peer(seen.clone()).await; + + let ts = format!( + r#" +declare const Object: any; +export async function main(): Promise {{ + const base = {{ method: "POST", headers: {{ "x-probe": "yes" }} }}; + const res = await fetch("http://127.0.0.1:{port}/inherited", Object.create(base)); + return res.status; +}} +"# + ); + + let out = run_with_timeout_secs(&ts, TIMEOUT_SECS) + .await + .expect("request should succeed"); + assert_eq!(out, "200"); + + let body = String::from_utf8_lossy(&seen.lock().await.clone()).to_string(); + assert!( + body.starts_with("POST /inherited"), + "inherited `method` should survive, got: {body}", + ); + assert!( + body.to_lowercase().contains("x-probe: yes"), + "inherited `headers` should survive, got: {body}", + ); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn a_non_dictionary_init_still_fails_loudly() { + // deno's dictionary converter throws on a non-object init. Copying members + // into a fresh object instead of inheriting would turn `"POST"` into + // {0:"P",1:"O",...} -- a valid dictionary with ignored keys, i.e. a silent + // GET where the caller used to get a TypeError. + let seen = Arc::new(Mutex::new(Vec::new())); + let port = spawn_echo_peer(seen).await; + + let ts = format!( + r#" +export async function main(): Promise {{ + try {{ + await fetch("http://127.0.0.1:{port}/x", "POST" as any); + return "unexpectedly resolved"; + }} catch (e) {{ + return (e as Error).name; + }} +}} +"# + ); + + let out = run_with_timeout_secs(&ts, TIMEOUT_SECS) + .await + .expect("script should catch the error"); + assert_eq!(out, "\"TypeError\""); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn an_accessor_backed_init_reads_against_its_own_receiver() { + // A getter on the init must run with the object it was defined on as `this`, + // or a private field is unreachable and it throws. Carrying the init across + // by inheritance rather than by handing it to Request would break this. + let seen = Arc::new(Mutex::new(Vec::new())); + let port = spawn_echo_peer(seen.clone()).await; + + let ts = format!( + r#" +class Init {{ + #method = "POST"; + #body = "from-private-field"; + get method(): string {{ return this.#method; }} + get body(): string {{ return this.#body; }} +}} +export async function main(): Promise {{ + const res = await fetch("http://127.0.0.1:{port}/accessor", new Init() as any); + return res.status; +}} +"# + ); + + let out = run_with_timeout_secs(&ts, TIMEOUT_SECS) + .await + .expect("an accessor-backed init must not throw"); + assert_eq!(out, "200"); + + let body = String::from_utf8_lossy(&seen.lock().await.clone()).to_string(); + assert!( + body.starts_with("POST /accessor") && body.contains("from-private-field"), + "getter-provided method and body should both reach the wire, got: {body}", + ); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn a_request_input_keeps_its_body_and_headers() { + // The wrapper builds a Request and hands that to fetch, so the body survives + // one more construction than it used to -- deno proxies it rather than + // consuming it, and this pins that. + let seen = Arc::new(Mutex::new(Vec::new())); + let port = spawn_echo_peer(seen.clone()).await; + + let ts = format!( + r#" +export async function main(): Promise {{ + const req = new Request("http://127.0.0.1:{port}/from-request", {{ + method: "PUT", + headers: {{ "x-probe": "yes" }}, + body: "payload-body", + }}); + const res = await fetch(req); + return res.status; +}} +"# + ); + + let out = run_with_timeout_secs(&ts, TIMEOUT_SECS) + .await + .expect("a Request input must work"); + assert_eq!(out, "200"); + + let body = String::from_utf8_lossy(&seen.lock().await.clone()).to_string(); + assert!( + body.starts_with("PUT /from-request") + && body.to_lowercase().contains("x-probe: yes") + && body.contains("payload-body"), + "method, headers and body should all survive, got: {body}", + ); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn an_already_aborted_fetch_settles_in_the_same_tick() { + // deno_fetch keeps its outer fetch non-async and returns an already-settled + // rejection untouched, because WPT pins that an aborted fetch settles in the + // same tick. Adopting it through another promise pushes the rejection behind + // any microtask queued after the call. + let ts = r#" +export async function main(): Promise { + const order: string[] = []; + const ac = new AbortController(); + ac.abort(); + const f = fetch("http://127.0.0.1:1/x", { signal: ac.signal }) + .catch(() => { order.push("fetch"); }); + Promise.resolve().then(() => { order.push("queued-after"); }); + await f; + await new Promise((r) => setTimeout(r, 0)); + return order.join(","); +} +"#; + + let out = run_with_timeout_secs(ts, TIMEOUT_SECS) + .await + .expect("script should run"); + assert_eq!( + out, "\"fetch,queued-after\"", + "the rejection must land before a microtask queued after the call", + ); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn the_wrapper_keeps_fetch_s_own_shape() { + // The wrapper is indistinguishable from deno's fetch on three counts a + // script can observe: its arity, the error for an empty call, and not + // depending on a mutable `Promise.prototype.then` the way an ordinary + // property lookup would. + let seen = Arc::new(Mutex::new(Vec::new())); + let port = spawn_echo_peer(seen).await; + + let ts = format!( + r#" +declare const Promise: any; +declare const AbortSignal: any; +declare const AbortController: any; +export async function main(): Promise {{ + const arity = (fetch as any).length; + + // The message, not just the type: forwarding two explicit `undefined`s + // would still throw a TypeError, just deno's invalid-URL one instead of + // its required-argument one. + let emptyCall = "resolved"; + try {{ + await (fetch as any)(); + }} catch (e) {{ + emptyCall = (e as Error).message.includes("1 argument required") + ? "required-argument" + : `other(${{(e as Error).message}})`; + }} + + // Patched only across the call: the wrapper reaches for these while + // building its return value, and awaiting under a patched Promise + // prototype would instead measure V8 treating it as a plain thenable. + const originalThen = Promise.prototype.then; + const originalAny = AbortSignal.any; + const originalAbort = AbortController.prototype.abort; + let pending: any; + try {{ + Promise.prototype.then = undefined; + (AbortSignal as any).any = undefined; + (AbortController.prototype as any).abort = undefined; + pending = fetch("http://127.0.0.1:{port}/shape"); + }} finally {{ + Promise.prototype.then = originalThen; + (AbortSignal as any).any = originalAny; + (AbortController.prototype as any).abort = originalAbort; + }} + const status = (await pending).status; + + return `${{arity}}:${{emptyCall}}:${{status}}`; +}} +"# + ); + + let out = run_with_timeout_secs(&ts, TIMEOUT_SECS) + .await + .expect("script should run"); + assert_eq!(out, "\"1:required-argument:200\""); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn the_timer_aborts_through_a_captured_intrinsic() { + // The timeout has to fire for this one: `AbortController.prototype.abort` + // is patched out for the whole wait, so an ordinary lookup would throw + // inside the timer callback and leave the request pending forever. + let seen = Arc::new(Mutex::new(Vec::new())); + let port = spawn_silent_peer(seen, Duration::from_secs(TIMEOUT_SECS * 5)).await; + + let ts = format!( + r#" +declare const AbortController: any; +export async function main(): Promise {{ + const originalAbort = AbortController.prototype.abort; + AbortController.prototype.abort = undefined; + try {{ + await fetch("http://127.0.0.1:{port}/patched-abort"); + return "unexpectedly resolved"; + }} catch (e) {{ + return (e as Error).name; + }} finally {{ + AbortController.prototype.abort = originalAbort; + }} +}} +"# + ); + + let out = run_with_timeout_secs(&ts, TIMEOUT_SECS) + .await + .expect("the timeout must still fire"); + assert_eq!(out, "\"TimeoutError\""); +} diff --git a/backend/windmill-runtime-nativets/tests/otel_e2e.rs b/backend/windmill-runtime-nativets/tests/otel_e2e.rs index cadddb2fcf..22fd18de90 100644 --- a/backend/windmill-runtime-nativets/tests/otel_e2e.rs +++ b/backend/windmill-runtime-nativets/tests/otel_e2e.rs @@ -136,7 +136,7 @@ export async function main(): Promise {{ "# ); let js = transpile_ts(ts).expect("transpile failed"); - let ann = NativeAnnotation { useragent: None, proxy: None }; + let ann = NativeAnnotation::default(); let mut iso = PrewarmedIsolate::spawn(String::new(), js, ann, vec![], None); iso.wait_ready().await.expect("isolate failed to pre-warm"); diff --git a/backend/windmill-runtime-nativets/tests/response_timeout_env.rs b/backend/windmill-runtime-nativets/tests/response_timeout_env.rs new file mode 100644 index 0000000000..a206c63134 --- /dev/null +++ b/backend/windmill-runtime-nativets/tests/response_timeout_env.rs @@ -0,0 +1,13 @@ +//! Its own test binary: the setting is read once per process through a +//! `LazyLock`, so nothing else may resolve it first. Adding a second test to +//! this file breaks that isolation. + +use windmill_runtime_nativets::default_fetch_response_timeout_secs; + +#[test] +fn the_env_var_is_what_operators_actually_set() { + // A typo in the variable's name would compile, pass every other test, and + // silently hand every operator the built-in default instead. + std::env::set_var("WINDMILL_FETCH_RESPONSE_TIMEOUT_SECS", "17"); + assert_eq!(default_fetch_response_timeout_secs(), 17); +}