event_loop: make DeferredTaskQueue::run tolerate re-entrant map mutation

src/event_loop/DeferredTaskQueue.rs

@@ -25,6 +25,15 @@
 //!
 //! The DeferredTaskQueue is drained after the microtask queue, but before other tasks are executed. This avoids re-entrancy
 //! issues with the event loop.
+//!
+//! `run()` must tolerate re-entrant mutation of the map: a callback can reach user JS, which can
+//! `post_task` / `unregister_task` any entry (for example, `H2FrameParser`'s auto-flush unregisters
+//! itself and then returns `true`, and before the cork slot became self-scoped one parser's flush
+//! could unregister another's). Each pass is capped at the number of entries present when it
+//! started; that cap is a livelock bound, not a strict "no new entries this pass" frontier, because
+//! a swap-remove can pull a freshly appended entry forward. Every callback flushes its own object's
+//! buffer to its own fd or socket, so relative order within a pass has no correctness effect, and
+//! anything a pass skips runs on the next one.
 
 use core::ffi::c_void;
 use core::ptr::NonNull;
@@ -57,39 +66,36 @@ impl DeferredTaskQueue {
     }
 
     pub fn unregister_task(&mut self, ctx: Option<NonNull<c_void>>) -> bool {
-        // Order is irrelevant for this map's contract (see file doc — "order
-        // may not particularly matter"), so plain `remove().is_some()` works.
-        self.map.remove(&ctx).is_some()
+        // Order is irrelevant for this map's contract (see file doc), so
+        // swap-remove is fine and O(1).
+        self.map.swap_remove(&ctx)
     }
 
     pub fn run(&mut self) {
-        // The current `ArrayHashMap` exposes `keys()/values()` slices and
-        // `swap_remove(&K)` (hash lookup) but not `swap_remove_at` (O(1) by
-        // index). Keys here are `Copy` pointers, so copy the key out and
-        // remove by key — semantically identical (keys are unique), just an
-        // extra hash per removal.
-        // PERF: swap_remove(&K) re-hashes; restore swap_remove_at when
-        // bun_collections::ArrayHashMap grows it.
+        // Callbacks may re-entrantly mutate `self.map` (see the re-entrancy
+        // note in the file doc), so re-read `len()` every iteration and
+        // re-check slot `i` after each callback. `remaining` is a livelock bound.
         let mut i: usize = 0;
-        let mut last = self.map.len();
-        while i < last {
+        let mut remaining = self.map.len();
+        while remaining > 0 && i < self.map.len() {
+            remaining -= 1;
             let key = self.map.keys()[i];
             let Some(nn) = key else {
-                self.map.swap_remove(&key);
-                last = self.map.len();
+                self.map.swap_remove_at(i);
                 continue;
             };
 
-            // Copy the fn ptr out before calling (borrowck).
             let task = self.map.values()[i];
             // SAFETY: `nn` is the live `*mut T` registered by the caller; the
             // callback contract (`HasAutoFlusher::on_auto_flush`) is that
             // `task` may be invoked with exactly that pointer until it returns
             // `false` or is explicitly unregistered.
-            if !unsafe { task(nn.as_ptr()) } {
+            let keep = unsafe { task(nn.as_ptr()) };
+            if !keep {
+                // `key` may already be gone (callback unregistered it) or may
+                // have moved; remove by key, not index.
                 self.map.swap_remove(&key);
-                last = self.map.len();
-            } else {
+            } else if self.map.keys().get(i) == Some(&key) {
                 i += 1;
             }
         }

test/js/node/http2/node-http2-deferred-task-queue.fixture.js

@@ -0,0 +1,88 @@
+// DeferredTaskQueue::run iterates the auto-flusher map and invokes each
+// entry's callback. H2FrameParser::on_auto_flush calls flush -> uncork ->
+// unregister_auto_flush, which removes its own entry from the map while
+// run() is iterating, and then returns true (keep). The loop must not
+// assume the map length is unchanged after the callback returns.
+//
+// This fixture registers two auto-flushers in the same microtask so they
+// share one DeferredTaskQueue::run pass: an H2FrameParser (via
+// client.request, which corks) and an HTTPServerWritable (via a small
+// controller.write on a Bun.serve direct stream, which buffers and
+// registers). The H2 entry is first in the insertion-ordered map; when its
+// callback removes it and returns true, run() must re-read the map length
+// before indexing the next slot.
+
+const http2 = require("node:http2");
+const { once } = require("node:events");
+
+async function main() {
+  const h2srv = http2.createServer();
+  h2srv.on("stream", s => {
+    s.respond({ ":status": 200 });
+    s.end("ok");
+  });
+  h2srv.on("error", () => {});
+  h2srv.listen(0, "127.0.0.1");
+  await once(h2srv, "listening");
+  const h2port = h2srv.address().port;
+
+  const client = http2.connect(`http://127.0.0.1:${h2port}`);
+  client.on("error", () => {});
+  await once(client, "connect");
+
+  let resolveController;
+  const controllerPromise = new Promise(r => (resolveController = r));
+  let resolveDone;
+  const donePromise = new Promise(r => (resolveDone = r));
+
+  const app = Bun.serve({
+    port: 0,
+    fetch() {
+      return new Response(
+        new ReadableStream({
+          type: "direct",
+          pull(controller) {
+            resolveController(controller);
+            return donePromise;
+          },
+        }),
+      );
+    },
+  });
+
+  const fetchPromise = fetch(`http://127.0.0.1:${app.port}/`).then(r => r.text());
+  const controller = await controllerPromise;
+
+  // Move to a fresh event-loop task so the registrations below land in a
+  // microtask whose drain runs DeferredTaskQueue::run directly after.
+  await new Promise(r => setImmediate(r));
+
+  queueMicrotask(() => {
+    // H2FrameParser::write -> cork -> register_auto_flush (map[0]).
+    const req = client.request({ ":path": "/" });
+    req.on("error", () => {});
+    req.resume();
+    // HTTPServerWritable::write (< highWaterMark) -> register_auto_flusher (map[1]).
+    controller.write("x");
+  });
+  await new Promise(r => setImmediate(r));
+
+  controller.write("y");
+  controller.close();
+  resolveDone();
+  const body = await fetchPromise;
+  if (body !== "xy") throw new Error(`unexpected body: ${JSON.stringify(body)}`);
+
+  app.stop(true);
+  client.close();
+  await new Promise(r => h2srv.close(r));
+  console.log("OK");
+}
+
+main().then(
+  () => process.exit(0),
+  e => {
+    console.error(e);
+    process.exit(1);
+  },
+);

test/js/node/http2/node-http2-streams-rehash.test.ts

@@ -1,7 +1,12 @@
 import { expect, test } from "bun:test";
-import { bunEnv, bunExe } from "harness";
+import { bunEnv, bunExe, isASAN, isDebug } from "harness";
 import path from "node:path";
 
+// The spawned fixtures finish in well under a second in release but take
+// several seconds under the ASAN-instrumented, unoptimized debug build, so
+// the default 5s per-test budget kills them before they can exit.
+const ASAN_MULTIPLIER = isDebug ? 10 : isASAN ? 3 : 1;
+
 // H2FrameParser stored Stream by value in a HashMap. Any *Stream obtained
 // from getPtr/value_ptr/valueIterator pointed into the map's backing storage
 // and dangled if a re-entrant JS callback inserted a new stream and triggered
@@ -20,8 +25,10 @@ test("session.request() from a stream 'timeout' listener during forEachStream do
   expect({ stdout: stdout.trim(), exitCode, stderr }).toMatchObject({ stdout: "OK", exitCode: 0 });
 });
 
-test("http2 client request() does not hold *Stream across user-controlled options getters", async () => {
-  const script = /* js */ `
+test(
+  "http2 client request() does not hold *Stream across user-controlled options getters",
+  async () => {
+    const script = /* js */ `
     const http2 = require("node:http2");
 
     const server = http2.createServer();
@@ -86,15 +93,17 @@ test("http2 client request() does not hold *Stream across user-controlled option
     });
   `;
 
-  await using proc = Bun.spawn({
-    cmd: [bunExe(), "-e", script],
-    env: bunEnv,
-    stdout: "pipe",
-    stderr: "pipe",
-  });
-  const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]);
-  expect({ stdout: stdout.trim(), exitCode, stderr }).toMatchObject({ stdout: "done", exitCode: 0 });
-});
+    await using proc = Bun.spawn({
+      cmd: [bunExe(), "-e", script],
+      env: bunEnv,
+      stdout: "pipe",
+      stderr: "pipe",
+    });
+    const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]);
+    expect({ stdout: stdout.trim(), exitCode, stderr }).toMatchObject({ stdout: "done", exitCode: 0 });
+  },
+  10_000 * ASAN_MULTIPLIER,
+);
 
 test("http2 client write callback that opens new streams during flushQueue does not UAF", async () => {
   await using proc = Bun.spawn({
@@ -106,3 +115,23 @@ test("http2 client write callback that opens new streams during flushQueue does
   const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]);
   expect({ stdout: stdout.trim(), exitCode, stderr }).toMatchObject({ stdout: "ok", exitCode: 0 });
 });
+
+// H2FrameParser::on_auto_flush calls flush -> uncork -> unregister_auto_flush,
+// removing its own entry from the DeferredTaskQueue mid-iteration and then
+// returning true. With a second auto-flusher (an HTTPServerWritable small
+// write) sitting after it in the map, DeferredTaskQueue::run would index past
+// the new length and panic.
+test(
+  "DeferredTaskQueue::run tolerates an on_auto_flush callback that unregisters itself and returns true",
+  async () => {
+    await using proc = Bun.spawn({
+      cmd: [bunExe(), path.join(import.meta.dir, "node-http2-deferred-task-queue.fixture.js")],
+      env: bunEnv,
+      stdout: "pipe",
+      stderr: "pipe",
+    });
+    const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]);
+    expect({ stdout: stdout.trim(), exitCode, stderr }).toMatchObject({ stdout: "OK", exitCode: 0 });
+  },
+  10_000 * ASAN_MULTIPLIER,
+);