diff --git a/packages/bun-usockets/src/context.c b/packages/bun-usockets/src/context.c index c70d6b20b245..eac220477201 100644 --- a/packages/bun-usockets/src/context.c +++ b/packages/bun-usockets/src/context.c @@ -347,6 +347,7 @@ static void us_internal_init_listen_socket(struct us_listen_socket_t *ls, s->unclassified_send_failures = 0; s->read_eof = 0; s->hangup_closes_unsent = 0; + s->no_reader = 0; s->next = 0; s->prev = 0; s->connect_state = NULL; @@ -527,6 +528,7 @@ static inline void us_internal_init_connect_socket(struct us_socket_t *s, s->unclassified_send_failures = 0; s->read_eof = 0; s->hangup_closes_unsent = 0; + s->no_reader = 0; s->connect_state = NULL; s->connect_next = NULL; } diff --git a/packages/bun-usockets/src/eventing/libuv.c b/packages/bun-usockets/src/eventing/libuv.c index a76d0e73f48d..4cedf419d4de 100644 --- a/packages/bun-usockets/src/eventing/libuv.c +++ b/packages/bun-usockets/src/eventing/libuv.c @@ -95,8 +95,9 @@ static void poll_cb(uv_poll_t *p, int status, int events) { * the same EOF and busy-loop on_end; keeping DISCONNECT armed would * complete instantly forever. A dead peer surfaces via SO_ERROR or the * zero-byte send probe and goes through the shared error path (which - * reads off whatever is still queued and closes); a FIN, fresh on a - * paused socket or re-reported on a half-open one, quiesces with only + * reads off whatever is still queued, unless the socket is no_reader, + * and closes); a FIN, fresh on a paused socket or re-reported on a + * half-open one, quiesces with only * the ABORT-only subscription (UV_PRIORITIZED) kept armed so a later * RST still has an event to ride, and a paused socket meets the FIN * again through recv() once resume() re-arms READABLE. Non-SOCKET kinds diff --git a/packages/bun-usockets/src/internal/internal.h b/packages/bun-usockets/src/internal/internal.h index 5c87844169fa..03a77c6413dd 100644 --- a/packages/bun-usockets/src/internal/internal.h +++ b/packages/bun-usockets/src/internal/internal.h @@ -370,6 +370,8 @@ struct us_socket_t { /* A hangup that leaves bytes unsent closes the socket even while it is paused (loop.c defers it otherwise). * For an owner whose pause can wait for those bytes to drain: node:http's pipelining. */ unsigned char hangup_closes_unsent : 1; + /* Paused by an owner that never reads (us_socket_pause_no_reader): an error event closes without draining the receive queue. */ + unsigned char no_reader : 1; /* The close code passed to the deferred close (e.g. a reset requested from * inside a handshake callback must still RST, not FIN, when it is finally * performed). */ diff --git a/packages/bun-usockets/src/libusockets.h b/packages/bun-usockets/src/libusockets.h index bb78632bb6a0..9e8baf47a1fb 100644 --- a/packages/bun-usockets/src/libusockets.h +++ b/packages/bun-usockets/src/libusockets.h @@ -763,6 +763,8 @@ int us_socket_set_tos(us_socket_r s, int tos); int us_socket_get_tos(us_socket_r s); void us_socket_resume(us_socket_r s); void us_socket_pause(us_socket_r s); +/* us_socket_pause for an owner that never reads: until us_socket_resume, an error event closes the socket without draining the receive queue into on_data. */ +void us_socket_pause_no_reader(us_socket_r s); #ifdef __cplusplus } diff --git a/packages/bun-usockets/src/loop.c b/packages/bun-usockets/src/loop.c index f803999f6590..4d6a63c3b8a2 100644 --- a/packages/bun-usockets/src/loop.c +++ b/packages/bun-usockets/src/loop.c @@ -556,6 +556,7 @@ void us_internal_dispatch_ready_poll(struct us_poll_t *p, int error, int eof, in s->unclassified_send_failures = 0; s->read_eof = 0; s->hangup_closes_unsent = 0; + s->no_reader = 0; /* We always use nodelay */ bsd_socket_nodelay(client_fd, 1); @@ -643,8 +644,10 @@ void us_internal_dispatch_ready_poll(struct us_poll_t *p, int error, int eof, in * control, and there is no later for a dead connection to flow into. recv() * then returns the data and after it the error, which is what libuv reports * to node as well. A socket parked in the low-priority queue is not linked - * where on_data expects it and takes the plain error close. */ - const int drain_for_error = error && !s->read_eof && s->flags.low_prio_state != 1; + * where on_data expects it and takes the plain error close. So does a + * no_reader socket: nothing takes its bytes, and libuv reads only while + * UV_HANDLE_READING is set (nodejs/node v26.3.0 deps/uv/src/unix/stream.c#L1043-L1045). */ + const int drain_for_error = error && !s->read_eof && !s->no_reader && s->flags.low_prio_state != 1; if ((events & LIBUS_SOCKET_READABLE) || drain_for_error) { /* Contexts may prioritize down sockets that are currently readable, e.g. when SSL handshake has to be done. * SSL handshakes are CPU intensive, so we limit the number of handshakes per loop iteration, and move the rest diff --git a/packages/bun-usockets/src/socket.c b/packages/bun-usockets/src/socket.c index 4cb0f83b90e3..419bc78ac461 100644 --- a/packages/bun-usockets/src/socket.c +++ b/packages/bun-usockets/src/socket.c @@ -445,6 +445,7 @@ struct us_socket_t *us_socket_from_fd(struct us_socket_group_t *group, unsigned s->unclassified_send_failures = 0; s->read_eof = 0; s->hangup_closes_unsent = 0; + s->no_reader = 0; s->connect_state = NULL; /* We always use nodelay */ @@ -864,9 +865,15 @@ void us_socket_pause(struct us_socket_t *s) { s->flags.is_paused = 1; } +void us_socket_pause_no_reader(struct us_socket_t *s) { + us_socket_pause(s); + s->no_reader = s->flags.is_paused; +} + void us_socket_resume(struct us_socket_t *s) { if (!s->flags.is_paused) return; s->flags.is_paused = 0; + s->no_reader = 0; // closed cannot be resumed if (us_socket_is_closed(s)) return; diff --git a/src/js/node/net.ts b/src/js/node/net.ts index d7bb3f5e7b22..d00ab3cbcdda 100644 --- a/src/js/node/net.ts +++ b/src/js/node/net.ts @@ -955,6 +955,13 @@ function readStop(self, handle) { if (!self[kwriteCallback]) handle?.unref?.(); } +let pauseNoReader; +// A socket that never reads (`readable: false`, no `onread` buffer) also leaves a reset's unread bytes in the kernel: libuv reads only while UV_HANDLE_READING, https://github.com/nodejs/node/blob/v26.3.0/deps/uv/src/unix/stream.c#L1043-L1045 +function readStopNoReader(self, handle) { + (pauseNoReader ??= $newRustFunction("runtime/socket/socket.rs", "jsPauseNoReader", 1))(handle); + readStop(self, handle); +} + // https://github.com/nodejs/node/blob/v26.3.0/lib/net.js#L494-L498 function pauseOnCreate(self, handle) { readStop(self, handle); @@ -2324,6 +2331,7 @@ Socket.prototype.connect = function connect(...args) { this.emit("close", true); } }); + if (this.readableEnded && this[kOnreadBuffer] === undefined) readStopNoReader(this, this._handle); } if (pauseOnConnect) { // An fd is open already; a dial is paused when it opens, and afterConnect releases its hold. @@ -3608,8 +3616,12 @@ function afterConnect(status, handle, req, readable, writable) { // Ours already reads, Node's starts at read(): stop a paused plain socket now, and after the listeners unless one asked for a read. // A socket built with `readable: false` never reads: read() cannot reach _read once the readable side has ended. - const pausedBeforeConnect = self.isPaused() || (self.readableEnded && self[kOnreadBuffer] === undefined); - if (pausedBeforeConnect && !self.encrypted) readStop(self, self._handle); + const neverReads = self.readableEnded && self[kOnreadBuffer] === undefined; + const pausedBeforeConnect = neverReads || self.isPaused(); + if (pausedBeforeConnect && !self.encrypted) { + if (neverReads) readStopNoReader(self, self._handle); + else readStop(self, self._handle); + } self.emit("connect"); self.emit("ready"); diff --git a/src/runtime/socket/mod.rs b/src/runtime/socket/mod.rs index 52232423205a..da7b0fb2c600 100644 --- a/src/runtime/socket/mod.rs +++ b/src/runtime/socket/mod.rs @@ -123,8 +123,9 @@ pub(crate) use udp_socket::UDPSocket; pub(crate) mod socket { pub(crate) use super::socket_body::{ js_check_server_identity_in_handshake, js_create_socket_pair, js_first_flight_before_fin, - js_get_buffered_amount, js_is_named_pipe_socket, js_release_held_flight, - js_set_socket_options, js_upgrade_duplex_to_tls, js_upgrade_tls_deferred, testing_ap_is, + js_get_buffered_amount, js_is_named_pipe_socket, js_pause_no_reader, + js_release_held_flight, js_set_socket_options, js_upgrade_duplex_to_tls, + js_upgrade_tls_deferred, testing_ap_is, }; } diff --git a/src/runtime/socket/socket_body.rs b/src/runtime/socket/socket_body.rs index 64522212099d..245a794d760f 100644 --- a/src/runtime/socket/socket_body.rs +++ b/src/runtime/socket/socket_body.rs @@ -780,6 +780,13 @@ impl NewSocket { } } + /// Not gated on `IS_PAUSED`: a socket that `pauseOnConnect` paused still has to be marked. + fn pause_reads_no_reader(&self) { + if self.socket.get().pause_stream_no_reader() { + self.update_flags(|f| f.insert(Flags::IS_PAUSED)); + } + } + /// Consumes `PAUSE_ON_CONNECT`: a TLS client dispatches another handshake after a renegotiation. fn pause_on_connect_once(&self) { if self.flags.get().contains(Flags::PAUSE_ON_CONNECT) { @@ -5058,6 +5065,20 @@ pub(crate) fn js_get_buffered_amount( Ok(JSValue::js_number(0.0)) } +/// node:net stops a socket that never reads (`readable: false`). A TLS socket always reads. +#[bun_jsc::host_fn] +pub(crate) fn js_pause_no_reader( + _global: &JSGlobalObject, + callframe: &CallFrame, +) -> JsResult { + jsc::mark_binding!(); + let [socket] = callframe.arguments_as_array::<1>(); + if let Some(this) = socket.as_class_ref::() { + this.pause_reads_no_reader(); + } + Ok(JSValue::UNDEFINED) +} + #[bun_jsc::host_fn] pub(crate) fn js_create_socket_pair( global: &JSGlobalObject, diff --git a/src/uws_sys/socket.rs b/src/uws_sys/socket.rs index bc50a421c102..d4c976e1cac8 100644 --- a/src/uws_sys/socket.rs +++ b/src/uws_sys/socket.rs @@ -504,6 +504,17 @@ impl NewSocketHandler { ) } + /// `pause_stream` for an owner that never reads: see `us_socket_pause_no_reader`. + pub fn pause_stream_no_reader(&self) -> bool { + match self.socket { + InternalSocket::Connected(s) if sock(s).is_established() => { + sock(s).pause_no_reader(); + true + } + _ => self.pause_stream(), + } + } + pub fn resume_stream(&self) -> bool { on_socket!(self.socket; connected s => if s.is_established() { s.resume(); true } else { false }, diff --git a/src/uws_sys/us_socket_t.rs b/src/uws_sys/us_socket_t.rs index d10392132d24..ffe4a67c068a 100644 --- a/src/uws_sys/us_socket_t.rs +++ b/src/uws_sys/us_socket_t.rs @@ -79,6 +79,11 @@ impl us_socket_t { c::us_socket_pause(self); } + pub(crate) fn pause_no_reader(&mut self) { + bun_core::scoped_log!(uws, "us_socket_pause_no_reader({:p})", self); + c::us_socket_pause_no_reader(self); + } + pub fn resume(&mut self) { bun_core::scoped_log!(uws, "us_socket_resume({:p})", self); c::us_socket_resume(self); @@ -562,6 +567,7 @@ mod c { pub(super) safe fn us_socket_flush(s: &mut us_socket_t); pub(super) safe fn us_socket_pause(s: &mut us_socket_t); + pub(super) safe fn us_socket_pause_no_reader(s: &mut us_socket_t); pub(super) safe fn us_socket_resume(s: &mut us_socket_t); pub(super) fn us_socket_close( s: *mut us_socket_t, diff --git a/test/js/node/net/node-net.test.ts b/test/js/node/net/node-net.test.ts index 957a90819872..211716959c90 100644 --- a/test/js/node/net/node-net.test.ts +++ b/test/js/node/net/node-net.test.ts @@ -1,4 +1,5 @@ import { Socket as _BunSocket, TCPSocketListener } from "bun"; +import { createSocketPair } from "bun:internal-for-testing"; import { heapStats } from "bun:jsc"; import { describe, expect, it } from "bun:test"; import { @@ -10,6 +11,7 @@ import { gc, isASAN, isDebug, + isLinux, isMusl, isWindows, tempDir, @@ -35,7 +37,7 @@ import { Stream, } from "node:net"; import { join } from "node:path"; -import { TLSSocket } from "node:tls"; +import { connect as tlsConnect, TLSSocket } from "node:tls"; const socket_domain = tmpdirSync(); @@ -1357,6 +1359,156 @@ it("a client dialed with readable: false never reads and keeps writing", async ( } }); +// A reset is reported to a stopped handle too. For a flow-control pause the native layer first +// reads what the kernel still holds, so that the tail of a stream is not lost (#39846). A +// `readable: false` socket never reads, so its bytes stay in the kernel, as in node. Reading and +// pushing them into the ended Readable replaced the reset with ERR_STREAM_PUSH_AFTER_EOF. (node +// does not poll such a handle at all, so it finds the reset at the next write, also as ECONNRESET.) +it.each([false, true])( + "a client dialed with readable: false reports a reset that follows unread data as ECONNRESET (pauseOnConnect: %p)", + async pauseOnConnect => { + const accepted = Promise.withResolvers(); + const server = createServer(socket => { + socket.on("error", () => {}); + accepted.resolve(socket); + }); + await once(server.listen(0, "127.0.0.1"), "listening"); + try { + const client = connect({ + port: (server.address() as import("node:net").AddressInfo).port, + host: "127.0.0.1", + readable: false, + pauseOnConnect, + } as import("node:net").NetConnectOpts); + const events: string[] = []; + client.on("data", chunk => events.push(`data:${chunk.length}`)); + client.on("end", () => events.push("end")); + client.on("error", err => events.push(`error:${(err as NodeJS.ErrnoException).code}`)); + const closed = Promise.withResolvers(); + client.on("close", closed.resolve); + const [peer] = await Promise.all([accepted.promise, once(client, "connect")]); + // The banner sits unread in the client's kernel buffer when the reset arrives. + peer.write("banner", () => peer.resetAndDestroy()); + expect({ hadError: await closed.promise, events, bytesRead: client.bytesRead }).toEqual({ + hadError: true, + events: ["error:ECONNRESET"], + bytesRead: 0, + }); + } finally { + server.close(); + } + }, +); + +// The same rule for a connected fd the socket adopts: every byte the peer sent was pushed after EOF. +// createSocketPair() is not implemented on Windows. +it.skipIf(isWindows)("a readable: false socket over an adopted fd never reads and keeps writing", async () => { + const [fd, peerFd] = createSocketPair(); + const client = new Socket({ readable: false }); + try { + const events: string[] = []; + client.on("data", chunk => events.push(`data:${chunk.length}`)); + client.on("end", () => events.push("end")); + client.on("error", err => events.push(`error:${(err as NodeJS.ErrnoException).code}`)); + client.on("close", () => events.push("close")); + client.connect({ fd } as never); + fs.writeSync(peerFd, "banner"); + // The banner is readable the whole time; give the loop several turns to (wrongly) read it. + for (let i = 0; i < 5 && events.length === 0; i++) await new Promise(resolve => setImmediate(resolve)); + let late = ""; + if (!client.destroyed) { + await new Promise((resolve, reject) => client.write("late", err => (err ? reject(err) : resolve()))); + const chunk = Buffer.alloc(16); + late = chunk.toString("latin1", 0, fs.readSync(peerFd, chunk)); + } + expect({ events, bytesRead: client.bytesRead, late }).toEqual({ events: [], bytesRead: 0, late: "late" }); + } finally { + client.destroy(); + fs.closeSync(peerFd); + } +}); + +// Linux resets a unix-domain connection when one end closes with a byte unread. Other kernels are not relied on here. +it.skipIf(!isLinux)( + "a readable: false socket over an adopted fd reports a reset that follows unread data as ECONNRESET", + async () => { + const [fd, peerFd] = createSocketPair(); + let peerOpen = true; + const client = new Socket({ readable: false }); + try { + const events: string[] = []; + client.on("data", chunk => events.push(`data:${chunk.length}`)); + client.on("end", () => events.push("end")); + client.on("error", err => events.push(`error:${(err as NodeJS.ErrnoException).code}`)); + const closed = Promise.withResolvers(); + client.on("close", closed.resolve); + client.connect({ fd } as never); + // The peer never reads this byte, so its close() is a reset, with the banner unread on the client's side. + await new Promise((resolve, reject) => client.write("x", err => (err ? reject(err) : resolve()))); + fs.writeSync(peerFd, "banner"); + fs.closeSync(peerFd); + peerOpen = false; + expect({ hadError: await closed.promise, events, bytesRead: client.bytesRead }).toEqual({ + hadError: true, + events: ["error:ECONNRESET"], + bytesRead: 0, + }); + } finally { + client.destroy(); + if (peerOpen) fs.closeSync(peerFd); + } + }, +); + +// The wrapped socket never reads, the TLS layer that adopts its fd does: a reset behind unread data +// still hands the tail to the TLS socket (#39846). Windows discards the receive queue on a reset. +it.skipIf(isWindows)( + "a TLS socket over a socket dialed with readable: false gets the data a reset follows", + async () => { + const highWaterMark = 64 * 1024; + const accepted = Promise.withResolvers<_BunSocket>(); + using listener = Bun.listen({ + hostname: "127.0.0.1", + port: 0, + tls: tlsCert, + socket: { + handshake(peer, success, authorizationError) { + if (success) accepted.resolve(peer); + else accepted.reject(authorizationError ?? new Error("server handshake failed")); + }, + data() {}, + error: (_peer, error) => accepted.reject(error), + }, + }); + const raw = connect({ port: listener.port, host: "127.0.0.1", readable: false }); + raw.on("error", () => {}); + await once(raw, "connect"); + const secure = tlsConnect({ socket: raw, ca: tlsCert.cert, servername: "localhost", highWaterMark }); + const events: string[] = []; + const closed = Promise.withResolvers(); + secure.on("end", () => events.push("end")); + secure.on("error", err => events.push(`error:${(err as NodeJS.ErrnoException).code}`)); + secure.on("close", () => closed.resolve()); + const [peer] = await Promise.all([accepted.promise, once(secure, "secureConnect")]); + + // Nothing reads `secure`, so one highWaterMark of data stops its handle. + expect(peer.write(Buffer.alloc(highWaterMark, "a"))).toBe(highWaterMark); + peer.flush(); + while (secure.readableLength < highWaterMark && !secure.destroyed) { + await new Promise(resolve => setImmediate(resolve)); + } + // In the kernel ahead of the reset: same connection, so TCP orders them. + expect(peer.write("tail")).toBe(4); + peer.flush(); + peer.terminate(); + await closed.promise; + expect({ events, bytesRead: secure.bytesRead }).toEqual({ + events: ["error:ECONNRESET"], + bytesRead: highWaterMark + 4, + }); + }, +); + // https://github.com/nodejs/node/blob/v26.3.0/lib/net.js#L830-L845 it("an onread client dialed with readable: false still reads into its buffer", async () => { const server = createServer(socket => socket.end("banner"));