diff --git a/src/runtime/webcore/s3/client.rs b/src/runtime/webcore/s3/client.rs index 329617c105d6..bb20e6567df5 100644 --- a/src/runtime/webcore/s3/client.rs +++ b/src/runtime/webcore/s3/client.rs @@ -295,8 +295,9 @@ pub(crate) fn list_objects( let headers = bun_http::Headers::from_pico_http_headers(result.headers()); let task_ptr = bun_core::heap::into_raw(Box::new(S3HttpSimpleTask { - // Written below via `MaybeUninit::write` before any read. + // Both written by `S3HttpSimpleTask::schedule` below. http: core::mem::MaybeUninit::uninit(), + async_http_id: 0, sign_result: result, callback_context, callback: s3_simple_request::Callback::ListObjects(callback), @@ -343,7 +344,7 @@ pub(crate) fn list_objects( // JS thread (request setup): read options from the current VM. let vm = VirtualMachine::get(); - task.http.write(bun_http::AsyncHTTP::init( + let http = bun_http::AsyncHTTP::init( bun_http::Method::GET, url, task.headers.entries.clone().expect("OOM"), @@ -364,19 +365,10 @@ pub(crate) fn list_objects( signals: Some(task.signal_store.to()), ..Default::default() }, - )); - - // queue http request - bun_http::http_thread::init(&Default::default()); - let mut batch = bun_threading::thread_pool::Batch::default(); - // SAFETY: `http` was initialised by `task.http.write(...)` immediately above. - unsafe { task.http.assume_init_mut() }.schedule(&mut batch); - // Out on the HTTP thread until its final callback: the VM aborts it at - // teardown (registry) and waits for it (embedded work). - task.loop_handle.embedded_work_scheduled(); - crate::jsc_hooks::ActiveHandle::S3Request(core::ptr::NonNull::new(task_ptr).expect("task")) - .register(); - bun_http::HTTPThread::schedule(batch); + ); + // SAFETY: `task_ptr` was allocated above and has not been handed to anything yet + // (`task` is not used past this point); `http`'s callback context is `task_ptr`. + unsafe { S3HttpSimpleTask::schedule(task_ptr, http) }; Ok(()) } @@ -1227,8 +1219,9 @@ fn download_stream( }; let task_ptr = bun_core::heap::into_raw(S3HttpDownloadStreamingTask::new( S3HttpDownloadStreamingTask { - // `http: undefined` — fully overwritten by `task.http.write(AsyncHTTP::init(...))` below. + // Both written by `S3HttpDownloadStreamingTask::schedule` below. http: core::mem::MaybeUninit::uninit(), + async_http_id: 0, sign_result: result, proxy_url: owned_proxy, callback_context: NonNull::new(callback_context.cast::<()>()) @@ -1252,7 +1245,6 @@ fn download_stream( crate::webcore::s3::download_stream::State::default().0, ), concurrent_task: Default::default(), - async_http_id: 0, }, )); // SAFETY: just allocated via heap::alloc, non-null; lifetime owned by HTTP callback @@ -1283,7 +1275,7 @@ fn download_stream( let verbose = vm.get_verbose_fetch(); let reject_unauthorized = vm.get_tls_reject_unauthorized(); - task.http.write(bun_http::AsyncHTTP::init( + let http = bun_http::AsyncHTTP::init( bun_http::Method::GET, url, task.headers.entries.clone().expect("OOM"), @@ -1304,22 +1296,10 @@ fn download_stream( reject_unauthorized: Some(reject_unauthorized), ..Default::default() }, - )); - // SAFETY: `http` was initialised by `task.http.write(...)` immediately above. - let http = unsafe { task.http.assume_init_mut() }; - task.async_http_id = http.async_http_id; - // enable streaming - http.enable_response_body_streaming(); - // queue http request - bun_http::http_thread::init(&Default::default()); - let mut batch = bun_threading::thread_pool::Batch::default(); - http.schedule(&mut batch); - // Out on the HTTP thread until its final callback: the VM aborts it at - // teardown (registry) and waits for it (embedded work). - task.loop_handle.embedded_work_scheduled(); - crate::jsc_hooks::ActiveHandle::S3Download(core::ptr::NonNull::new(task_ptr).expect("task")) - .register(); - bun_http::HTTPThread::schedule(batch); + ); + // SAFETY: `task_ptr` was allocated above and has not been handed to anything yet + // (`task` is not used past this point); `http`'s callback context is `task_ptr`. + unsafe { S3HttpDownloadStreamingTask::schedule(task_ptr, http) }; task_ptr } diff --git a/src/runtime/webcore/s3/download_stream.rs b/src/runtime/webcore/s3/download_stream.rs index a07a56e11f1b..664642eae2cd 100644 --- a/src/runtime/webcore/s3/download_stream.rs +++ b/src/runtime/webcore/s3/download_stream.rs @@ -12,12 +12,18 @@ use bun_s3_signing::error::S3Error; use crate::webcore::s3::xml_response; use bun_threading::Mutex; +use bun_threading::thread_pool::Batch; bun_core::declare_scope!(S3, hidden); pub struct S3HttpDownloadStreamingTask { // `MaybeUninit` because `AsyncHTTP` contains non-null references, so // `mem::zeroed()` can't be used here (mirrors `S3HttpSimpleTask`). + // + // The HTTP thread's from `schedule` until the final callback hands the task + // back (`update_state` overwrites it on every callback); JS-thread code + // reaches the in-flight request through `async_http_id` instead. Enforced by + // test/internal/source-lints/s3-task-http-field.test.ts. pub(crate) http: core::mem::MaybeUninit>, /// How the HTTP thread reaches the VM to deliver chunks. pub(crate) loop_handle: bun_jsc::LoopHandle, @@ -39,9 +45,8 @@ pub struct S3HttpDownloadStreamingTask { pub(crate) concurrent_task: ConcurrentTask, pub(crate) proxy_url: Box<[u8]>, - /// Captured once on the main thread before the request is queued so the cancel - /// path can call `schedule_shutdown_by_id` without dereferencing `http` (which - /// `update_state` overwrites on the HTTP thread under `mutex`). + /// Set by `schedule` before the hand-off; what the cancel and VM-teardown + /// paths pass to `schedule_shutdown_by_id` instead of reading `http`. pub(crate) async_http_id: u32, } @@ -59,6 +64,29 @@ impl S3HttpDownloadStreamingTask { Box::new(init) } + /// Stores `http` in the task and hands the request to the HTTP thread (see + /// the `http` field for what that gives up). + /// + /// # Safety + /// `this` is a live task from `Self::new` that nothing else references yet; + /// `http`'s callback context is `this`. JS thread. + pub(crate) unsafe fn schedule(this: *mut Self, mut http: AsyncHTTP<'static>) { + http.enable_response_body_streaming(); + bun_http::http_thread::init(&Default::default()); + let mut batch = Batch::default(); + // SAFETY: fn contract; statement-scoped accesses. The `&mut AsyncHTTP` + // from `write` ends with the statement that queues its task node. + unsafe { + (*this).async_http_id = http.async_http_id; + (*this).http.write(http).schedule(&mut batch); + // Out on the HTTP thread until its final callback: the VM aborts it + // at teardown (registry) and waits for it (embedded work). + (*this).loop_handle.embedded_work_scheduled(); + } + crate::jsc_hooks::ActiveHandle::S3Download(NonNull::new(this).expect("task")).register(); + bun_http::HTTPThread::schedule(batch); + } + pub(crate) fn get_state(&self) -> State { State(self.state.load(Ordering::Acquire)) } @@ -205,8 +233,8 @@ impl S3HttpDownloadStreamingTask { // SAFETY: `async_http` points to a live AsyncHTTP owned by the HTTP thread; a // bitwise read+write copies its current state into `self.http` without running // destructors (the HTTP thread retains ownership of the source until the request - // completes). `self.http` was previously initialised in - // `client::download_stream`. + // completes). `self.http` was initialised in `Self::schedule`, and nothing reads + // it on the JS thread while the request is in flight. unsafe { core::ptr::write(self.http.as_mut_ptr(), core::ptr::read(async_http)) }; } wait_until_done @@ -349,15 +377,17 @@ impl S3HttpDownloadStreamingTask { /// # Safety /// `this` is live (registered ⇒ not yet freed by `on_response`); JS thread. pub(crate) unsafe fn stop_for_vm_teardown(this: *mut Self) { - // SAFETY: fn contract; `http` is initialised before the task is registered. + // SAFETY: fn contract. Registered ⇒ in flight ⇒ the HTTP thread may be writing + // `http` right now; only the atomic abort flag and the schedule-time id are read. unsafe { (*this).signal_store.aborted.store(true, Ordering::Relaxed); - bun_http::http_thread().schedule_shutdown((*this).http.assume_init_ref()); + bun_http::http_thread().schedule_shutdown_by_id((*this).async_http_id); } } fn release_portable(&mut self) { - // SAFETY: `http` is always initialised before the task is scheduled / dropped. + // SAFETY: `http` was initialised by `Self::schedule`, and a task is only dropped + // once the final callback has handed it back, so the HTTP thread is done with it. let http = unsafe { self.http.assume_init_mut() }; http.clear_data(); http.request_headers = Default::default(); diff --git a/src/runtime/webcore/s3/simple_request.rs b/src/runtime/webcore/s3/simple_request.rs index 30eea745852d..9501248601b9 100644 --- a/src/runtime/webcore/s3/simple_request.rs +++ b/src/runtime/webcore/s3/simple_request.rs @@ -112,9 +112,17 @@ pub struct S3HttpSimpleTask { // `http.clear_data()`, never a full destructor, and `http_callback` does a no-drop bitwise // overwrite. Wrapping in `MaybeUninit` makes both possible: write-without- // drop on assignment, and `clear_data()`-only in `Drop`. Invariant: `http` is initialised by - // `execute_simple_s3_request` before the task pointer escapes, so every later access (in - // `http_callback` / `Drop`) may `assume_init`. + // `Self::schedule` before the task pointer escapes, so `http_callback` / `Drop` may + // `assume_init`. + // + // The HTTP thread's from `schedule` until the final callback hands the task back + // (`stage_http_result` overwrites it on every callback); JS-thread code reaches the + // in-flight request through `async_http_id` instead. Enforced by + // test/internal/source-lints/s3-task-http-field.test.ts. pub(crate) http: core::mem::MaybeUninit>, + /// Set by `schedule` before the hand-off; what `stop_for_vm_teardown` passes to + /// `schedule_shutdown_by_id` instead of reading `http`. + pub(crate) async_http_id: u32, /// How the HTTP thread reaches the VM to deliver the response. pub(crate) loop_handle: bun_jsc::LoopHandle, pub(crate) sign_result: SignResult, @@ -212,6 +220,29 @@ impl S3HttpSimpleTask { bun_core::heap::into_raw(Box::new(init)) } + /// Stores `http` in the task and hands the request to the HTTP thread (see + /// the `http` field for what that gives up). + /// + /// # Safety + /// `this` is a live task from `Self::new` that nothing else references yet; + /// `http`'s callback context is `this`. JS thread. + pub(crate) unsafe fn schedule(this: *mut Self, http: AsyncHTTP<'static>) { + bun_http::http_thread::init(&Default::default()); + let mut batch = thread_pool::Batch::default(); + // SAFETY: fn contract; statement-scoped accesses. The `&mut AsyncHTTP` + // from `write` ends with the statement that queues its task node. + unsafe { + (*this).async_http_id = http.async_http_id; + (*this).http.write(http).schedule(&mut batch); + // Out on the HTTP thread until its final callback: the VM aborts it + // at teardown (registry) and waits for it (embedded work). + (*this).loop_handle.embedded_work_scheduled(); + } + crate::jsc_hooks::ActiveHandle::S3Request(core::ptr::NonNull::new(this).expect("task")) + .register(); + bun_http::HTTPThread::schedule(batch); + } + fn error_with_body(&self, error_type: ErrorType) -> JsTerminatedResult<()> { let mut code: &[u8] = b"UnknownError"; let mut message: &[u8] = b"an unexpected error has occurred"; @@ -405,7 +436,8 @@ impl S3HttpSimpleTask { // conceptually transfers here; the http-thread side must free only its outer // allocation (TrivialDeinit). // SAFETY: `async_http` is a valid live pointer for the duration of this callback; - // `self.http` was previously initialised in `execute_simple_s3_request`. + // `self.http` was initialised in `Self::schedule`, and nothing reads it on the JS + // thread while the request is in flight. unsafe { core::ptr::write(self.http.as_mut_ptr(), core::ptr::read(async_http)) }; } @@ -477,16 +509,17 @@ impl S3HttpSimpleTask { /// # Safety /// `this` is live (registered ⇒ its response has not run); JS thread. pub(crate) unsafe fn stop_for_vm_teardown(this: *mut Self) { - // SAFETY: fn contract; `http` is initialised before the task is registered. + // SAFETY: fn contract. Registered ⇒ in flight ⇒ the HTTP thread may be writing + // `http` right now; only the atomic abort flag and the schedule-time id are read. unsafe { (*this).signal_store.aborted.store(true, Ordering::Relaxed); - bun_http::http_thread().schedule_shutdown((*this).http.assume_init_ref()); + bun_http::http_thread().schedule_shutdown_by_id((*this).async_http_id); } } fn release_portable(&mut self) { - // SAFETY: `http` is always initialised before the task pointer escapes (see - // `execute_simple_s3_request`). + // SAFETY: `http` was initialised by `Self::schedule`, and a task is only dropped + // once the final callback has handed it back, so the HTTP thread is done with it. let http = unsafe { self.http.assume_init_mut() }; http.clear_data(); http.request_headers = Default::default(); @@ -631,8 +664,9 @@ pub(crate) fn execute_simple_s3_request( )); let proxy = options.proxy_url.unwrap_or(b""); let task_ptr = S3HttpSimpleTask::new(S3HttpSimpleTask { - // written below via `MaybeUninit::write` before any read. + // Both written by `S3HttpSimpleTask::schedule` below. http: core::mem::MaybeUninit::uninit(), + async_http_id: 0, sign_result: result, callback_context, callback, @@ -650,8 +684,8 @@ pub(crate) fn execute_simple_s3_request( poll_ref, signal_store: Default::default(), }); - // SAFETY: `task_ptr` is a freshly heap-allocated pointer; shared reads only until - // the scoped exclusive `http` writes below. + // SAFETY: `task_ptr` is a freshly heap-allocated pointer; shared reads only, all of + // them before the exclusive accesses through `task_ptr` below. let task = unsafe { &*task_ptr }; // SAFETY: lifetime extension — `url`, `headers_buf`, and `proxy_url` borrow from // heap-allocated fields of `*task` (sign_result.url / headers.buf / proxy_url) which the task @@ -702,20 +736,9 @@ pub(crate) fn execute_simple_s3_request( ..Default::default() }, ); - // SAFETY: `task_ptr` is still the sole pointer (the HTTP thread only sees it after - // `schedule` below); scoped exclusive write of the `http` field. - unsafe { (*task_ptr).http.write(async_http) }; - // queue http request - bun_http::http_thread::init(&Default::default()); - let mut batch = thread_pool::Batch::default(); - // SAFETY: `http` was initialised immediately above; scoped exclusive access. - unsafe { (*task_ptr).http.assume_init_mut() }.schedule(&mut batch); - // Out on the HTTP thread until its final callback: the VM aborts it at - // teardown (registry) and waits for it (embedded work). - // SAFETY: as above. - unsafe { (*task_ptr).loop_handle.embedded_work_scheduled() }; - crate::jsc_hooks::ActiveHandle::S3Request(core::ptr::NonNull::new(task_ptr).expect("task")) - .register(); - bun_http::HTTPThread::schedule(batch); + // SAFETY: `task_ptr` is still the sole pointer (the HTTP thread only sees it once + // `schedule` queues it) and `task` is not used past this point; `async_http`'s callback + // context is `task_ptr`. + unsafe { S3HttpSimpleTask::schedule(task_ptr, async_http) }; Ok(()) } diff --git a/test/internal/source-lints/s3-task-http-field.test.ts b/test/internal/source-lints/s3-task-http-field.test.ts new file mode 100644 index 000000000000..698c8d51d567 --- /dev/null +++ b/test/internal/source-lints/s3-task-http-field.test.ts @@ -0,0 +1,139 @@ +import { file } from "bun"; +import { expect, test } from "bun:test"; +import { realpathSync } from "fs"; +import path from "path"; +import { globAllSources } from "../../../scripts/glob-sources.ts"; + +// The S3 tasks (`S3HttpSimpleTask` in simple_request.rs, `S3HttpDownloadStreamingTask` +// in download_stream.rs) store their `AsyncHTTP` inline, in a `http` field. From +// the moment `schedule` hands the task to the HTTP thread until the final +// callback comes back to the JS thread, that field is the HTTP thread's: every +// progress callback bitwise-overwrites the whole struct (`stage_http_result`, +// `update_state`), and the JS thread runs concurrently with those writes. A +// JS-thread read of the field in that window (`stop_for_vm_teardown` used to do +// `schedule_shutdown(http.assume_init_ref())` to get at `http.async_http_id`) +// is a data race, even though the bytes it wants never change value. Anything +// that needs to reach the in-flight request from the JS thread goes through the +// task's `async_http_id`, captured by `schedule` before the hand-off, and +// `HttpThread::schedule_shutdown_by_id`. +// +// So the field may be touched in exactly three places per task type, all inside +// the task's own file: +// - `schedule`: the JS thread initialises it and hands it over. +// - `stage_http_result` / `update_state`: the HTTP thread's own overwrite. +// - `release_portable`: `Drop`, which only runs once the final callback has +// handed the task back. +// Any other access (a new JS-thread abort/resume path, a caller in client.rs +// or multipart.rs) fails here; route it through `async_http_id` instead. + +const root = path.resolve(import.meta.dir, "..", "..", ".."); +const S3_DIR = "src/runtime/webcore/s3/"; + +// Only scan files tracked in HEAD (a `git stash` round-trip can leave stray +// `.rs` files in the working tree; CI runs on a clean checkout). Same guard as +// dead-code-escapes.test.ts. +const tracked: Set | null = (() => { + const r = Bun.spawnSync({ + cmd: ["git", "-C", root, "ls-tree", "-r", "--name-only", "-z", "HEAD"], + stdout: "pipe", + stderr: "ignore", + }); + if (!r.success) return null; + return new Set(r.stdout.toString().split("\0").filter(Boolean)); +})(); + +// Every spelling that reaches the `AsyncHTTP` inside a `http: MaybeUninit` +// field: `x.http.assume_init_ref()`, `.assume_init_mut()`, `.assume_init()`, +// `.assume_init_read()`, `.assume_init_drop()`, `.as_ptr()`, `.as_mut_ptr()`, +// `.write(..)`. `\s*` between the tokens so a rustfmt-wrapped chain still matches; +// the `.` after `http` keeps `.http_proxy` and the `bun_http` crate path out. +const HTTP_FIELD_ACCESS = /\.\s*http\s*\.\s*(?:assume_init(?:_ref|_mut|_read|_drop)?|as_ptr|as_mut_ptr|write)\s*\(/g; + +// `fn name` item headers. fn-pointer types (`fn(..)`) have no name and do not match. +const FN_HEADER = /\bfn\s+([A-Za-z_]\w*)/g; + +// file -> the functions in it that may touch the field (see the header comment). +const ALLOWED: Record = { + [`${S3_DIR}simple_request.rs`]: ["schedule", "stage_http_result", "release_portable"], + [`${S3_DIR}download_stream.rs`]: ["schedule", "update_state", "release_portable"], +}; + +const offenders: string[] = []; +// `file::fn` for every access attributed to an ALLOWED function. +const allowedHits = new Set(); +let scanned = 0; +for (const abs of globAllSources().rust) { + if (!abs.endsWith(".rs")) continue; + const source = path.relative(root, abs).replaceAll(path.sep, "/"); + if (!source.startsWith(S3_DIR)) continue; + if (path.relative(root, realpathSync(abs)).replaceAll(path.sep, "/") !== source) continue; + if (tracked !== null && !tracked.has(source)) continue; + scanned++; + const content = await file(abs).text(); + // Strip full-line comments so prose about the field doesn't count. `[ \t]*`, + // not `\s*`: `\s` crosses newlines and would shift the reported line numbers. + const stripped = content.replace(/^[ \t]*\/\/.*$/gm, ""); + const fns = [...stripped.matchAll(FN_HEADER)].map(m => ({ index: m.index!, name: m[1] })); + for (const m of stripped.matchAll(HTTP_FIELD_ACCESS)) { + // The access belongs to the nearest `fn` header above it. + const fn = fns.findLast(f => f.index < m.index!)?.name ?? ""; + if (ALLOWED[source]?.includes(fn)) { + allowedHits.add(`${source}::${fn}`); + continue; + } + const line = stripped.slice(0, m.index).split("\n").length; + offenders.push(`${source}:${line} (in fn ${fn}): ${m[0].replace(/\s+/g, "")}`); + } +} + +function matches(snippet: string): boolean { + HTTP_FIELD_ACCESS.lastIndex = 0; + return HTTP_FIELD_ACCESS.test(snippet); +} + +test("scans the S3 task sources", () => { + // If the directory moves, the scan would otherwise pass with nothing to check. + expect(scanned).toBeGreaterThanOrEqual(Object.keys(ALLOWED).length); +}); + +test("the pattern recognizes the spellings it claims to", () => { + const banned = [ + // The teardown race this lint was written for. + "bun_http::http_thread().schedule_shutdown((*this).http.assume_init_ref());", + "let http = unsafe { self.http.assume_init_mut() };", + "unsafe { core::ptr::write(self.http.as_mut_ptr(), core::ptr::read(async_http)) };", + "let id = unsafe { (*task).http.assume_init_ref() }.async_http_id;", + "unsafe { (*task_ptr).http.write(async_http) };", + "task.http.write(bun_http::AsyncHTTP::init(", + "unsafe { task.http.assume_init_mut() }.schedule(&mut batch);", + "ptr::read((*this).http.as_ptr())", + // rustfmt-wrapped chain. + "(*this)\n .http\n .assume_init_ref()", + ]; + const allowed = [ + // The id captured at schedule time is the sanctioned JS-thread handle. + "bun_http::http_thread().schedule_shutdown_by_id((*this).async_http_id);", + "(*this).async_http_id = http.async_http_id;", + // Other things called `http`: the crate, a local, an unrelated field, the declaration. + "bun_http::http_thread::init(&Default::default());", + "http.enable_response_body_streaming();", + "http.schedule(&mut batch);", + "options.http_proxy.take()", + "pub(crate) http: core::mem::MaybeUninit>,", + ]; + expect(banned.filter(s => !matches(s))).toEqual([]); + expect(allowed.filter(matches)).toEqual([]); +}); + +test("the task's `http` field is only touched at schedule, in the HTTP-thread callback, and in Drop", () => { + expect(offenders).toEqual([]); +}); + +test("every allowed function still touches the field", () => { + // Ratchet: an entry whose function no longer accesses `http` (renamed, + // restructured) must be removed so the name cannot be reused to smuggle in a + // JS-thread access later. This also proves HTTP_FIELD_ACCESS still matches + // the real code, so the ban above cannot pass vacuously. + const expected = Object.entries(ALLOWED).flatMap(([source, fns]) => fns.map(fn => `${source}::${fn}`)); + expect([...allowedHits].sort()).toEqual(expected.sort()); +}); diff --git a/test/js/node/worker_threads/worker_threads.test.ts b/test/js/node/worker_threads/worker_threads.test.ts index f2fffb50980a..f7f54ba7650e 100644 --- a/test/js/node/worker_threads/worker_threads.test.ts +++ b/test/js/node/worker_threads/worker_threads.test.ts @@ -2567,6 +2567,103 @@ describe("VM teardown ordering", () => { expect(stdout).toBe("exit 1\n"); expect(exitCode).toBe(0); }); + + // Teardown aborts each in-flight S3 request through the request id its task + // captured when it was scheduled (while a request is in flight, the task's + // AsyncHTTP belongs to the HTTP thread). This server answers every request + // with headers and one body chunk and then stalls, so a request against it + // sits in flight on an idle socket; the only way the HTTP thread hands it back, + // and so the only way terminate() ever resolves, is that abort reaching it. + // + // Two requests per test: request ids come from a process-wide counter that + // starts at 0, so a task that never captured its id would still happen to + // abort the process's first request. The second one is what a missing id + // shows up on. + function stalledS3Server(onResponded: () => void) { + return Bun.listen<{ responded: boolean }>({ + port: 0, + hostname: "127.0.0.1", + socket: { + open(socket) { + socket.data = { responded: false }; + }, + data(socket) { + if (socket.data.responded) return; + socket.data.responded = true; + socket.write( + "HTTP/1.1 200 OK\r\nTransfer-Encoding: chunked\r\nContent-Type: application/octet-stream\r\n\r\n" + + "400\r\n" + + Buffer.alloc(0x400, "x").toString() + + "\r\n", + ); + onResponded(); + }, + error() {}, + }, + }); + } + const S3_WORKER_PRELUDE = + 'const { workerData, parentPort } = require("worker_threads");' + + 'const s3 = new Bun.S3Client({ accessKeyId: "k", secretAccessKey: "s", bucket: "b", endpoint: workerData.endpoint });'; + // S3 requests go through an inherited HTTP_PROXY without consulting NO_PROXY, + // which would route them past the stub server. + const envWithoutProxy = { ...bunEnv, HTTP_PROXY: undefined, http_proxy: undefined }; + + test("terminating a worker with S3 streaming downloads in flight aborts them", async () => { + using server = stalledS3Server(() => {}); + await using proc = Bun.spawn({ + cmd: [ + bunExe(), + "-e", + `const { Worker } = require("worker_threads"); + const w = new Worker( + ${JSON.stringify(S3_WORKER_PRELUDE)} + + // Each download has delivered its first chunk once this resolves: both are in flight. + 'Promise.all([1, 2].map(() => s3.file("key").stream().getReader().read())).then(() => parentPort.postMessage("streaming"));', + { eval: true, workerData: { endpoint: "http://127.0.0.1:${server.port}" } }); + w.once("message", async () => console.log("exit", await w.terminate()));`, + ], + env: envWithoutProxy, + stdout: "pipe", + stderr: "inherit", + }); + const [stdout, exitCode] = await Promise.all([proc.stdout.text(), proc.exited]); + expect(stdout).toBe("exit 1\n"); + expect(exitCode).toBe(0); + }); + + test("terminating a worker with buffered S3 downloads in flight aborts them", async () => { + // A buffered download (.text()) gives JS nothing to observe until it completes, + // so the server reports when both requests are in flight and the parent then + // tells the child, over stdin, to terminate the worker. + const bothInFlight = Promise.withResolvers(); + let responded = 0; + using server = stalledS3Server(() => { + if (++responded === 2) bothInFlight.resolve(); + }); + await using proc = Bun.spawn({ + cmd: [ + bunExe(), + "-e", + `const { Worker } = require("worker_threads"); + const w = new Worker( + ${JSON.stringify(S3_WORKER_PRELUDE)} + + 'for (const key of ["a", "b"]) s3.file(key).text().catch(() => {});', + { eval: true, workerData: { endpoint: "http://127.0.0.1:${server.port}" } }); + process.stdin.once("data", async () => console.log("exit", await w.terminate()));`, + ], + env: envWithoutProxy, + stdin: "pipe", + stdout: "pipe", + stderr: "inherit", + }); + await bothInFlight.promise; + proc.stdin.write("terminate\n"); + await proc.stdin.end(); + const [stdout, exitCode] = await Promise.all([proc.stdout.text(), proc.exited]); + expect(stdout).toBe("exit 1\n"); + expect(exitCode).toBe(0); + }); }); // A native completion on the worker's own loop (here: a dns lookup finishing)