diff --git a/src/runtime/cli/repl.rs b/src/runtime/cli/repl.rs index 9e61285f8ef9..6b1c2ca21ddf 100644 --- a/src/runtime/cli/repl.rs +++ b/src/runtime/cli/repl.rs @@ -1081,6 +1081,25 @@ impl<'a> Repl<'a> { self.stdin_buf_start += 1; return Some(b); } + + // Refill buffer. + // + // If a VM is attached, pump the JS event loop while waiting for stdin + // so timers, IPC messages, setImmediate, and other async callbacks + // actually run between keystrokes — otherwise the REPL blocks forever + // in read() and child processes (including `bun repl` spawned with + // IPC from a parent) never receive `process.on("message", ...)` + // payloads. See #30559. + if self.vm.is_some() { + let n = self.wait_for_stdin_readable()?; + if n == 0 { + return None; + } + self.stdin_buf_start = 1; + self.stdin_buf_end = n; + return Some(self.stdin_buf[0]); + } + // Refill buffer (stdio fd: `File::Drop` is a no-op, so this is safe to // re-create on every call). let stdin = sys::File::stdin(); @@ -1096,6 +1115,211 @@ impl<'a> Repl<'a> { Some(self.stdin_buf[0]) } + /// Wait until stdin has data, pumping the JS event loop in the meantime + /// so timers, IPC messages, setImmediate, etc. fire while the user is + /// thinking. Returns the number of bytes read, or `None` on EOF / fatal + /// error. + fn wait_for_stdin_readable(&mut self) -> Option { + use crate::jsc_hooks::runtime_state; + use crate::timer; + let vm = self.vm?; + let vm_ptr = core::ptr::from_ref(vm).cast_mut(); + // SAFETY: `vm` was set by `run_with_vm` and is stable for the lifetime + // of the REPL loop. + let event_loop = vm_mut(vm).event_loop_mut(); + // SAFETY: `event_loop` is the live per-thread event loop. + let loop_ = unsafe { (*core::ptr::from_mut(event_loop)).usockets_loop() }; + + loop { + // Drain pending JS work (concurrent tasks, microtasks, + // setImmediate, fully-elapsed timers) and any ready uSockets I/O. + // `vm.tick()` loops internally until the task queue is empty. + // `tick_immediate_tasks` processes setImmediate callbacks, which + // live on a separate queue. `tick_without_idle` on the loop is a + // non-blocking uSockets tick that fires any ready IPC reads / + // socket callbacks without blocking on stdin. + vm_mut(vm).tick(); + // SAFETY: `event_loop` is the live per-thread event loop. + unsafe { (*core::ptr::from_mut(event_loop)).tick_immediate_tasks(vm_ptr) }; + // SAFETY: `loop_` is the live per-thread uws loop. + unsafe { (*loop_).tick_without_idle() }; + #[cfg(unix)] + { + let state = runtime_state(); + if !state.is_null() { + // SAFETY: `state` is the live per-thread `RuntimeState`; + // see the PORT NOTE on `auto_tick` in jsc_hooks.rs re: + // aliased-&mut across `fire()`. + unsafe { + timer::All::drain_timers(&mut (*state).timer, vm_ptr.cast()); + } + } + } + // An I/O / timer callback can queue up more work that won't wake + // the loop and isn't seen by get_timeout: setImmediate goes on + // its own queue, enqueueTask just appends without signaling, and + // a worker thread landing an enqueueTaskConcurrent after our + // last tick_concurrent has its wakeup eventfd consumed by + // tick_without_idle. Drain all of them now so we don't sleep on + // stranded callbacks. + // SAFETY: `event_loop` is the live per-thread event loop. + unsafe { (*core::ptr::from_mut(event_loop)).tick_immediate_tasks(vm_ptr) }; + if event_loop.tasks.readable_length() > 0 || !event_loop.concurrent_tasks.is_empty() { + vm_mut(vm).tick(); + } + vm_mut(vm).on_after_event_loop(); + // Report unhandled rejections created by timer / IPC / + // setImmediate callbacks *now* rather than stalling them until + // the next stdin wake. Mirrors the final step of `auto_tick`. + vm.global().handle_rejected_promises(); + + // Redraw the prompt + line_editor buffer so any async output + // (e.g. console.log from a parent IPC handler) doesn't visually + // clobber partially-typed input. Only meaningful in TTY mode. + if self.is_tty { + self.refresh_line(); + } else { + Output::flush(); + } + + // Compute how long we're allowed to sleep: the time until the + // next timer fires, or forever if nothing is scheduled. + #[cfg(unix)] + { + let state = runtime_state(); + let has_pending_immediate = !event_loop.immediate_tasks.is_empty(); + // SAFETY: `loop_` is the live per-thread uws loop. + let quic_next_tick_us = unsafe { + let ild = &(*loop_).internal_loop_data; + if ild.quic_head.is_null() { + None + } else { + Some(ild.quic_next_tick_us) + } + }; + let mut timespec = bun_core::Timespec { sec: 0, nsec: 0 }; + let has_deadline = if state.is_null() { + has_pending_immediate || quic_next_tick_us.is_some() + } else { + // SAFETY: `state` is the live per-thread `RuntimeState`; + // see PORT NOTE on `auto_tick` re: aliased-&mut across + // `fire()`. + unsafe { + timer::All::get_timeout( + &mut (*state).timer, + &mut timespec, + has_pending_immediate, + quic_next_tick_us, + vm_ptr.cast(), + ) + } + }; + // Clamp to [0, INT32_MAX]. A negative ms (timer already + // overdue) would otherwise become poll()'s infinite-wait + // sentinel and strand the timer until the next stdin/IPC + // event. + let timeout_ms: c_int = if has_deadline { + let ms = timespec.ms(); + ms.max(0).min(c_int::MAX as i64) as c_int + } else { + -1 + }; + + // Wait on BOTH stdin and the uSockets event-loop fd so an + // incoming IPC message / socket event wakes us as readily as + // a keystroke. + let stdin_fd: c_int = Fd::stdin().native(); + // SAFETY: `loop_` is the live per-thread uws loop; `.fd` is + // the epoll/kqueue fd. + let loop_fd: c_int = unsafe { (*loop_).fd }; + let mut fds = [ + sys::posix::PollFd { + fd: stdin_fd, + events: sys::posix::POLL_IN, + revents: 0, + }, + sys::posix::PollFd { + fd: loop_fd, + events: sys::posix::POLL_IN, + revents: 0, + }, + ]; + match sys::posix::poll(&mut fds, timeout_ms) { + Ok(_) => {} + Err(_) => return None, + } + // POLLNVAL is included so a user who closes fd 0 from inside + // the REPL (e.g. `require('fs').closeSync(0)`) falls through + // to the read() path, which returns EBADF → null → clean + // EOF. + let stdin_mask = libc::POLLIN | libc::POLLHUP | libc::POLLERR | libc::POLLNVAL; + if fds[0].revents & stdin_mask != 0 { + let stdin = sys::File::stdin(); + return stdin.read(&mut self.stdin_buf).ok(); + } + // Otherwise the loop fd fired (pending IPC / socket) or a + // timer is due — loop back and re-pump. + } + #[cfg(windows)] + { + // Windows: no single primitive covers console / pipe / loop + // fd; `windows_wait_for_stdin` slices the wait and branches + // on handle type. + let next_ts = { + let state = runtime_state(); + if state.is_null() { + None + } else { + let mut ts = bun_core::Timespec { sec: 0, nsec: 0 }; + let has_pending_immediate = !event_loop.immediate_tasks.is_empty(); + // Fold the QUIC deadline into the wait slice, same as + // the POSIX branch and `auto_tick`. + // SAFETY: `loop_` is the live per-thread uws loop. + let quic_next_tick_us = unsafe { + let ild = &(*loop_).internal_loop_data; + if ild.quic_head.is_null() { + None + } else { + Some(ild.quic_next_tick_us) + } + }; + // SAFETY: `state` is the live per-thread `RuntimeState`; + // see PORT NOTE on `auto_tick` re: aliased-&mut across + // `fire()`. + let have = unsafe { + timer::All::get_timeout( + &mut (*state).timer, + &mut ts, + has_pending_immediate, + quic_next_tick_us, + vm_ptr.cast(), + ) + }; + if have { Some(ts) } else { None } + } + }; + if !windows_wait_for_stdin(next_ts.as_ref()) { + let stdin = sys::File::stdin(); + let got = match stdin.read(&mut self.stdin_buf) { + sys::Result::Ok(n) => n, + sys::Result::Err(_) => return None, + }; + // On a real Windows console, WaitForSingleObject signals + // for non-character input records (focus/resize/mouse) + // and a subsequent ReadFile can return 0 bytes without + // indicating EOF — treat that as a spurious wake so the + // loop re-pumps instead of exiting the REPL. Gated on + // a real console handle because `FILE_TYPE_CHAR` also + // covers NUL / COM* / LPT* where 0 IS genuine EOF. + if got == 0 && is_windows_console_input(Fd::stdin().native()) { + continue; + } + return Some(got); + } + } + } + } + fn read_key(&mut self) -> Option { let byte = self.read_byte()?; @@ -2554,3 +2778,90 @@ fn is_incomplete_code(code: &[u8]) -> bool { use crate::api::js_transpiler::is_likely_object_literal; const VERSION: &str = Environment::VERSION_STRING; + +// ============================================================================ +// Windows stdin wait helpers +// ============================================================================ + +/// Windows variant of the POSIX `poll` in `wait_for_stdin_readable`. Blocks +/// for up to ~50ms (or less if a timer fires sooner) and returns: +/// `true` → sleep elapsed / spurious wake, caller should re-pump the loop. +/// `false` → stdin has data, caller should read it. +/// +/// Windows has no single wait primitive that covers console handles, +/// anonymous pipes, and the uSockets/libuv loop fd at once, so we branch on +/// handle kind: +/// - Real console input (`GetConsoleMode` succeeds): +/// `WaitForSingleObject` — consoles signal on input events. +/// - Pipe: `WaitForSingleObject` isn't supported; pipe handles are +/// perpetually signaled, so we use `PeekNamedPipe` + `Sleep` instead. +/// - Disk file / NUL / COM* / anything else: treat as always ready. +#[cfg(windows)] +fn windows_wait_for_stdin(next_ts: Option<&bun_core::Timespec>) -> bool { + use bun_sys::windows::{FILE_TYPE_PIPE, GetFileType, WAIT_OBJECT_0, kernel32}; + + let slice_ms: u32 = match next_ts { + None => 50, + Some(ts) => ts.ms().max(0).min(50) as u32, + }; + let stdin_handle = Fd::stdin().native(); + let file_type = GetFileType(stdin_handle); + if is_windows_console_input(stdin_handle) { + let wait = kernel32::WaitForSingleObject(stdin_handle, slice_ms); + // WAIT_OBJECT_0: a console input event is ready — let the caller + // consume it. Consoles signal for *any* input record, not just key + // events; the caller treats a 0-byte return as a spurious wake and + // re-pumps. + if wait == WAIT_OBJECT_0 { + return false; + } + // WAIT_TIMEOUT / WAIT_FAILED / anything else: re-pump. + return true; + } + if file_type == FILE_TYPE_PIPE { + let mut bytes_avail: u32 = 0; + // SAFETY: `stdin_handle` is live while the REPL runs; the out-pointer + // is a valid local and the null buffer / optional out-params follow + // the documented PeekNamedPipe contract. + let ok = unsafe { + kernel32::PeekNamedPipe( + stdin_handle, + core::ptr::null_mut(), + 0, + core::ptr::null_mut(), + &mut bytes_avail, + core::ptr::null_mut(), + ) + }; + if ok == 0 { + // Pipe closed / invalid / other error — let the subsequent + // `read()` surface the EOF consistently. + return false; + } + if bytes_avail > 0 { + return false; + } + kernel32::Sleep(slice_ms); + return true; + } + // Disk / NUL / COM* / unknown — stdin isn't interactive, so we don't + // expect to block. Let the read surface a real EOF. + false +} + +/// True iff `handle` is a real Windows console-input handle. `GetFileType` +/// returns `FILE_TYPE_CHAR` for every character device (console, NUL, COM*, +/// LPT*), but only real consoles accept `GetConsoleMode` / wait-on-input +/// semantics — the others behave like files that return EOF on read. +#[cfg(windows)] +fn is_windows_console_input(handle: *mut core::ffi::c_void) -> bool { + use bun_sys::windows::{FILE_TYPE_CHAR, GetFileType, kernel32}; + + if GetFileType(handle) != FILE_TYPE_CHAR { + return false; + } + let mut mode: u32 = 0; + // SAFETY: `handle` is live while the REPL runs; `mode` is a valid + // out-pointer. + unsafe { kernel32::GetConsoleMode(handle, &mut mode) != 0 } +} diff --git a/src/sys/windows/mod.rs b/src/sys/windows/mod.rs index e00a3b57b020..177a0835b9f6 100644 --- a/src/sys/windows/mod.rs +++ b/src/sys/windows/mod.rs @@ -69,6 +69,18 @@ pub mod kernel32 { // `WAIT_FAILED` + GetLastError, no UB. pub safe fn WaitForSingleObject(hHandle: HANDLE, dwMilliseconds: DWORD) -> DWORD; + // ── pipe readiness / thread sleep (`bun_runtime` repl stdin wait) ── + // safe: by-value `DWORD`; blocks the calling thread, no UB. + pub safe fn Sleep(dwMilliseconds: DWORD); + pub fn PeekNamedPipe( + hNamedPipe: HANDLE, + lpBuffer: *mut c_void, + nBufferSize: DWORD, + lpBytesRead: *mut DWORD, + lpTotalBytesAvail: *mut DWORD, + lpBytesLeftThisMessage: *mut DWORD, + ) -> BOOL; + // ── file moves ── pub fn MoveFileExW( lpExistingFileName: LPCWSTR, @@ -241,6 +253,8 @@ pub use bun_windows_sys::ULONG_PTR; pub type HRESULT = i32; /// `WaitForSingleObject` infinite timeout sentinel. pub const INFINITE: DWORD = 0xFFFF_FFFF; +/// `WaitForSingleObject` success value. +pub use bun_windows_sys::WAIT_OBJECT_0; // ── SRWLOCK / CONDITION_VARIABLE (`bun_threading` windows arm) ──────────── // Win32 defines both as `struct { PVOID Ptr; }`; static-init is all-zero diff --git a/test/js/bun/repl/repl.test.ts b/test/js/bun/repl/repl.test.ts index 0d8e2ce79af9..8fcb9534353e 100644 --- a/test/js/bun/repl/repl.test.ts +++ b/test/js/bun/repl/repl.test.ts @@ -1175,6 +1175,82 @@ describe.todoIf(isWindows)("Bun REPL (Terminal)", () => { await waitFor(/\n\s*2\b/); }); }); + + // Regression: #30559 — the REPL used to block on stdin.read() without + // pumping the JS event loop, so timers and IPC messages from a parent never + // fired inside a `bun repl` child. Now that readByte() polls stdin AND the + // uSockets event-loop fd while ticking the VM between polls, async events + // deliver normally. + test("setTimeout inside REPL fires while waiting for input", async () => { + await withTerminalRepl(async ({ send, waitFor }) => { + // Build the marker from concatenation so it only appears verbatim in + // the timer's console.log output — not in the echoed keystrokes. + send("setTimeout(() => console.log('TIMER' + '_OK'), 100)\n"); + await waitFor("TIMER_OK"); + }); + }); +}); + +describe.todoIf(isWindows)("Bun REPL IPC", () => { + test("process.on('message') fires in a spawned `bun repl` (#30559)", async () => { + // The parent process sends a message on IPC. The REPL child sets up a + // process.on("message") handler and exits 0 when the message arrives. + // Before the fix, readByte() blocked on stdin so the IPC callback never + // ran, the test hangs and the child has to be force-killed. + const { promise: childReady, resolve: resolveChildReady } = Promise.withResolvers(); + const { promise: exited, resolve: resolveExited } = Promise.withResolvers(); + let fromChild: unknown = null; + + await using proc = Bun.spawn({ + cmd: [bunExe(), "repl"], + env: { ...bunEnv, TERM: "dumb", NO_COLOR: "1" }, + stdout: "ignore", + stderr: "pipe", + stdin: "pipe", + ipc(message) { + fromChild = message; + resolveChildReady(); + }, + onExit(subprocess, exitCode) { + resolveExited(exitCode); + // Unblock `await childReady` if the child dies before sending + // "ready" so the failure surfaces as an assertion diff with the + // child's stderr instead of an opaque test timeout. + resolveChildReady(); + }, + }); + + // Set up the child: tell parent we're ready, listen for a reply, exit 0 + // when we get it. No in-child watchdog — the bun:test outer timeout and + // `await using proc` cleanup bound the hang case. + const iife = `(() => { + process.send("ready"); + process.on("message", (data) => { + if (data === "hello-repl") process.exit(0); + process.exit(2); + }); + })();\n`; + proc.stdin.write(iife); + proc.stdin.flush(); + + await childReady; + if (fromChild !== "ready") { + // Child exited before the IIFE ran — put its stderr in the diff. + expect(await new Response(proc.stderr).text()).toBe(""); + } + expect(fromChild).toBe("ready"); + + proc.send("hello-repl"); + + const code = await exited; + // Surface the child's stderr if anything goes wrong so the CI diff is + // diagnostic, per CLAUDE.md. On the happy path the child exits before + // writing anything to stderr. + if (code !== 0) { + expect(await new Response(proc.stderr).text()).toBe(""); + } + expect(code).toBe(0); + }); }); // History file written on REPL exit must be owner-only (0600), since it can