event_loop: yield mid-tick thread-pool re-drain to due timers and pending setImmediate

src/jsc/event_loop.rs

@@ -191,6 +191,10 @@ unsafe extern "Rust" {
     /// `WTFTimer::run` — `timer` is an erased `*mut bun_runtime::timer::WTFTimer`.
     /// Defined in `bun_runtime::dispatch`. Link-time resolved.
     fn __bun_run_wtf_timer(timer: *mut (), vm: *mut VirtualMachine);
+    /// `timer::All::has_due_regular_timer` — peek the JS timer heap and
+    /// report whether its soonest entry is already due. Defined in
+    /// `bun_runtime::dispatch`. Link-time resolved.
+    safe fn __bun_has_due_timer() -> bool;
     /// Tag-specific shutdown release for a queued-but-never-run task. Called
     /// from `release_queued_tasks_for_shutdown` (after `shutdown_for_exit`,
     /// before `destructOnExit`) for every entry left in `self.tasks`.
@@ -455,6 +459,29 @@ impl EventLoop {
         let _ = self.tick_concurrent_with_count();
     }
 
+    /// Re-drain the concurrent queue mid-tick only when nothing is waiting on
+    /// `tick()` to return. libuv delivers thread-pool completions once per
+    /// poll and runs the timers/check phases between polls, so Node interleaves
+    /// due `setInterval`/`setTimeout` (and `setImmediate`) with a chain of
+    /// `fs.read` callbacks that each submit the next read. The unconditional
+    /// `tick_concurrent()` at the start of `tick()` is that once-per-iteration
+    /// batch boundary; re-draining here is a throughput nicety that must yield
+    /// when a timer is due or an immediate is pending.
+    ///
+    /// The yield probe is guarded on `concurrent_tasks` being non-empty so the
+    /// clock read in `__bun_has_due_timer` is only paid when there is actually
+    /// something to re-drain (an HTTP server always has `DateHeaderTimer`
+    /// armed, so an unguarded probe would add a `clock_gettime` to every
+    /// inner-loop iteration of `tick()`).
+    fn tick_concurrent_unless_due(&mut self) {
+        if !self.concurrent_tasks.is_empty()
+            && (!self.immediate_tasks.is_empty() || __bun_has_due_timer())
+        {
+            return;
+        }
+        self.tick_concurrent();
+    }
+
     /// Check whether refConcurrently has been called but the change has not yet been applied to the
     /// underlying event loop's `active` counter
     pub fn has_pending_refs(&self) -> bool {
@@ -620,7 +647,7 @@ impl EventLoop {
 
         loop {
             while self.tick_with_count(ctx) > 0 {
-                self.tick_concurrent();
+                self.tick_concurrent_unless_due();
                 self.global_ref().handle_rejected_promises();
             }
             if self
@@ -631,15 +658,15 @@ impl EventLoop {
                 self.entered_event_loop_count -= 1;
                 return;
             }
-            self.tick_concurrent();
+            self.tick_concurrent_unless_due();
             if self.tasks.readable_length() > 0 {
                 continue;
             }
             break;
         }
 
         while self.tick_with_count(ctx) > 0 {
-            self.tick_concurrent();
+            self.tick_concurrent_unless_due();
         }
 
         self.global_ref().handle_rejected_promises();

src/runtime/dispatch.rs

@@ -878,6 +878,23 @@ unsafe fn __bun_cancel_pending_immediate(
     }
 }
 
+/// `__bun_has_due_timer` body — declared `extern "Rust"` in
+/// `bun_jsc::event_loop`. `EventLoop::tick()` polls this to decide whether to
+/// keep re-draining `concurrent_tasks` mid-tick or yield so `auto_tick*` can
+/// run the timers phase (Node interleaves due timers between thread-pool
+/// completion batches; without this a self-feeding `fs.read` chain starves
+/// `setInterval`).
+#[unsafe(no_mangle)]
+fn __bun_has_due_timer() -> bool {
+    let all = crate::jsc_hooks::timer_all();
+    if all.is_null() {
+        return false;
+    }
+    // SAFETY: `all` is the live per-thread `All`; single JS thread, and
+    // `has_due_regular_timer` only peeks (no heap mutation, no JS re-entry).
+    unsafe { (*all).has_due_regular_timer() }
+}
+
 /// `__bun_run_wtf_timer` body — cast the low-tier erased `*mut ()` to the real
 /// `crate::timer::WTFTimer` and fire it.
 ///

src/runtime/timer/mod.rs

@@ -1013,6 +1013,26 @@ impl All {
         }
     }
 
+    /// Cheap "is the soonest JS timer already due" probe for
+    /// `EventLoop::tick()`'s mid-tick yield check. Reads the clock only when
+    /// the regular heap is non-empty. WTF timers are ignored: taking the
+    /// `wtf_timers` lock on every inner-loop iteration of `tick()` is not
+    /// worth it, and the `drain_due_wtf_timers` call at the next
+    /// `get_timeout`/`drain_timers` is never more than one loop iteration
+    /// away once `tick()` returns.
+    pub(crate) fn has_due_regular_timer(&self) -> bool {
+        let Some(timer) = self.timers.peek() else {
+            return false;
+        };
+        // SAFETY: `peek` returns a live heap node.
+        let next = unsafe { &(*timer).next };
+        let next = Timespec {
+            sec: next.sec,
+            nsec: next.nsec,
+        };
+        !next.greater(&Timespec::now(TimespecMockMode::ForceRealTime))
+    }
+
     /// Pop the next due timer. `now` is filled lazily on first call so we
     /// don't pay for `clock_gettime` when the heap is empty.
     fn next(&mut self, has_set_now: &mut bool, now: &mut Timespec) -> Option<*mut EventLoopTimer> {

test/js/node/timers/node-timers.test.ts

@@ -245,3 +245,83 @@ describe.each(["with", "without"])("setImmediate %s timers running", mode => {
 it("should defer microtasks when an exception is thrown in an immediate", async () => {
   expect(await bunRun(["run", path.join(import.meta.dir, "timers-immediate-exception-fixture.js")])).toSpawn();
 });
+
+test("chained fs.read callbacks yield to due timers", async () => {
+  // Node/libuv delivers thread-pool completions once per poll and runs the
+  // timers phase between polls, so a chain of fs.read callbacks that each
+  // submit the next read interleaves with an overdue setInterval. Bun used to
+  // keep re-draining thread-pool completions inside one EventLoop::tick(),
+  // which meant the chain ran to completion before any due timer fired.
+  const script = `
+    const fs = require('fs');
+    const fd = fs.openSync(process.execPath, 'r');
+    let order = '';
+    setInterval(() => { order += 'T'; }, 1).unref();
+    setTimeout(() => {
+      (function go(i) {
+        if (i >= 20) {
+          const first = order.indexOf('R');
+          const last = order.lastIndexOf('R');
+          const between = order.slice(first, last + 1).split('T').length - 1;
+          console.log(JSON.stringify({ order, between }));
+          process.exit(0);
+        }
+        fs.read(fd, Buffer.alloc(1), 0, 1, 0, () => {
+          order += 'R';
+          go(i + 1);
+        });
+        // Spin so the 1ms interval is overdue and the thread-pool read has
+        // landed before the event loop decides whether to re-drain mid-tick.
+        // The spin is wall-clock, not a wait-for-condition: it constructs the
+        // state the yield check must observe.
+        const s = Date.now(); while (Date.now() - s < 2);
+      })(0);
+    }, 5);
+  `;
+  await using proc = Bun.spawn({
+    cmd: [bunExe(), "-e", script],
+    env: bunEnv,
+    stderr: "pipe",
+  });
+  const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]);
+  expect(stderr).toBe("");
+  const out = JSON.parse(stdout.trim());
+  // Node fires ~N-1 intervals between N reads with this shape; the fix only
+  // needs to guarantee the chain yields at all, so assert at least one.
+  expect({ between: out.between > 0, order: out.order }).toEqual({ between: true, order: out.order });
+  expect(exitCode).toBe(0);
+});
+
+test("chained fs.read callbacks yield to pending setImmediate", async () => {
+  // Same mid-tick re-drain: a setImmediate queued from the first read
+  // callback must run before the rest of the chain, matching Node's
+  // poll/check alternation.
+  const script = `
+    const fs = require('fs');
+    const fd = fs.openSync(process.execPath, 'r');
+    let order = '';
+    setTimeout(() => {
+      (function go(i) {
+        if (i >= 20) { console.log(JSON.stringify({ order })); process.exit(0); }
+        fs.read(fd, Buffer.alloc(1), 0, 1, 0, () => {
+          order += 'R';
+          if (i === 0) setImmediate(() => { order += 'I'; });
+          go(i + 1);
+        });
+        const s = Date.now(); while (Date.now() - s < 2);
+      })(0);
+    }, 1);
+  `;
+  await using proc = Bun.spawn({
+    cmd: [bunExe(), "-e", script],
+    env: bunEnv,
+    stderr: "pipe",
+  });
+  const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]);
+  expect(stderr).toBe("");
+  const out = JSON.parse(stdout.trim());
+  // Node: "RIRRRR...". Bun before the fix: "RRRR...RI" (or "RRRR...R" with
+  // the immediate dropped by process.exit).
+  expect(out.order.slice(0, 2)).toBe("RI");
+  expect(exitCode).toBe(0);
+});