diff --git a/packages/bun-usockets/src/context.c b/packages/bun-usockets/src/context.c index 6005f46f7e26..8c07322be9c4 100644 --- a/packages/bun-usockets/src/context.c +++ b/packages/bun-usockets/src/context.c @@ -379,6 +379,7 @@ static void us_internal_init_listen_socket(struct us_listen_socket_t *ls, ls->on_server_name = NULL; ls->socket_ext_size = socket_ext_size; ls->deferred_accept = 0; + ls->accept_paused = (options & LIBUS_SOCKET_OPEN_PAUSED) && !ssl_ctx; /* Link into the group so close_all() / test-isolation can find it. */ ls->next = group->head_listen_sockets; diff --git a/packages/bun-usockets/src/eventing/epoll_kqueue.c b/packages/bun-usockets/src/eventing/epoll_kqueue.c index 7fe3f4272d6a..3ee1d0a09680 100644 --- a/packages/bun-usockets/src/eventing/epoll_kqueue.c +++ b/packages/bun-usockets/src/eventing/epoll_kqueue.c @@ -690,7 +690,11 @@ int us_poll_start_rc(struct us_poll_t *p, struct us_loop_t *loop, int events) { } while (IS_EINTR(ret)); return ret; #else - return kqueue_change(loop->fd, p->state.fd, 0, events, p, 0); + /* A socket that starts without read interest (LIBUS_SOCKET_OPEN_PAUSED) takes the same + * transition as one that pauses, so it too keeps the read knote that reports the peer's + * FIN or RST while it is paused (see kqueue_change). */ + int starts_paused = kqueue_is_socket_poll(p) && !(events & LIBUS_SOCKET_READABLE); + return kqueue_change(loop->fd, p->state.fd, starts_paused ? LIBUS_SOCKET_READABLE : 0, events, p, starts_paused); #endif } diff --git a/packages/bun-usockets/src/internal/internal.h b/packages/bun-usockets/src/internal/internal.h index d8289364c3ed..b91c9de8c29f 100644 --- a/packages/bun-usockets/src/internal/internal.h +++ b/packages/bun-usockets/src/internal/internal.h @@ -475,6 +475,8 @@ struct us_listen_socket_t { unsigned char accept_kind; /* Set when TCP_DEFER_ACCEPT/SO_ACCEPTFILTER was successfully applied. */ unsigned char deferred_accept; + /* LIBUS_SOCKET_OPEN_PAUSED: accepted sockets start without read interest. */ + unsigned char accept_paused; }; void us_internal_socket_group_link_connecting_socket(us_socket_group_r group, struct us_connecting_socket_t *c); diff --git a/packages/bun-usockets/src/libusockets.h b/packages/bun-usockets/src/libusockets.h index a2b7522d2b8a..449697145929 100644 --- a/packages/bun-usockets/src/libusockets.h +++ b/packages/bun-usockets/src/libusockets.h @@ -160,6 +160,12 @@ enum { * unconnected socket it also makes the next send fail for a datagram * bound to a different, live peer. */ LIBUS_UDP_LINUX_RECVERR = 128, + /* A socket adopted by us_socket_from_fd, or accepted by a listener created with this option, + * is registered as if us_socket_pause had been called on it, so no extra poll change is needed + * per connection (node:net's pauseOnConnect; cluster adopts every connection this way). Not + * for connects, and ignored for TLS sockets: the handshake needs the reads, the owner pauses + * those sockets itself. */ + LIBUS_SOCKET_OPEN_PAUSED = 256, }; /* Library types publicly available */ diff --git a/packages/bun-usockets/src/loop.c b/packages/bun-usockets/src/loop.c index ce1335294657..2a1203d466a8 100644 --- a/packages/bun-usockets/src/loop.c +++ b/packages/bun-usockets/src/loop.c @@ -509,7 +509,7 @@ void us_internal_dispatch_ready_poll(struct us_poll_t *p, int error, int eof, in do { struct us_poll_t *accepted_p = us_create_poll(loop, 0, sizeof(struct us_socket_t) - sizeof(struct us_poll_t) + listen_socket->socket_ext_size); us_poll_init(accepted_p, client_fd, POLL_TYPE_SOCKET); - if (us_poll_start_rc(accepted_p, loop, LIBUS_SOCKET_READABLE) != 0) { + if (us_poll_start_rc(accepted_p, loop, listen_socket->accept_paused ? 0 : LIBUS_SOCKET_READABLE) != 0) { /* EPOLL_CTL_ADD failed (e.g. ENOSPC). Close the fd so the * peer sees a RST instead of a connection that silently * never answers. */ @@ -528,7 +528,7 @@ void us_internal_dispatch_ready_poll(struct us_poll_t *p, int error, int eof, in s->long_timeout = 255; s->flags.low_prio_state = 0; s->flags.allow_half_open = listen_socket->s.flags.allow_half_open; - s->flags.is_paused = 0; + s->flags.is_paused = listen_socket->accept_paused; s->flags.is_ipc = 0; s->flags.is_closed = 0; s->flags.adopted = 0; diff --git a/packages/bun-usockets/src/socket.c b/packages/bun-usockets/src/socket.c index 8d76a1116e6e..677f9dd656f4 100644 --- a/packages/bun-usockets/src/socket.c +++ b/packages/bun-usockets/src/socket.c @@ -444,7 +444,8 @@ int us_socket_write2(struct us_socket_t *s, const char *header, int header_lengt struct us_socket_t *us_socket_from_fd(struct us_socket_group_t *group, unsigned char kind, struct ssl_ctx_st *ssl_ctx, int socket_ext_size, LIBUS_SOCKET_DESCRIPTOR fd, int options, int ipc) { struct us_poll_t *p1 = us_create_poll(group->loop, 0, sizeof(struct us_socket_t) + socket_ext_size); us_poll_init(p1, fd, POLL_TYPE_SOCKET); - int rc = us_poll_start_rc(p1, group->loop, LIBUS_SOCKET_READABLE | LIBUS_SOCKET_WRITABLE); + int open_paused = (options & LIBUS_SOCKET_OPEN_PAUSED) && !ssl_ctx; + int rc = us_poll_start_rc(p1, group->loop, (open_paused ? 0 : LIBUS_SOCKET_READABLE) | LIBUS_SOCKET_WRITABLE); if (rc != 0) { us_poll_free(p1, group->loop); return 0; @@ -458,7 +459,7 @@ struct us_socket_t *us_socket_from_fd(struct us_socket_group_t *group, unsigned s->long_timeout = 255; s->flags.low_prio_state = 0; s->flags.allow_half_open = (options & LIBUS_SOCKET_ALLOW_HALF_OPEN) != 0; - s->flags.is_paused = 0; + s->flags.is_paused = open_paused; s->flags.is_ipc = ipc; s->flags.is_closed = 0; s->flags.adopted = 0; diff --git a/src/js/node/net.ts b/src/js/node/net.ts index 1e9d83b4dda9..116fbdec2698 100644 --- a/src/js/node/net.ts +++ b/src/js/node/net.ts @@ -156,7 +156,7 @@ const kPerfHooksNetConnectContext = Symbol("kPerfHooksNetConnectContext"); const khandshakeTimer = Symbol("khandshakeTimer"); const kerrorEmitted = Symbol("kerrorEmitted"); const kUserUnrefed = Symbol("kUserUnrefed"); -// Set when pause() dropped the handle's hold on the loop, so the read paths +// Set when readStop() dropped the handle's hold on the loop, so the read paths // only restore a hold they actually removed - re-refing a handle that never // held the loop (a wrapped duplex with no fd) would pin the process. const kPausedUnref = Symbol("kPausedUnref"); @@ -413,7 +413,7 @@ const SocketHandlers: SocketHandler = { self._unrefTimer(); self.bytesRead += buffer.length; if (!self.push(buffer)) { - pauseForBackpressure(self, socket); + readStop(self, socket); } }, drain(socket) { @@ -562,17 +562,27 @@ const SocketHandlers: SocketHandler = { binaryType: "buffer", } as const; -// push() (or an onread callback) said stop. Like node's readStop, a socket that -// is not reading does not hold the loop (see Socket.prototype.pause); a write -// awaiting drain still does, and unrefAfterDrain drops the hold once it completes. -function pauseForBackpressure(self, handle) { +// Node's readStop: https://github.com/nodejs/node/blob/v26.3.0/lib/internal/stream_base_commons.js#L191-L198 +function readStop(self, handle) { handle?.pause?.(); + releaseReadHold(self, handle); +} + +// A handle that is not reading does not hold the loop; a pending write still does (see unrefAfterDrain). +function releaseReadHold(self, handle) { + // A socket over a generic duplex has no fd and never held the loop. if (self[kupgraded] && !(self[kupgraded] instanceof Socket)) return; self[kPausedUnref] = true; if (!self[kwriteCallback]) handle?.unref?.(); } -// Reads are flowing again: give back the hold a pause dropped. Cleared even when +// The handle opened paused (pauseOnConnect in its native config): https://github.com/nodejs/node/blob/v26.3.0/lib/net.js#L494-L498 +function pauseOnCreate(self, handle) { + releaseReadHold(self, handle); + self.readableFlowing = false; +} + +// Reads are flowing again: give back the hold readStop dropped. Cleared even when // the user unref'd, so a later ref() is not undone by unrefAfterDrain. function restorePausedHold(self, handle) { if (!self[kPausedUnref]) return; @@ -596,7 +606,7 @@ function finishSocketEnd(self) { // the loop while reading or with a write in flight, so node lets the process exit even if // the (half-open) writable side stays open and the readable side was never consumed. Mirror // that: drop this handle's hold on the loop unless a write is still waiting on drain, and - // forget any pause()-time unref so a later read()/resume() does not pin the loop again. + // forget any readStop-time unref so a later read()/resume() does not pin the loop again. // A subsequent buffered write re-refs (see _write) so its callback can still fire. const socket = self._handle; if (socket && !self[kwriteCallback]) { @@ -755,7 +765,7 @@ const ServerHandlers: SocketHandler = { self._unrefTimer(); self.bytesRead += buffer.length; if (!self.push(buffer)) { - pauseForBackpressure(self, socket); + readStop(self, socket); } }, keylog(socket, line) { @@ -978,9 +988,9 @@ const ServerHandlers: SocketHandler = { self.authorized = true; } } - const pauseOnConnect = server && (server.pauseOnConnect ?? server[bunSocketServerOptions]?.pauseOnConnect); + const pauseOnConnect = server?.pauseOnConnect; if (pauseOnConnect) { - self.pause(); + pauseOnCreate(self, socket); } if (!pauseOnConnect && !self.destroyed) { // https://github.com/nodejs/node/blob/v26.3.0/lib/internal/tls/wrap.js#L502-L524 @@ -1154,7 +1164,9 @@ function onconnection(err, clientHandle) { } clientHandle[kServerSocket] = handle; const options = self[bunSocketServerOptions]; - const { pauseOnConnect, connectionListener, [kSocketClass]: SClass } = options; + const { connectionListener, [kSocketClass]: SClass } = options; + // Read per connection like node; the listener itself was created with the value listen() saw. + const pauseOnConnect = self.pauseOnConnect; // Propagate the server's half-open/highWaterMark settings to the accepted // socket so the Duplex's allowHalfOpen matches what the native layer was // configured with in kRealListen; without this, net.createServer({ @@ -1224,14 +1236,11 @@ function onconnection(err, clientHandle) { _socket._server = self; if (pauseOnConnect && !isTLS) { - _socket.pause(); + pauseOnCreate(_socket, clientHandle); } - if (typeof connectionListener === "function") { - clientHandle.pauseOnConnect = pauseOnConnect; - if (!isTLS) { - self.prependOnceListener("connection", connectionListener); - } + if (typeof connectionListener === "function" && !isTLS) { + self.prependOnceListener("connection", connectionListener); } if (isTLS) initAcceptedTLSSocket(self, _socket); @@ -1263,9 +1272,6 @@ const SocketHandlers2: SocketHandler { if (!this.destroyed) { this.emit("error", error); @@ -1921,9 +1929,10 @@ Socket.prototype.connect = function connect(...args) { } }); } - this.pauseOnConnect = pauseOnConnect; if (pauseOnConnect) { - this.pause(); + // An fd is open already; a dial is paused when it opens, and afterConnect releases its hold. + if (fd != null) pauseOnCreate(this, this._handle); + else this.readableFlowing = false; } else { process.nextTick(() => { // Honor pause()/resume() calls made while connecting — only start @@ -2345,19 +2354,11 @@ Socket.prototype.resume = function resume() { return ret; }; +// Only a connected onread socket stops reading here: https://github.com/nodejs/node/blob/v26.3.0/lib/net.js#L817-L827 Socket.prototype.pause = function pause() { - if (!this.destroyed) { - this._handle?.pause?.(); - // libuv only counts a stream handle as active - and therefore as keeping - // the event loop alive - while it is reading. A paused socket lets the - // process exit; resume() re-refs it unless the user explicitly unref'd. - this._handle?.unref?.(); - // Only remember the unref when this handle can actually hold the loop: a - // TLS socket wrapped over a generic duplex has no fd, so re-refing it - // later would newly pin the process. - if (!this[kupgraded] || this[kupgraded] instanceof Socket) { - this[kPausedUnref] = true; - } + const handle = this._handle; + if (handle && this[kOnreadBuffer] !== undefined && !this.connecting && !this.destroyed) { + readStop(this, handle); } return Duplex.prototype.pause.$call(this); }; @@ -3132,6 +3133,7 @@ function internalConnect(self, options, address, port, addressType, localAddress req.localPort = localPort; req.addressType = addressType; req.tls = tls; + req.pauseOnConnect = options.pauseOnConnect; traceConnectStart(req); err = kConnectTcp(self, addressType, req, address, port); @@ -3152,6 +3154,7 @@ function internalConnect(self, options, address, port, addressType, localAddress req.address = address; req.oncomplete = afterConnect; req.tls = tls; + req.pauseOnConnect = options.pauseOnConnect; traceConnectStart(req, address); err = kConnectPipe(self, req, address); @@ -3276,6 +3279,7 @@ function internalConnectMultiple(context, canceled?) { req.localPort = localPort; req.addressType = addressType; req.tls = tls; + req.pauseOnConnect = context.options.pauseOnConnect; ArrayPrototypePush.$call(self.autoSelectFamilyAttemptedAddresses, `${address}:${port}`); @@ -3382,6 +3386,9 @@ function afterConnect(status, handle, req, readable, writable) { self._handle.setKeepAlive(true, self[kSetKeepAliveInitialDelay]); } + // Node only starts reading at the read(0) below, which a paused stream skips. TLS needs the reads to handshake. + if (self.isPaused() && !self.encrypted) readStop(self, self._handle); + self.emit("connect"); self.emit("ready"); @@ -3391,6 +3398,7 @@ function afterConnect(status, handle, req, readable, writable) { // Start the first read, or get an immediate EOF. // this doesn't actually consume any bytes, because len=0. + // https://github.com/nodejs/node/blob/v26.3.0/lib/net.js#L1695-L1696 if (readable && !self.isPaused()) self.read(0); } else { let details; @@ -3894,6 +3902,7 @@ Server.prototype[kRealListen] = function ( exclusive: exclusive || this[bunSocketServerOptions]?.exclusive || false, socket: serverHandlersFor(this), data: this, + pauseOnConnect: this.pauseOnConnect, }); // Mirror libuv uv_pipe_chmod: readableAll/writableAll relax the unix socket // file's group/other permission bits. Skipped on Windows and abstract @@ -3926,6 +3935,7 @@ Server.prototype[kRealListen] = function ( exclusive: exclusive || this[bunSocketServerOptions]?.exclusive || false, socket: serverHandlersFor(this), data: this, + pauseOnConnect: this.pauseOnConnect, }); } else { this._handle = Bun.listen({ @@ -3938,6 +3948,7 @@ Server.prototype[kRealListen] = function ( exclusive: exclusive || this[bunSocketServerOptions]?.exclusive || false, socket: serverHandlersFor(this), data: this, + pauseOnConnect: this.pauseOnConnect, }); } diff --git a/src/runtime/socket/Handlers.rs b/src/runtime/socket/Handlers.rs index 05a15ed2c91f..d18fd8cf30fe 100644 --- a/src/runtime/socket/Handlers.rs +++ b/src/runtime/socket/Handlers.rs @@ -452,6 +452,8 @@ pub struct SocketConfig { pub(crate) allow_half_open: bool, pub(crate) reuse_port: bool, pub(crate) ipv6_only: bool, + /// node:net's `pauseOnConnect` (read off the options like `localAddress`): the socket, or every accepted one, opens paused. + pub(crate) pause_on_connect: bool, } impl SocketConfig { @@ -472,6 +474,9 @@ impl SocketConfig { if self.ipv6_only { flags |= uws::LIBUS_SOCKET_IPV6_ONLY; } + if self.pause_on_connect { + flags |= uws::LIBUS_SOCKET_OPEN_PAUSED; + } flags } @@ -518,6 +523,7 @@ impl SocketConfig { allow_half_open: false, reuse_port: false, ipv6_only: false, + pause_on_connect: false, }; }; // On any `?` below, `result` drops and releases what it owns — no @@ -581,8 +587,14 @@ impl SocketConfig { global_object: &JSGlobalObject, mode: SocketMode, ) -> JsResult { + // This getter can run user JS, so it runs before from_generated creates the unrooted handlers cell. + let pause_on_connect = opts + .get_truthy(global_object, "pauseOnConnect")? + .is_some_and(|v| v.to_boolean()); let generated = GeneratedSocketConfig::from_js(global_object, opts)?; - Self::from_generated(vm, global_object, &generated, mode) + let mut config = Self::from_generated(vm, global_object, &generated, mode)?; + config.pause_on_connect = pause_on_connect; + Ok(config) } } diff --git a/src/runtime/socket/Listener.rs b/src/runtime/socket/Listener.rs index d36d771db941..d58718b4343a 100644 --- a/src/runtime/socket/Listener.rs +++ b/src/runtime/socket/Listener.rs @@ -89,6 +89,8 @@ pub struct Listener { pub(crate) ssl: bool, pub(crate) protos: Option>, pub(crate) reject_unauthorized: bool, + /// Accepted sockets carry `Flags::PAUSE_ON_CONNECT` (see `NewSocket::on_open`). + pub(crate) pause_on_connect: bool, pub(crate) strong_data: JsCell, /// Reference to this listener's JS wrapper. Strong while it is listening or /// has connections, downgraded to weak once idle so GC can reclaim it. @@ -188,6 +190,7 @@ impl Listener { let port = socket_config.port; let ssl_enabled = socket_config.ssl.is_some(); let socket_flags = socket_config.socket_flags(); + let pause_on_connect = socket_config.pause_on_connect; #[cfg(windows)] if port.is_none() { @@ -229,6 +232,7 @@ impl Listener { ssl_cfg_taken.as_ref(), true, ), + pause_on_connect, poll_ref: JsCell::new(KeepAlive::init()), group: JsCell::new(uws::SocketGroup::default()), secure_ctx: Cell::new(None), @@ -356,6 +360,7 @@ impl Listener { ssl_cfg_taken.as_ref(), true, ), + pause_on_connect, listener: Cell::new(ListenerType::None), poll_ref: JsCell::new(KeepAlive::init()), group: JsCell::new(uws::SocketGroup::default()), @@ -603,11 +608,10 @@ impl Listener { // `OWNED_PROTOS` stays unset: accepted sockets clone the listener's `protos`. fn accepted_socket_flags(&self) -> SocketFlags { - if self.reject_unauthorized { - SocketFlags::REJECT_UNAUTHORIZED - } else { - SocketFlags::empty() - } + let mut flags = SocketFlags::empty(); + flags.set(SocketFlags::REJECT_UNAUTHORIZED, self.reject_unauthorized); + flags.set(SocketFlags::PAUSE_ON_CONNECT, self.pause_on_connect); + flags } #[cfg(windows)] @@ -1270,6 +1274,14 @@ impl Listener { ssl_taken.as_ref(), false, )); + { + let mut f = tls_ref.flags.get(); + f.set( + SocketFlags::PAUSE_ON_CONNECT, + socket_config.pause_on_connect, + ); + tls_ref.flags.set(f); + } TLSSocket::data_set_cached( tls_ref.get_this_value(global), global, @@ -1351,6 +1363,14 @@ impl Listener { }) }; let tcp_ref = tcp; + { + let mut f = tcp_ref.flags.get(); + f.set( + SocketFlags::PAUSE_ON_CONNECT, + socket_config.pause_on_connect, + ); + tcp_ref.flags.set(f); + } tcp_ref.ref_(); TCPSocket::data_set_cached( tcp_ref.get_this_value(global), @@ -1420,6 +1440,7 @@ impl Listener { default_data.ensure_still_alive(); let allow_half_open = socket_config.allow_half_open; + let pause_on_connect = socket_config.pause_on_connect; let mut ssl_taken = socket_config.ssl.take(); let promise = jsc::JSPromise::create(global); @@ -1442,6 +1463,7 @@ impl Listener { owned_ssl_ctx, default_data, allow_half_open, + pause_on_connect, port, promise_value, ) @@ -1456,6 +1478,7 @@ impl Listener { owned_ssl_ctx, default_data, allow_half_open, + pause_on_connect, port, promise_value, ) @@ -1537,6 +1560,7 @@ fn connect_finish( owned_ssl_ctx: Option>, default_data: JSValue, allow_half_open: bool, + pause_on_connect: bool, port: Option, promise_value: JSValue, ) -> JsResult { @@ -1617,6 +1641,7 @@ fn connect_finish( { let mut f = socket_ref.flags.get(); f.set(SocketFlags::ALLOW_HALF_OPEN, allow_half_open); + f.set(SocketFlags::PAUSE_ON_CONNECT, pause_on_connect); socket_ref.flags.set(f); } // Note: `do_connect` reads `self.connection` directly so no second diff --git a/src/runtime/socket/socket_body.rs b/src/runtime/socket/socket_body.rs index 40d5befa04f1..27858a21099e 100644 --- a/src/runtime/socket/socket_body.rs +++ b/src/runtime/socket/socket_body.rs @@ -667,6 +667,12 @@ impl NewSocket { self.socket.set(SocketHandler::::from(s)); } Some(UnixOrHost::Fd(f)) => { + // The fd is connected, so it is registered paused outright; a dial is paused in on_open. + let flags = if self.flags.get().contains(Flags::PAUSE_ON_CONNECT) { + flags | uws::LIBUS_SOCKET_OPEN_PAUSED + } else { + flags + }; // `LIBUS_SOCKET_DESCRIPTOR` is `c_int` on POSIX, `SOCKET` // (`usize`) on Windows; `Fd::native()` is `c_int` / HANDLE // (`*mut c_void`) respectively; cast to bridge the Rust-side `usize` alias. @@ -729,13 +735,25 @@ impl NewSocket { if this.flags.get().contains(Flags::BYPASS_TLS) { return Ok(JSValue::UNDEFINED); } - if !this.flags.get().contains(Flags::IS_PAUSED) { - let paused = this.socket.get().pause_stream(); - this.update_flags(|f| f.set(Flags::IS_PAUSED, paused)); - } + this.pause_reads(); Ok(JSValue::UNDEFINED) } + fn pause_reads(&self) { + if !self.flags.get().contains(Flags::IS_PAUSED) { + let paused = self.socket.get().pause_stream(); + self.update_flags(|f| f.set(Flags::IS_PAUSED, 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) { + self.update_flags(|f| f.remove(Flags::PAUSE_ON_CONNECT)); + self.pause_reads(); + } + } + #[bun_jsc::host_fn(method)] pub(crate) fn set_keep_alive( this: &Self, @@ -1423,6 +1441,11 @@ impl NewSocket { // update the internal socket instance to the one that was just connected // This socket must be replaced because the previous one is a connecting socket not a uSockets socket this.socket.set(socket); + // A reused wrapper still carries the previous connection's paused state; this one opens reading. + this.update_flags(|f| f.remove(Flags::IS_PAUSED)); + if !SSL { + this.pause_on_connect_once(); + } jsc::mark_binding!(); // Add SNI support for TLS (mongodb and others requires this) @@ -1818,6 +1841,8 @@ impl NewSocket { if let Some(twin) = this.twin.get().as_ref() { twin.update_flags(|f| f.insert(Flags::REJECTED)); } + } else if SSL { + this.pause_on_connect_once(); } let mut callback = handlers.on_handshake(); @@ -4158,6 +4183,8 @@ bitflags::bitflags! { /// even though its `Handlers` mode is `Client` (no listener), so the /// client-only server-identity check must not run against its peer. const TLS_SERVER_ROLE = 1 << 14; + /// node:net's `pauseOnConnect`: paused in `on_open` (plain; free for a socket that was registered with `LIBUS_SOCKET_OPEN_PAUSED`) or after the handshake (TLS). + const PAUSE_ON_CONNECT = 1 << 15; } } diff --git a/src/uws/lib.rs b/src/uws/lib.rs index 459d36949323..55ec009a303c 100644 --- a/src/uws/lib.rs +++ b/src/uws/lib.rs @@ -85,6 +85,7 @@ pub const DEDICATED_COMPRESSOR: i32 = 248; pub use bun_uws_sys::{ LIBUS_LISTEN_DEFAULT, LIBUS_LISTEN_EXCLUSIVE_PORT, LIBUS_LISTEN_REUSE_ADDR, LIBUS_LISTEN_REUSE_PORT, LIBUS_SOCKET_ALLOW_HALF_OPEN, LIBUS_SOCKET_IPV6_ONLY, + LIBUS_SOCKET_OPEN_PAUSED, }; // Re-export the `_sys` definitions so higher tiers see one type. `to_js` diff --git a/src/uws_sys/lib.rs b/src/uws_sys/lib.rs index fac78a2b0d41..3f08bbcf55f0 100644 --- a/src/uws_sys/lib.rs +++ b/src/uws_sys/lib.rs @@ -30,6 +30,7 @@ pub const LIBUS_LISTEN_REUSE_PORT: core::ffi::c_int = 4; pub const LIBUS_SOCKET_IPV6_ONLY: core::ffi::c_int = 8; pub const LIBUS_LISTEN_REUSE_ADDR: core::ffi::c_int = 16; pub const LIBUS_LISTEN_DISALLOW_REUSE_PORT_FAILURE: core::ffi::c_int = 32; +pub const LIBUS_SOCKET_OPEN_PAUSED: core::ffi::c_int = 256; /// BoringSSL `SSL_CTX` (alias so callers don't need a direct boringssl dep). pub type SslCtx = bun_boringssl_sys::SSL_CTX; diff --git a/test/js/bun/net/socket-syscall-fault.test.ts b/test/js/bun/net/socket-syscall-fault.test.ts index a69ac3de1fa8..d746c723fbfa 100644 --- a/test/js/bun/net/socket-syscall-fault.test.ts +++ b/test/js/bun/net/socket-syscall-fault.test.ts @@ -100,7 +100,9 @@ describe.skipIf(!fault.available())("poll_start failure is reported, not a crash // (EPOLLHUP is level-triggered and cannot be masked) and registered again by // resume(), which is a fresh EPOLL_CTL_ADD and can fail the way the first one // can. epoll only: kqueue and libuv never park the fd, so their resume is a -// plain filter/poll change with nothing for the hook to fail. +// plain filter/poll change with nothing for the hook to fail. onread mode, because +// like in node only that mode's pause() stops the handle (a plain pause() keeps +// reading into the stream's buffer, which would deliver the reply as data here). test.skipIf(!fault.available() || !isLinux)( "resume() of a parked socket that cannot be registered again fails the socket instead of leaving it deaf", async () => { @@ -120,7 +122,8 @@ test.skipIf(!fault.available() || !isLinux)( s.on("close", resolveClosed); }); server.listen(0, async () => { - const conn = net.connect({ port: server.address().port, allowHalfOpen: true }); + const onread = { buffer: Buffer.alloc(4096), callback: () => console.log("data") }; + const conn = net.connect({ port: server.address().port, allowHalfOpen: true, onread }); await once(conn, "connect"); conn.end(); conn.pause(); @@ -130,7 +133,6 @@ test.skipIf(!fault.available() || !isLinux)( // the next poll phase reports the hangup and parks the socket; an // immediate runs after that phase. await new Promise(r => setImmediate(r)); - conn.on("data", () => console.log("data")); conn.on("error", e => console.log("error", e.code, e.syscall)); conn.on("end", () => console.log("end")); conn.on("close", hadError => console.log("close", hadError)); diff --git a/test/js/node/cluster.test.ts b/test/js/node/cluster.test.ts index 64d101f15e8f..9266fd4cf610 100644 --- a/test/js/node/cluster.test.ts +++ b/test/js/node/cluster.test.ts @@ -1039,16 +1039,22 @@ if (cluster.isPrimary) { const worker = cluster.fork(); worker.on("message", m => { console.log(JSON.stringify(m)); worker.kill(); process.exit(0); }); cluster.on("listening", (_w, addr) => { - const c = net.connect(addr.port, "127.0.0.1", () => c.write("early")); + // "early" is in the worker's receive buffer before the write callback runs, so a + // report taken after this message proves the paused socket did not read it. + const c = net.connect(addr.port, "127.0.0.1", () => c.write("early", () => worker.send("written"))); c.on("error", () => {}); }); } else { - const server = net.createServer({ pauseOnConnect: true }, sock => { - let earlyData = false; - sock.once("data", () => { earlyData = true; }); - setImmediate(() => { - process.send({ paused: sock.isPaused(), earlyData, _server: sock._server === server }); - }); + let sock, written = false, earlyData = false; + const report = () => { + if (!sock || !written) return; + process.send({ paused: sock.isPaused(), bytesRead: sock.bytesRead, earlyData, _server: sock._server === server }); + }; + process.on("message", () => { written = true; report(); }); + const server = net.createServer({ pauseOnConnect: true }, s => { + sock = s; + s.once("data", () => { earlyData = true; }); + report(); }); server.listen(0, "127.0.0.1"); } @@ -1063,7 +1069,7 @@ if (cluster.isPrimary) { }); const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); expect({ out: JSON.parse(stdout.trim()), stderr }).toEqual({ - out: { paused: true, earlyData: false, _server: true }, + out: { paused: true, bytesRead: 0, earlyData: false, _server: true }, stderr: expect.any(String), }); expect(exitCode).toBe(0); diff --git a/test/js/node/net/node-net.test.ts b/test/js/node/net/node-net.test.ts index eadca761d75b..55bc6bb9f758 100644 --- a/test/js/node/net/node-net.test.ts +++ b/test/js/node/net/node-net.test.ts @@ -1374,7 +1374,9 @@ describe("paused socket whose peer sends RST", () => { // Regression: on Linux, epoll forwarded the raw EPOLLERR bit (8) as a libus // close code, which the JS error path read as errno 8 and surfaced as a // bogus `Error: read ENOEXEC` when the socket was not actively reading. - // kqueue already normalized the flag to 0/1. + // kqueue already normalized the flag to 0/1. Like node, only an onread + // socket's pause() stops the handle (a plain pause() keeps it reading), so + // onread mode is what puts the socket into the not-reading state here. it("does not surface a bogus errno error", async () => { const { promise, resolve } = Promise.withResolvers(); const errors: NodeJS.ErrnoException[] = []; @@ -1386,12 +1388,14 @@ describe("paused socket whose peer sends RST", () => { try { await new Promise(r => server.listen(0, "127.0.0.1", r)); const port = (server.address() as import("node:net").AddressInfo).port; - const c = connect(port, "127.0.0.1", () => { - c.pause(); - c.write("x"); - }); + const c = connect({ port, host: "127.0.0.1", onread: { buffer: Buffer.alloc(16), callback: () => {} } }); c.on("error", e => errors.push(e)); c.on("close", () => resolve()); + await once(c, "connect"); + // Every 'connect' listener has run by now, including the _read() that + // connect() queued, so nothing starts the handle again after this pause. + c.pause(); + c.write("x"); await promise; } finally { server.close(); @@ -1400,6 +1404,118 @@ describe("paused socket whose peer sends RST", () => { }); }); +// Node stops kernel reads when push() returns false, not on pause(): a paused +// socket keeps reading into its buffer, so it still sees the peer's FIN. +// https://github.com/nodejs/node/blob/v26.3.0/lib/net.js#L817-L827 +// Stopping the handle on pause() left the FIN unread forever, so a socket that +// nothing resumes (unpipe() pauses it) never emitted 'end' or 'close' and its +// server's close() never called back. +describe.concurrent("paused socket whose peer ends", () => { + // Pauses the accepted socket after the accept-time read(0), with nothing + // pushed afterwards that could start the handle again, then the peer ends. + async function pauseThenPeerEnds(endBeforePeer: boolean) { + const server = createServer(); + const accepted = Promise.withResolvers(); + server.on("connection", accepted.resolve); + await once(server.listen(0, "127.0.0.1"), "listening"); + const serverClosed = Promise.withResolvers(); + try { + const client = connect((server.address() as import("node:net").AddressInfo).port, "127.0.0.1"); + const clientClosed = once(client, "close"); + const [socket] = await Promise.all([accepted.promise, once(client, "connect")]); + const events: string[] = []; + const closed = Promise.withResolvers(); + socket.on("end", () => events.push("end")); + socket.on("close", hadError => { + events.push(`close hadError=${hadError}`); + closed.resolve(); + }); + socket.on("error", closed.reject); + socket.pause(); + if (endBeforePeer) socket.end(); + expect(socket.isPaused()).toBe(true); + client.end(); + await Promise.all([closed.promise, clientClosed]); + expect(events).toEqual(["end", "close hadError=false"]); + } finally { + server.close(err => (err ? serverClosed.reject(err) : serverClosed.resolve())); + } + // The connection is gone, so close() calls back. + await serverClosed.promise; + } + + it("still emits 'end' and 'close'", () => pauseThenPeerEnds(false)); + + it("still emits 'end' and 'close' after it end()ed first", () => pauseThenPeerEnds(true)); +}); + +describe.concurrent("pauseOnConnect", () => { + // The peer's bytes are queued on `socket` once the write callback ran; the poll + // between the two immediates is when a handle that reads would consume them. + async function expectNothingRead(socket: Socket, peer: Socket, bytes: string) { + await new Promise(resolve => peer.write(bytes, resolve)); + await new Promise(resolve => setImmediate(() => setImmediate(resolve))); + expect({ paused: socket.isPaused(), bytesRead: socket.bytesRead }).toEqual({ paused: true, bytesRead: 0 }); + const data = once(socket, "data"); + socket.resume(); + expect(String((await data)[0])).toBe(bytes); + } + + it("reads server.pauseOnConnect per connection, like node", async () => { + const server = createServer(); + server.pauseOnConnect = true; + const accepted = Promise.withResolvers(); + server.on("connection", accepted.resolve); + await once(server.listen(0, "127.0.0.1"), "listening"); + try { + const client = connect((server.address() as import("node:net").AddressInfo).port, "127.0.0.1"); + const [socket] = await Promise.all([accepted.promise, once(client, "connect")]); + await expectNothingRead(socket, client, "early"); + client.destroy(); + await once(socket, "close"); + } finally { + server.close(); + } + }); + + it("applies to a dialed socket", async () => { + const accepted = Promise.withResolvers(); + const server = createServer(accepted.resolve); + await once(server.listen(0, "127.0.0.1"), "listening"); + try { + const port = (server.address() as import("node:net").AddressInfo).port; + const client = connect({ port, host: "127.0.0.1", pauseOnConnect: true } as import("node:net").NetConnectOpts); + const [socket] = await Promise.all([accepted.promise, once(client, "connect")]); + await expectNothingRead(client, socket, "reply"); + socket.destroy(); + await once(client, "close"); + } finally { + server.close(); + } + }); + + // A socket that opened paused must notice its peer going away like a socket that + // pause()d later does, on every backend (kqueue keeps a read knote for that). + it("still reports a peer reset before resume()", async () => { + const server = createServer({ pauseOnConnect: true }); + const accepted = Promise.withResolvers(); + server.on("connection", accepted.resolve); + await once(server.listen(0, "127.0.0.1"), "listening"); + try { + const client = connect((server.address() as import("node:net").AddressInfo).port, "127.0.0.1"); + const [socket] = await Promise.all([accepted.promise, once(client, "connect")]); + const errors: NodeJS.ErrnoException[] = []; + socket.on("error", e => errors.push(e)); + const closed = new Promise(resolve => socket.on("close", resolve)); + client.resetAndDestroy(); + await closed; + expect(errors.map(e => e.code).filter(code => code !== "ECONNRESET")).toEqual([]); + } finally { + server.close(); + } + }); +}); + describe("net.Server accepted-socket buffering", () => { it("delivers bytes buffered before a 'readable' listener attaches, past peer FIN", async () => { // read(0) instead of resume(): bytes that arrive before the connection diff --git a/test/js/node/net/socket-reconnect-live.test.ts b/test/js/node/net/socket-reconnect-live.test.ts index 923e6ca2595c..f83ed929e4e3 100644 --- a/test/js/node/net/socket-reconnect-live.test.ts +++ b/test/js/node/net/socket-reconnect-live.test.ts @@ -37,6 +37,68 @@ describe.concurrent("socket.connect() on an already-connected socket", () => { expect({ stdout, stderr, exitCode }).toEqual({ stdout: "first second", stderr, exitCode: 0 }); }); + it("pauses the new connection although the previous one was left paused", async () => { + // The wrapper is reused, so the paused state recorded for the first connection must + // not turn the pause of the second one (a paused stream stops its new handle in + // afterConnect) into a no-op. + await using proc = Bun.spawn({ + cmd: [ + bunExe(), + "-e", + ` + const { createServer, connect } = require("node:net"); + const { once } = require("node:events"); + let connections = 0; + const wrote = Promise.withResolvers(); + const srv = createServer(c => { + c.on("error", () => {}); + if (++connections === 2) c.end("hello", wrote.resolve); + }); + srv.listen(0, "127.0.0.1", async () => { + const port = srv.address().port; + let resumed = false; + let deliveredWhilePaused = false; + let received = ""; + const onread = { + buffer: Buffer.alloc(64), + callback(n, buf) { + if (!resumed) deliveredWhilePaused = true; + received += buf.toString("utf8", 0, n); + if (received === "hello") { + console.log(JSON.stringify({ deliveredWhilePaused, received })); + s.destroy(); + srv.close(); + } + }, + }; + const s = connect({ port, host: "127.0.0.1", onread }); + await once(s, "connect"); + // Connected and reading: this pause stops the first connection's handle. + s.pause(); + s.connect({ port, host: "127.0.0.1" }); + await once(s, "connect"); + // "hello" is queued on the second connection once the server's end() callback + // ran. The loop polls once between the two immediates, so a handle that still + // reads delivers it (to the callback above, while !resumed) before resume(). + await wrote.promise; + await new Promise(done => setImmediate(() => setImmediate(done))); + resumed = true; + s.resume(); + }); + `, + ], + 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(), stderr, exitCode }).toEqual({ + stdout: JSON.stringify({ deliveredWhilePaused: false, received: "hello" }), + stderr, + exitCode: 0, + }); + }); + it("does not crash when reconnecting while the first connect is still in flight", async () => { await using proc = Bun.spawn({ cmd: [ diff --git a/test/js/node/tls/node-tls-connect.test.ts b/test/js/node/tls/node-tls-connect.test.ts index 2250aad54dbe..d327c30cacdb 100644 --- a/test/js/node/tls/node-tls-connect.test.ts +++ b/test/js/node/tls/node-tls-connect.test.ts @@ -1614,3 +1614,57 @@ describe("throwing 'secureConnect' listener", () => { expect(exitCode).toBe(0); }); }); + +describe.concurrent("a client paused while connecting", () => { + // Node starts the TLS handle's reads from its own read(0) whatever the stream's + // paused state, and pause() while connecting never readStops, onread mode + // included (lib/net.js pause). So the handshake completes, and the reading + // handle is what keeps the process alive until then: everything on the server + // side is unref'd. Verified against node v26.3.0 for both variants. + it.each([ + ["plain", ""], + ["onread", "onread: { buffer: Buffer.alloc(1024), callback() {} },"], + ])("%s: still completes the handshake and holds the process until then", async (_mode, onreadOption) => { + const script = ` + const tlsMod = require("node:tls"); + const server = tlsMod.createServer(${JSON.stringify(COMMON_CERT_)}, function onConn(socket) { + socket.unref(); + }); + server.listen(0, "127.0.0.1", function onListen() { + server.unref(); + const client = tlsMod.connect({ + port: server.address().port, + host: "127.0.0.1", + rejectUnauthorized: false, + ${onreadOption} + }); + client.pause(); + client.on("secureConnect", function onSecureConnect() { + console.log("secureConnect paused=" + client.isPaused()); + client.destroy(); + server.close(); + }); + client.on("error", function onError(err) { + console.log("error " + (err.code || err.message)); + }); + }); + process.on("exit", function onExit(code) { + console.log("exit " + code); + }); + `; + 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]); + // Debug builds may write benign diagnostics to stderr, so it is only shown + // when the fixture failed (the convention of the fixtures above). + expect({ stdout: stdout.trim().split("\n"), exitCode, failureDetail: exitCode === 0 ? "" : stderr }).toEqual({ + stdout: ["secureConnect paused=true", "exit 0"], + exitCode: 0, + failureDetail: "", + }); + }); +}); diff --git a/test/js/node/tls/node-tls-server.test.ts b/test/js/node/tls/node-tls-server.test.ts index 32f369386c1c..5108bc58cee2 100644 --- a/test/js/node/tls/node-tls-server.test.ts +++ b/test/js/node/tls/node-tls-server.test.ts @@ -2563,3 +2563,105 @@ describe.each(["tls", "net"])("%s server socket whose peer resets the connection if (!isWindows) expect(t.bytesRead()).toBeGreaterThan(0); }); }); + +describe.each(["tls", "net"])("%s server socket that unpipe() paused after it end()ed", transport => { + // A front server pipes each accepted socket to an upstream connection and back. + // The upstream replies and closes: inner.pipe(sock) end()s sock, and the cleanup + // of sock.pipe(inner) unpipes sock, which pauses it with nothing left to resume + // it. The peer then closes. Node delivers 'end' and 'close' to the paused + // socket, so the connection count drops and server.close() calls back. + it("still emits 'end' and 'close' when the peer closes, and server.close() completes", async () => { + const upstream = net.createServer(s => s.on("data", () => s.end("reply"))); + await once(upstream.listen(0, "127.0.0.1"), "listening"); + const serverEvents: string[] = []; + const serverSocketClosed = Promise.withResolvers(); + const onConnection = (sock: net.Socket) => { + const inner = net.connect((upstream.address() as AddressInfo).port, "127.0.0.1", () => { + sock.pipe(inner); + inner.pipe(sock); + }); + inner.on("close", () => serverEvents.push(`upstream closed, paused=${sock.isPaused()}`)); + sock.on("end", () => serverEvents.push("end")); + sock.on("close", hadError => { + serverEvents.push(`close hadError=${hadError}`); + serverSocketClosed.resolve(); + }); + sock.on("error", serverSocketClosed.reject); + }; + const front = transport === "tls" ? createServer(COMMON_CERT, onConnection) : net.createServer(onConnection); + await once(front.listen(0, "127.0.0.1"), "listening"); + const frontClosed = Promise.withResolvers(); + const upstreamClosed = Promise.withResolvers(); + try { + const port = (front.address() as AddressInfo).port; + const clientEvents: string[] = []; + const clientClosed = Promise.withResolvers(); + const onConnect = () => client.write("hi"); + const client: net.Socket = + transport === "tls" + ? connect({ port, host: "127.0.0.1", rejectUnauthorized: false }, onConnect) + : net.connect({ port, host: "127.0.0.1" }, onConnect); + client.setEncoding("utf8"); + client.on("data", chunk => clientEvents.push(`data ${chunk}`)); + client.on("end", () => clientEvents.push("end")); + client.on("close", () => { + clientEvents.push("close"); + clientClosed.resolve(); + }); + client.on("error", clientClosed.reject); + await Promise.all([clientClosed.promise, serverSocketClosed.promise]); + expect({ clientEvents, serverEvents }).toEqual({ + clientEvents: ["data reply", "end", "close"], + serverEvents: ["upstream closed, paused=true", "end", "close hadError=false"], + }); + } finally { + front.close(err => (err ? frontClosed.reject(err) : frontClosed.resolve())); + upstream.close(err => (err ? upstreamClosed.reject(err) : upstreamClosed.resolve())); + } + // Both connections are gone, so both servers call back. + await Promise.all([frontClosed.promise, upstreamClosed.promise]); + }); +}); + +describe("pauseOnConnect", () => { + it("hands out the accepted socket paused after the handshake and reads nothing until resume()", async () => { + const server = createServer({ ...COMMON_CERT, pauseOnConnect: true }); + const accepted = Promise.withResolvers(); + server.on("secureConnection", accepted.resolve); + server.on("tlsClientError", accepted.reject); + await once(server.listen(0, "127.0.0.1"), "listening"); + // A second connection is the barrier: its bytes reach this process after the + // client's "hello" reached the paused socket's receive buffer. + const probe = net.createServer(); + const probed = Promise.withResolvers(); + probe.on("connection", socket => socket.once("data", () => probed.resolve())); + await once(probe.listen(0, "127.0.0.1"), "listening"); + const client = connect({ + port: (server.address() as AddressInfo).port, + host: "127.0.0.1", + rejectUnauthorized: false, + }); + const clientSecure = once(client, "secureConnect"); + try { + const socket = await accepted.promise; + expect(socket.isPaused()).toBe(true); + await clientSecure; + await new Promise((resolve, reject) => client.write("hello", err => (err ? reject(err) : resolve()))); + const probeClient = net.connect((probe.address() as AddressInfo).port, "127.0.0.1", () => probeClient.end("x")); + await probed.promise; + // Node's TLSWrap reads ahead here (bytesRead would be 5); ours leaves the bytes in the kernel. + expect({ paused: socket.isPaused(), bytesRead: socket.bytesRead }).toEqual({ paused: true, bytesRead: 0 }); + const received = once(socket, "data"); + socket.resume(); + const [chunk] = await received; + expect(String(chunk)).toBe("hello"); + const clientClosed = once(client, "close"); + socket.end(); + client.end(); + await clientClosed; + } finally { + server.close(); + probe.close(); + } + }); +}); diff --git a/test/js/node/tls/renegotiation.test.ts b/test/js/node/tls/renegotiation.test.ts index 3a96abf3c59b..a84e2fe55621 100644 --- a/test/js/node/tls/renegotiation.test.ts +++ b/test/js/node/tls/renegotiation.test.ts @@ -131,6 +131,71 @@ it("allow renegotiation in tls module", async () => { expect(body).toBe("Hello World"); }); +it("pauseOnConnect acts on the first handshake only, not on a renegotiation", async () => { + // The client reports a renegotiation through a second handshake callback when the + // first application data after it arrives ("first", still delivered by that same + // read). A client that pauseOnConnect paused again there would never read "second", + // which the server only sends once the client acknowledged "first". + await using server = Bun.spawn({ + cmd: [ + "node", + "-e", + ` + const tls = require("tls"); + const server = tls.createServer( + { cert: process.env.SERVER_CERT, key: process.env.SERVER_KEY, minVersion: "TLSv1.2", maxVersion: "TLSv1.2" }, + socket => { + socket.on("error", () => {}); + socket.on("data", () => socket.end("second")); + socket.renegotiate({ rejectUnauthorized: false }, err => { + if (err) socket.destroy(err); + else socket.write("first"); + }); + }, + ); + server.listen(0, "127.0.0.1", () => console.log(server.address().port)); + `, + ], + stdout: "pipe", + stderr: "inherit", + stdin: "ignore", + env: { ...bunEnv, SERVER_CERT: tls.cert, SERVER_KEY: tls.key }, + }); + const { value } = await server.stdout.getReader().read(); + const port = Number(new TextDecoder().decode(value).trim()); + + const outcome = Promise.withResolvers<{ handshakes: number; received: string }>(); + let handshakes = 0; + let received = ""; + const socket = await Bun.connect({ + hostname: "127.0.0.1", + port, + tls: { rejectUnauthorized: false }, + pauseOnConnect: true, + socket: { + handshake(socket) { + if (++handshakes === 1) socket.resume(); + }, + data(socket, chunk) { + received += chunk.toString(); + if (received === "first") socket.write("ack"); + else if (received === "firstsecond") outcome.resolve({ handshakes, received }); + }, + error(_socket, error) { + outcome.reject(error); + }, + close() { + outcome.reject(new Error(`closed after ${handshakes} handshake(s) with ${JSON.stringify(received)}`)); + }, + }, + }); + try { + expect(await outcome.promise).toEqual({ handshakes: 2, received: "firstsecond" }); + } finally { + socket.end(); + } +}); + it("should not crash when socket is closed inside the renegotiation handshake callback", async () => { // When a TLS 1.2 server initiates renegotiation and then sends application data, the // client-side SSL_read loop fires the on_handshake callback once the renegotiated