Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
14 commits
Select commit Hold shift + click to select a range
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 8 additions & 2 deletions packages/bun-usockets/src/bsd.c
Original file line number Diff line number Diff line change
Expand Up @@ -63,6 +63,12 @@ static void init_debug_logging() {

#if defined(__APPLE__)
extern int Bun__doesMacOSVersionSupportSendRecvMsgX();
/* XNU's sosend() only skips waiting for buffer space for MSG_NBIO (private, 0x20000); MSG_DONTWAIT alone still blocks there. */
#define LIBUS_SEND_DONTWAIT (MSG_DONTWAIT | 0x20000)

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🟡 nit (optional): maintainers now have the private XNU MSG_NBIO value written twice as a bare literal, so a future correction to one copy silently misses the other. bsd.c:66 spells it as 0x20000 inside LIBUS_SEND_DONTWAIT while src/sys/lib.rs:3104 defines MSG_NBIO: i32 = 0x20000 for the same purpose. Fix: give the C side a named MSG_NBIO-style define (with the XNU source line it comes from) that LIBUS_SEND_DONTWAIT ORs in, so both send paths (bsd_send, bsd_sendmsg) and the Rust SEND_FLAGS_NONBLOCK name the same constant instead of repeating the magic number.

Why this was flagged

The diff adds the Darwin-private send flag in two unrelated modules: packages/bun-usockets/src/bsd.c:66 #define LIBUS_SEND_DONTWAIT (MSG_DONTWAIT | 0x20000) and src/sys/lib.rs:3104 const MSG_NBIO: i32 = 0x20000;. Neither references the other, and the C copy is an unnamed literal inside a macro. If the value or its applicability is ever revised (for example after the private-flag concern already discussed on this PR), whoever edits src/sys/lib.rs will not find the bsd.c copy by name, and usockets TCP/unix socket sends on macOS keep the stale flag while FileSink/spawn stdio sends use the new one. No runtime failure today; the base branch had neither copy. Repository review guidance asks that a fact live in one place and that magic numbers be named constants.

Verification: nit. Both citations check out: packages/bun-usockets/src/bsd.c:66 adds #define LIBUS_SEND_DONTWAIT (MSG_DONTWAIT | 0x20000) with the value as an anonymous literal inside the macro (only the comment on line 65 names it as MSG_NBIO), and src/sys/lib.rs:3104 adds const MSG_NBIO: i32 = 0x20000; (macOS-gated, folded into SEND_FLAGS_NONBLOCK at :3108). The two definitions are in separate modules…

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Leaving the literal in both places: the C side shares no header with bun_sys, and each site carries the one-line XNU note.

#elif defined(_WIN32)
#define LIBUS_SEND_DONTWAIT 0
#else
#define LIBUS_SEND_DONTWAIT MSG_DONTWAIT
#endif

#if defined(_WIN32)
Expand Down Expand Up @@ -946,7 +952,7 @@ ssize_t bsd_recvmsg(LIBUS_SOCKET_DESCRIPTOR fd, struct msghdr *msg, int flags) {
#define MSG_NOSIGNAL 0
#endif
/* Every write to a stream socket uses these, vectored or not. */
#define BSD_SEND_FLAGS (MSG_NOSIGNAL | MSG_DONTWAIT)
#define BSD_SEND_FLAGS (MSG_NOSIGNAL | LIBUS_SEND_DONTWAIT)

#if !defined(_WIN32)
#include <sys/uio.h>
Expand Down Expand Up @@ -1047,7 +1053,7 @@ ssize_t bsd_sendmsg(LIBUS_SOCKET_DESCRIPTOR fd, const struct msghdr *msg, int fl
if (US_FAULT_CHECK(US_FAULT_SENDMSG, fd, injected, unused)) return injected;
(void)unused;
while (1) {
ssize_t rc = sendmsg(fd, msg, flags);
ssize_t rc = sendmsg(fd, msg, flags | MSG_NOSIGNAL | LIBUS_SEND_DONTWAIT);

if (UNLIKELY(IS_EINTR(rc))) {
continue;
Expand Down
25 changes: 24 additions & 1 deletion src/io/PipeReader.rs
Original file line number Diff line number Diff line change
Expand Up @@ -186,6 +186,8 @@ impl ReadLimit {

pub struct PosixBufferedReader {
pub handle: PollOrFd,
/// Set once `preadv2(RWF_NOWAIT)` said this fd's file type does not support it (tty), so we stop asking.
rwf_unsupported: core::cell::Cell<bool>,
pub _buffer: Vec<u8>,
pub(crate) _offset: usize,
limit: ReadLimit,
Expand Down Expand Up @@ -223,6 +225,7 @@ impl PosixBufferedReader {
pub fn init<T: BufferedReaderParent>() -> PosixBufferedReader {
PosixBufferedReader {
handle: PollOrFd::Closed,
rwf_unsupported: core::cell::Cell::new(false),
_buffer: Vec::new(),
_offset: 0,
limit: ReadLimit::NONE,
Expand Down Expand Up @@ -261,6 +264,7 @@ impl PosixBufferedReader {
let kind = self.vtable.kind;
*self = PosixBufferedReader {
handle: mem::replace(&mut other.handle, PollOrFd::Closed),
rwf_unsupported: other.rwf_unsupported.clone(),
_buffer: mem::take(other.buffer()),
_offset: other._offset,
limit: other.limit,
Expand Down Expand Up @@ -674,7 +678,26 @@ impl PosixBufferedReader {
}
FileType::File => sys::read(fd, buf),
FileType::Socket => sys::recv_non_block(fd, buf),
FileType::NonblockingPipe | FileType::Pipe => sys::read_nonblocking(fd, buf),
FileType::NonblockingPipe | FileType::Pipe => {
#[cfg(any(target_os = "linux", target_os = "android"))]
{
if !self.rwf_unsupported.get() {
match sys::read_nowait(fd, buf) {
Ok(None) => self.rwf_unsupported.set(true),
Ok(Some(n)) => return Ok(n),
Err(e) => return Err(e),
}
}
// Poll first even when labelled nonblocking: some callers (FileResponseStream) label by fd kind, not by O_NONBLOCK.
match bun_core::is_readable(fd) {
bun_core::Pollable::Ready | bun_core::Pollable::Hup => sys::read(fd, buf),
bun_core::Pollable::NotReady => Err(sys::Error::retry().with_fd(fd)),
}
}
// macOS poll(2) is unreliable on FIFOs; the kqueue registration drives readiness there.
#[cfg(not(any(target_os = "linux", target_os = "android")))]
sys::read(fd, buf)
}
}
}

Expand Down
72 changes: 56 additions & 16 deletions src/io/PipeWriter.rs
Original file line number Diff line number Diff line change
Expand Up @@ -61,26 +61,57 @@ pub trait PosixPipeWriter {

fn handle(&self) -> &PollOrFd;

/// Set once `pwritev2(RWF_NOWAIT)` said this fd's file type does not support it, so we stop asking.
fn rwf_unsupported(&self) -> &core::cell::Cell<bool>;

/// Only reads `get_file_type()` / `get_fd()` from `self`; takes `&self` so
/// callers may pass a `buf` that borrows from a field of `self` (e.g.
/// `self.outgoing.slice()`) without raw-pointer aliasing escapes.
fn try_write(&self, force_sync: bool, buf: &[u8]) -> WriteResult {
// PERF: try_write_with_write_fn is not monomorphized per FileType —
// profile if hot.
let ft = if !force_sync {
self.get_file_type()
} else {
FileType::File
};
match ft {
FileType::NonblockingPipe | FileType::File => {
self.try_write_with_write_fn(buf, sys::write)
let ft = self.get_file_type();
// Linux: RWF_NOWAIT is a free per-call nonblocking write, so a pipe stays async even after a spawn cleared O_NONBLOCK on it.
#[cfg(any(target_os = "linux", target_os = "android"))]
if matches!(ft, FileType::Pipe | FileType::NonblockingPipe) && !self.rwf_unsupported().get()
{
match self.try_write_nowait(buf) {
Some(rc) => return rc,
// Not a pipe as far as pwritev2 is concerned (tty, pty master, old kernel): remember, use the plain path.
None => self.rwf_unsupported().set(true),
}
FileType::Pipe => self.try_write_with_write_fn(buf, write_to_blocking_pipe),
}
match ft {
// send(MSG_DONTWAIT | MSG_NBIO) is nonblocking per call on every Unix, whatever the fd's flags.
FileType::Socket => self.try_write_with_write_fn(buf, write_to_socket),
FileType::Pipe if !force_sync => {
self.try_write_with_write_fn(buf, write_to_blocking_pipe)
}
_ => self.try_write_with_write_fn(buf, sys::write),
}
}

/// `None` if `RWF_NOWAIT` is not usable on this fd (nothing was written).
#[cfg(any(target_os = "linux", target_os = "android"))]
fn try_write_nowait(&self, buf: &[u8]) -> Option<WriteResult> {
let fd = self.get_fd();
if fd == Fd::INVALID {
return Some(WriteResult::Done(0));
}
let mut offset: usize = 0;
while offset < buf.len() {
match sys::write_nowait(fd, &buf[offset..]) {
Ok(None) if offset == 0 => return None,
Ok(None) => return Some(WriteResult::Pending(offset)),
Ok(Some(0)) => return Some(WriteResult::Done(offset)),
Ok(Some(wrote)) => offset += wrote,
Err(err) if err.is_retry() => return Some(WriteResult::Pending(offset)),
Err(err) => return Some(WriteResult::Err(err)),
}
}
Some(WriteResult::Wrote(offset))
}

fn try_write_with_write_fn(
&self,
buf: &[u8],
Expand Down Expand Up @@ -219,13 +250,6 @@ pub trait PosixPipeWriter {
/// Free fn for the blocking-pipe path; the other file types are handled
/// inline in `try_write` above.
fn write_to_blocking_pipe(fd: Fd, buf: &[u8]) -> sys::Result<usize> {
#[cfg(any(target_os = "linux", target_os = "android"))]
{
if bun_sys::linux::RWFFlagSupport::is_maybe_supported() {
return sys::write_nonblocking(fd, buf);
}
}

match bun_core::is_writable(fd) {
bun_core::Pollable::Ready | bun_core::Pollable::Hup => sys::write(fd, buf),
bun_core::Pollable::NotReady => sys::Result::Err(sys::Error::retry()),
Expand All @@ -237,6 +261,12 @@ fn write_to_blocking_pipe(fd: Fd, buf: &[u8]) -> sys::Result<usize> {
fn write_to_socket(fd: Fd, buf: &[u8]) -> sys::Result<usize> {
sys::send_non_block(fd, buf).map_err(|err| sys::Error {
syscall: sys::Tag::write,
// XNU's send() says ENOTCONN once a stream peer is fully gone; write(2) on the same fd (and Node) say EPIPE.
errno: if err.get_errno() == sys::E::ENOTCONN {
Comment on lines +264 to +265

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🟡 nit (optional): the new ENOTCONN-to-EPIPE remap in write_to_socket ships with no assertion that observes it, so a future regression back to ENOTCONN passes the suite. The only test that hits this path, test/js/bun/console/console-write.test.ts:134-136, still accepts either code on darwin, an allowance this change makes dead. Fix: tighten that test to expect exactly "caught EPIPE\n" on every platform (or add a sibling case), so the remap at src/io/PipeWriter.rs:265 is covered and the stale darwin branch goes away.

Why this was flagged

A macOS user runs a child whose stdout is a Bun.spawn socketpair and the parent cancels its reader mid-write; the child's console.write / process.stdout.write reaches write_to_socket (src/io/PipeWriter.rs:262) and XNU's send() returns ENOTCONN. The diff at src/io/PipeWriter.rs:264-269 now rewrites that errno to EPIPE before it reaches JS, which is the intended Node-matching behaviour and is not exercised by any of the nine new tests in test/js/node/process/process-stdio.test.ts. The pre-existing test test/js/bun/console/console-write.test.ts:106-138 reproduces exactly this scenario, but line 135 keeps const codes = process.platform === "darwin" ? ["EPIPE", "ENOTCONN"] : ["EPIPE"] and line 136 uses toBeOneOf, so it passes whether or not the remap works. On the base branch the darwin allowance was load-bearing; after this change it is dead and hides a regression. No runtime failure results; the cost is that the behavioural change has no test that can fail, which REVIEW.md asks for, and an assertion on the old behaviour was not updated in the same PR.

Verification: nit — triggers whenever the ENOTCONN→EPIPE remap regresses (or is dropped) on macOS: no test in the PR would fail. The diff adds at src/io/PipeWriter.rs:264-269 errno: if err.get_errno() == sys::E::ENOTCONN { sys::E::EPIPE as _ } else { err.errno } inside write_to_socket, which is the only caller-facing socket write path (src/io/PipeWriter.rs:86 `FileType::Socket =>… | nit — triggers whenever…

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This comes from the rebase. #44072 merged into main on 2026-09-27, after this commit was written. It added the macOS ENOTCONN allowance to that test. The rebase put the two together. The test passes with either errno, so CI is green.

Two facts from main are relevant to this commit:

I changed nothing here. The request was a rebase, and I cannot run tests on macOS. There are two ways to make the branch consistent:

  1. Keep the mapping. Then the test can expect EPIPE only on every platform, and it covers the mapping.
  2. Pass the kernel errno through. Then this commit can go, and the test stays as it is on main.

Tell me which one you want and I will push it.

sys::E::EPIPE as _
} else {
err.errno
},
..err
})
}
Expand Down Expand Up @@ -285,6 +315,7 @@ pub struct PosixBufferedWriter<Parent: PosixBufferedWriterParent> {
pub(crate) pollable: bool,
pub(crate) closed_without_reporting: bool,
pub close_fd: bool,
rwf_unsupported: core::cell::Cell<bool>,
}

impl<Parent: PosixBufferedWriterParent> Default for PosixBufferedWriter<Parent> {
Expand All @@ -296,6 +327,7 @@ impl<Parent: PosixBufferedWriterParent> Default for PosixBufferedWriter<Parent>
pollable: false,
closed_without_reporting: false,
close_fd: true,
rwf_unsupported: core::cell::Cell::new(false),
}
}
}
Expand Down Expand Up @@ -325,6 +357,9 @@ impl<Parent: PosixBufferedWriterParent> PosixPipeWriter for PosixBufferedWriter<
fn handle(&self) -> &PollOrFd {
&self.handle
}
fn rwf_unsupported(&self) -> &core::cell::Cell<bool> {
&self.rwf_unsupported
}
}

// SAFETY: writer is an intrusive field of `Parent`; `Parent::on_write`
Expand Down Expand Up @@ -610,6 +645,7 @@ pub struct PosixStreamingWriter<Parent: PosixStreamingWriterParent> {
pub force_sync: bool,
/// Last reported `WriteStatus == Pending` (i.e. write(2) returned EAGAIN).
backed_up: core::cell::Cell<bool>,
rwf_unsupported: core::cell::Cell<bool>,
}

impl<Parent: PosixStreamingWriterParent> Default for PosixStreamingWriter<Parent> {
Expand All @@ -622,6 +658,7 @@ impl<Parent: PosixStreamingWriterParent> Default for PosixStreamingWriter<Parent
closed_without_reporting: false,
force_sync: false,
backed_up: core::cell::Cell::new(false),
rwf_unsupported: core::cell::Cell::new(false),
}
}
}
Expand Down Expand Up @@ -651,6 +688,9 @@ impl<Parent: PosixStreamingWriterParent> PosixPipeWriter for PosixStreamingWrite
fn handle(&self) -> &PollOrFd {
&self.handle
}
fn rwf_unsupported(&self) -> &core::cell::Cell<bool> {
&self.rwf_unsupported
}
}

// SAFETY: see `PosixBufferedWriter`'s `LaunderedSelf` impl — identical shape.
Expand Down
20 changes: 7 additions & 13 deletions src/io/openForWriting.rs
Original file line number Diff line number Diff line change
Expand Up @@ -59,7 +59,7 @@ pub fn open_for_writing<P, C>(
force_sync: bool,
out_nonblocking: &mut bool,
ctx: C,
on_force_sync_or_isa_tty: fn(C),
on_tty: fn(C),
is_pollable: fn(mode: Mode) -> bool,
) -> bun_sys::Result<Fd>
where
Expand All @@ -75,7 +75,7 @@ where
force_sync,
out_nonblocking,
ctx,
on_force_sync_or_isa_tty,
on_tty,
is_pollable,
bun_sys::openat,
)
Expand All @@ -91,7 +91,7 @@ pub fn open_for_writing_impl<P, C>(
force_sync: bool,
out_nonblocking: &mut bool,
ctx: C,
on_force_sync_or_isa_tty: fn(C),
on_tty: fn(C),
is_pollable: fn(mode: Mode) -> bool,
openat: fn(dir: Fd, path: &ZStr, flags: i32, mode: Mode) -> bun_sys::Result<Fd>,
) -> bun_sys::Result<Fd>
Expand All @@ -100,13 +100,7 @@ where
{
#[cfg(windows)]
{
let _ = (
is_socket,
out_nonblocking,
ctx,
on_force_sync_or_isa_tty,
is_pollable,
);
let _ = (is_socket, out_nonblocking, ctx, on_tty, is_pollable);
}
// TODO: this should be concurrent.
#[cfg(unix)]
Expand Down Expand Up @@ -136,6 +130,9 @@ where

*is_socket = bun_sys::S::ISSOCK(stat.st_mode as Mode);

if isatty {
on_tty(ctx);
}
if force_sync || isatty {
// Prevents interleaved or dropped stdout/stderr output for terminals.
// As noted in the following reference, local TTYs tend to be quite fast and
Expand All @@ -144,9 +141,6 @@ where
// Ref: https://github.com/nodejs/node/pull/1771#issuecomment-119351671
let _ = bun_sys::update_nonblocking(fd, false);
is_nonblocking = false;
// this.force_sync = true;
// this.writer.force_sync = true;
on_force_sync_or_isa_tty(ctx);
} else if !is_nonblocking {
let flags = match bun_sys::get_fcntl_flags(fd) {
Ok(flags) => flags,
Expand Down
4 changes: 4 additions & 0 deletions src/io/posix_event_loop.rs
Original file line number Diff line number Diff line change
Expand Up @@ -324,6 +324,9 @@ impl FilePoll {
if flags.contains(Flags::Socket) {
return FileType::Socket;
}
if flags.contains(Flags::Tty) {
return FileType::File;
}
if flags.contains(Flags::Nonblocking) {
return FileType::NonblockingPipe;
}
Expand Down Expand Up @@ -1212,6 +1215,7 @@ pub enum Flags {
IgnoreUpdates,

Socket,
Tty,
}

pub type FlagsSet = enumset::EnumSet<Flags>;
Expand Down
Loading
Loading