s3: stage simple-request results through the task pointer instead of a &mut self receiver

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()

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(

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<string, readonly string[]> = {
   [`${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<string, readonly string[]> = {
+  [`${S3_DIR}simple_request.rs`]: ["http_callback", "stage_http_result", "release_at_shutdown", "stop_for_vm_teardown"],
+};
+
+// The first parameter of `fn <name>(..)`, 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<string>();
+// `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([]);
+});