diff --git a/src/http/Signals.rs b/src/http/Signals.rs index bf744ebd0cd8..61bacfac5fbf 100644 --- a/src/http/Signals.rs +++ b/src/http/Signals.rs @@ -102,7 +102,10 @@ impl Default for Store { } impl Store { - pub fn to(&mut self) -> Signals { + /// The flags' addresses, for the HTTP thread. `&self` on purpose: the owner keeps storing + /// into the flags (abort) while the HTTP thread holds these, so they must not descend from + /// an exclusive borrow of the store. + pub fn to(&self) -> Signals { Signals { header_progress: Some(NonNull::from(&self.header_progress)), response_body_streaming: Some(NonNull::from(&self.response_body_streaming)), @@ -112,7 +115,7 @@ impl Store { } } - pub fn to_with_backpressure(&mut self) -> Signals { + pub fn to_with_backpressure(&self) -> Signals { Signals { body_receive_mode: Some(NonNull::from(&self.body_receive_mode)), ..self.to() diff --git a/src/runtime/webcore/s3/simple_request.rs b/src/runtime/webcore/s3/simple_request.rs index 9501248601b9..ee319d97081f 100644 --- a/src/runtime/webcore/s3/simple_request.rs +++ b/src/runtime/webcore/s3/simple_request.rs @@ -413,40 +413,48 @@ impl S3HttpSimpleTask { Ok(()) } - fn stage_http_result( - &mut self, + /// Stores one result callback's payload on the task (HTTP thread). Takes the task as a + /// pointer because `stop_for_vm_teardown` may run on the JS thread at any point during + /// the call (test/internal/source-lints/s3-task-http-field.test.ts). + /// + /// # Safety + /// `this` is the live task this callback was registered with, and `stop_for_vm_teardown` + /// is the only thing that may touch it concurrently; `async_http` is the HTTP thread's + /// initialised copy of the request, live for the duration of the call. + unsafe fn stage_http_result( + this: *mut Self, async_http: *mut AsyncHTTP<'static>, mut result: HTTPClientResult<'_>, ) { - let previous_metadata = self.result.metadata.take(); - result.body_into(&mut self.response_buffer.list); - // SAFETY: `result.body` (the only borrowed field) points at `self.response_buffer`, - // which lives for the task's lifetime — extending to `'static` here is sound for - // self-reference. - self.result = unsafe { result.detach_lifetime() }; - if self.result.metadata.is_none() { - self.result.metadata = previous_metadata; + // SAFETY: fn contract. Every reference formed below covers one of `result`, + // `response_buffer` and `http` for one statement; `stop_for_vm_teardown` touches only + // `signal_store` and `async_http_id`. `detach_lifetime`: the stored result's `body` is + // never read; its bytes were just moved into `response_buffer`. + unsafe { + let previous_metadata = (*this).result.metadata.take(); + result.body_into(&mut (*this).response_buffer.list); + (*this).result = result.detach_lifetime(); + if (*this).result.metadata.is_none() { + (*this).result.metadata = previous_metadata; + } + // `AsyncHTTP` transitively owns Drop types (`HTTPClient`, header `EntryList`s), so + // a plain `=` here would (a) drop the old `http`, freeing heap buffers that + // `*async_http` (a bitwise clone created by the HTTP thread) still aliases, and + // (b) leave the http-thread side to drop them again → double-free. We instead + // write through `MaybeUninit` to suppress the LHS drop, doing a bitwise struct + // overwrite with no destructor on either side. Ownership of the inner heap data + // conceptually transfers here; the http-thread side must free only its outer + // allocation (TrivialDeinit). + core::ptr::write((*this).http.as_mut_ptr(), core::ptr::read(async_http)); } - // `AsyncHTTP` transitively owns Drop types (`HTTPClient`, header - // `EntryList`s), so a plain `=` here would (a) drop the old `self.http`, freeing heap - // buffers that `*async_http` (a bitwise clone created by the HTTP thread) still - // aliases, and (b) leave the http-thread side to drop them again → double-free. We - // instead write through `MaybeUninit` to suppress the LHS drop, doing a bitwise struct - // overwrite with no destructor on either side. Ownership of the inner heap data - // 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 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)) }; } /// this is the AsyncHTTP callback and is always called from the HTTPThread /// /// # Safety - /// `this` must be a live heap pointer produced by `S3HttpSimpleTask::new` and exclusively - /// owned by the HTTP thread for the duration of this call. `async_http` must be a valid - /// pointer to an initialised `AsyncHTTP` for the duration of this call. + /// `this` must be the live task from `S3HttpSimpleTask::new` this callback was registered + /// with, which only the JS thread's `stop_for_vm_teardown` may touch concurrently; + /// `async_http` must point to an initialised `AsyncHTTP` for the duration of this call. // // `HTTPClientResultCallback` entrypoint: invoked by the HTTP thread with the raw task and // request pointers it captured at schedule time, both non-null by construction. @@ -457,13 +465,13 @@ impl S3HttpSimpleTask { result: HTTPClientResult<'_>, ) { let is_done = !result.has_more; - // SAFETY: `this` was produced by `S3HttpSimpleTask::new` and is exclusively owned - // by the HTTP thread until the handoff below; this borrow is scoped to the call. - unsafe { (*this).stage_http_result(async_http, result) }; + // SAFETY: fn contract, which is `stage_http_result`'s contract. + unsafe { Self::stage_http_result(this, async_http, result) }; if is_done { - // SAFETY: same exclusivity as above; the queue takes ownership of the inline - // `concurrent_task` field's `next` link. The VM waits for its S3 requests - // (embedded work) before closing its handle: always queued. + // SAFETY: fn contract; `stop_for_vm_teardown` touches neither field used here. + // The queue takes ownership of the inline `concurrent_task` field's `next` link. + // The VM waits for its S3 requests (embedded work) before closing its handle: + // always queued. unsafe { let handle = (*this).loop_handle.clone(); let queued = core::ptr::NonNull::from( diff --git a/test/internal/source-lints/s3-task-http-field.test.ts b/test/internal/source-lints/s3-task-http-field.test.ts index 698c8d51d567..0612fa189c64 100644 --- a/test/internal/source-lints/s3-task-http-field.test.ts +++ b/test/internal/source-lints/s3-task-http-field.test.ts @@ -25,6 +25,24 @@ import { globAllSources } from "../../../scripts/glob-sources.ts"; // 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. +// +// The same window has a second rule, checked below for `S3HttpSimpleTask`: the +// functions that run on the task while it is out (IN_FLIGHT_FNS) take it as a +// raw pointer and borrow one field per statement, never as `&mut self`. While +// the HTTP thread is inside one of them, the JS thread's `stop_for_vm_teardown` +// (VM teardown, and the sweep between files under `bun test`) may store into +// the task's `signal_store`; a `&mut self` argument is a protected exclusive +// borrow of the whole task, atomics included, for the duration of the call, so +// that store is undefined behaviour under Tree Borrows (the model `bun run +// rust:miri` uses: "protected tags must never be Disabled") and Stacked +// Borrows alike, and rustc passes the same claim to LLVM as `noalias`. +// `http_callback` additionally returns only after the post that lets the JS +// thread free the task. The simple task's other methods (`error_with_body`, +// `fail_if_contains_error`, `release_portable`, `Drop`) run from `on_response`, +// after the hand-back, and keep their receivers, hence a list rather than a +// ban on every receiver in the file. The streaming task is not listed: it has +// no after-the-hand-back phase (chunks are delivered while more arrive), so the +// rule there is every method of the type, which is a different check. const root = path.resolve(import.meta.dir, "..", "..", ".."); const S3_DIR = "src/runtime/webcore/s3/"; @@ -58,9 +76,48 @@ const ALLOWED: Record = { [`${S3_DIR}download_stream.rs`]: ["schedule", "update_state", "release_portable"], }; +// file -> the functions that run on the task while it is out on the HTTP thread +// (see the header comment); each must take the task as a raw pointer. +const IN_FLIGHT_FNS: Record = { + [`${S3_DIR}simple_request.rs`]: ["http_callback", "stage_http_result", "release_at_shutdown", "stop_for_vm_teardown"], +}; + +// The first parameter of `fn (..)`, up to the first `,` or `)` (a trailing +// `()` unit type included), across rustfmt's one-parameter-per-line wrapping. +// `(?!\w)` keeps `fn foo` from matching `fn foo_bar`. +function firstParamPattern(fn: string): RegExp { + return new RegExp(String.raw`\bfn\s+${fn}(?!\w)\s*(?:<[^>]*>)?\s*\(\s*([^,()]*(?:\(\))?)`, "g"); +} + +// `release_at_shutdown` receives the type-erased `*mut ()`; the others `*mut Self`. +const RAW_TASK_PARAM = /^this\s*:\s*\*\s*mut\b/; + +interface Declaration { + fn: string; + line: number; + firstParam: string; +} + +function inFlightDeclarations(stripped: string, fns: readonly string[]): Declaration[] { + const found: Declaration[] = []; + for (const fn of fns) { + for (const m of stripped.matchAll(firstParamPattern(fn))) { + found.push({ + fn, + line: stripped.slice(0, m.index).split("\n").length, + firstParam: m[1].replace(/\s+/g, " ").trim(), + }); + } + } + return found; +} + const offenders: string[] = []; // `file::fn` for every access attributed to an ALLOWED function. const allowedHits = new Set(); +// `file::fn` for every IN_FLIGHT_FNS declaration found, and the ones not taking a pointer. +const inFlightDeclared: string[] = []; +const receiverOffenders: string[] = []; let scanned = 0; for (const abs of globAllSources().rust) { if (!abs.endsWith(".rs")) continue; @@ -84,6 +141,12 @@ for (const abs of globAllSources().rust) { const line = stripped.slice(0, m.index).split("\n").length; offenders.push(`${source}:${line} (in fn ${fn}): ${m[0].replace(/\s+/g, "")}`); } + for (const d of inFlightDeclarations(stripped, IN_FLIGHT_FNS[source] ?? [])) { + inFlightDeclared.push(`${source}::${d.fn}`); + if (!RAW_TASK_PARAM.test(d.firstParam)) { + receiverOffenders.push(`${source}:${d.line}: fn ${d.fn}(${d.firstParam}, ..)`); + } + } } function matches(snippet: string): boolean { @@ -137,3 +200,40 @@ test("every allowed function still touches the field", () => { const expected = Object.entries(ALLOWED).flatMap(([source, fns]) => fns.map(fn => `${source}::${fn}`)); expect([...allowedHits].sort()).toEqual(expected.sort()); }); + +test("the receiver check reads the first parameter out of the spellings it claims to", () => { + const parsed = inFlightDeclarations( + [ + // `stage_http_result` as it was, and as it is. + "fn stage_http_result(\n &mut self,\n async_http: *mut AsyncHTTP<'static>,\n ) {", + "unsafe fn stage_http_result(\n this: *mut Self,\n async_http: *mut AsyncHTTP<'static>,\n ) {", + // The other spellings of a task reference. + "pub(crate) fn http_callback(&self, async_http: *mut AsyncHTTP<'static>) {", + "pub(crate) unsafe fn release_at_shutdown(self: &mut Self) {", + "pub(crate) unsafe fn stop_for_vm_teardown<'a>(this: &'a mut Self) {", + // Type-erased is still a raw pointer. + "pub(crate) unsafe fn release_at_shutdown(this: *mut ()) {", + // Not the functions in question. + "fn stage_http_result_for_tests(&mut self) {}", + ].join("\n"), + IN_FLIGHT_FNS[`${S3_DIR}simple_request.rs`], + ); + expect(parsed.map(d => [d.fn, d.firstParam, RAW_TASK_PARAM.test(d.firstParam)])).toEqual([ + ["http_callback", "&self", false], + ["stage_http_result", "&mut self", false], + ["stage_http_result", "this: *mut Self", true], + ["release_at_shutdown", "self: &mut Self", false], + ["release_at_shutdown", "this: *mut ()", true], + ["stop_for_vm_teardown", "this: &'a mut Self", false], + ]); +}); + +test("every in-flight function is still declared under its listed name", () => { + // A rename would otherwise silently drop the function out of the check below. + const expected = Object.entries(IN_FLIGHT_FNS).flatMap(([source, fns]) => fns.map(fn => `${source}::${fn}`)); + expect(inFlightDeclared.sort()).toEqual(expected.sort()); +}); + +test("functions that run on an in-flight S3HttpSimpleTask take it as a raw pointer", () => { + expect(receiverOffenders).toEqual([]); +});