diff --git a/src/js/node/net.ts b/src/js/node/net.ts index d7bb3f5e7b22..5b3ad31da3c6 100644 --- a/src/js/node/net.ts +++ b/src/js/node/net.ts @@ -1925,6 +1925,11 @@ function kConnectDispatch(self, req, opts) { req.dispatching = true; const promise = doConnect(self._handle, opts); req.dispatching = false; + // A handle that refuses the connect dials nothing: doConnect returns the errno. + if (typeof promise === "number") { + traceConnectEnd(req); + return promise; + } promise.catch(_reason => { // eat this so there's no unhandledRejection // we already catch this in connectError and error @@ -2310,7 +2315,7 @@ Socket.prototype.connect = function connect(...args) { connection = socket; } if (fd != null) { - doConnect(this._handle, { + const opened = doConnect(this._handle, { data: this, fd: fd, ...(options.fdIsRawSocket === true ? { fdIsRawSocket: true } : {}), @@ -2318,7 +2323,13 @@ Socket.prototype.connect = function connect(...args) { // Always half-open natively; see kConnect. allowHalfOpen: true, pauseOnConnect, - }).catch(error => { + }); + // libuv opens an fd only on a handle that has none. + if (typeof opened === "number") { + this.destroy(new ErrnoException(opened, "open")); + return this; + } + opened.catch(error => { if (!this.destroyed) { this.emit("error", error); this.emit("close", true); @@ -3359,6 +3370,8 @@ function internalConnect(self, options, address, port?, addressType?, localAddre addressType, ); + // A handle that still owns a socket is bound already: libuv reports the EINVAL of its bind(2). + if (ownsSocket(self._handle)) err = uv().UV_EINVAL; err = checkBindError(err, localPort, self._handle); if (err) { const ex = new ExceptionWithHostPort(err, "bind", localAddress, localPort); @@ -3413,11 +3426,22 @@ function internalConnect(self, options, address, port?, addressType?, localAddre } if (err) { - const ex = new ExceptionWithHostPort(err, "connect", address, port); + const ex = new ExceptionWithHostPort(err, "connect", address, port, boundAddress(self._handle)); self.destroy(ex); } } +// Whether a native handle owns a socket: one that is connecting, connected, or not closed yet. +function ownsSocket(handle) { + return handle != null && handle.readyState !== -1; +} + +// The "- Local (ip:port)" detail: https://github.com/nodejs/node/blob/v26.3.0/lib/net.js#L1161-L1170 +function boundAddress(handle) { + const address = handle?.localAddress; + if (address) return address + ":" + handle.localPort; +} + function internalConnectMultiple(context, canceled?) { clearTimeout(context[kTimeout]); const self = context.socket; @@ -3443,6 +3467,11 @@ function internalConnectMultiple(context, canceled?) { const current = context.current++; if (current > 0) { + // Node closes this socket with no event and dials on: https://github.com/nodejs/node/blob/v26.3.0/lib/net.js#L1201-L1203 + if (ownsSocket(self._handle)) { + self.destroy(context.errors[context.errors.length - 1]); + return; + } self[kReinitializeHandle](newDetachedSocket(typeof self[bunTlsSymbol] === "function")); } @@ -3470,6 +3499,8 @@ function internalConnectMultiple(context, canceled?) { addressType, ); + // As in internalConnect. + if (ownsSocket(self._handle)) err = uv().UV_EINVAL; err = checkBindError(err, localPort, self._handle); if (err) { ArrayPrototypePush.$call(context.errors, new ExceptionWithHostPort(err, "bind", localAddress, localPort)); @@ -3508,7 +3539,7 @@ function internalConnectMultiple(context, canceled?) { err = kConnectTcp(self, addressType, req, address, port); if (err) { - const ex = new ExceptionWithHostPort(err, "connect", address, port); + const ex = new ExceptionWithHostPort(err, "connect", address, port, boundAddress(self._handle)); ArrayPrototypePush.$call(context.errors, ex); self.emit("connectionAttemptFailed", address, port, addressType, ex); diff --git a/src/runtime/socket/Listener.rs b/src/runtime/socket/Listener.rs index d06a5dd0a3bb..21c11ed5ffce 100644 --- a/src/runtime/socket/Listener.rs +++ b/src/runtime/socket/Listener.rs @@ -1211,6 +1211,22 @@ impl Listener { Some((local_addr_bytes.to_vec().into_boxed_slice(), local_port)) }; + // Past the last option getter: script cannot give the handle a socket after this check. + let refusal = match (prev_maybe_tcp, prev_maybe_tls) { + // SAFETY: the caller passes a live socket, owned by its JS wrapper. + (Some(prev), _) => unsafe { &*prev }.connect_refusal(&connection), + // SAFETY: as above. + (_, Some(prev)) => unsafe { &*prev }.connect_refusal(&connection), + _ => None, + }; + if let Some(errno) = refusal { + #[cfg(windows)] + let uv_errno = uv::e_discriminant_to_uv(errno as u16).unwrap_or(uv::UV_EISCONN); + #[cfg(not(windows))] + let uv_errno = -(errno as c_int); + return Ok(JSValue::js_number_from_int32(uv_errno)); + } + #[cfg(windows)] let mut connection = connection; #[cfg(windows)] diff --git a/src/runtime/socket/socket_body.rs b/src/runtime/socket/socket_body.rs index 64522212099d..767f8c0296b9 100644 --- a/src/runtime/socket/socket_body.rs +++ b/src/runtime/socket/socket_body.rs @@ -1388,6 +1388,53 @@ impl NewSocket { socket.close(code); } + /// The errno that node's handle answers to this `connect()` while it owns a socket. `None`: dial. + pub(crate) fn connect_refusal( + &self, + connection: &super::listener::UnixOrHost, + ) -> Option { + use super::listener::UnixOrHost; + use sys::SystemErrno as E; + let socket = self.socket.get(); + if socket.is_detached() { + return None; + } + // libuv opens an fd only on a handle that has none. + if matches!(connection, UnixOrHost::Fd(_)) { + return Some(E::EBUSY); + } + let is_tcp = matches!( + socket.local_address(&mut [0u8; 64]).map(<[u8]>::len), + Some(4 | 16) + ); + if !is_tcp { + // A pipe handle reports what connect(2) says for the new path: EISCONN when it is a socket. + return Some(E::EISCONN); + } + // Node parses the path as an IPv4 address first: https://github.com/nodejs/node/blob/v26.3.0/src/tcp_wrap.cc#L433-L465 + if let UnixOrHost::Unix(path) = connection { + if !matches!( + bun_core::ip_address::parse_uv_inet_pton(path), + Some(core::net::IpAddr::V4(_)) + ) { + return Some(E::EINVAL); + } + } + if !socket.is_established() { + // One connect per handle (ConnectEx says EISCONN): https://github.com/nodejs/node/blob/v26.3.0/deps/uv/src/unix/tcp.c#L288-L289 + return Some(if cfg!(windows) { + E::EISCONN + } else { + E::EALREADY + }); + } + // Linux answers 0 once for a client socket that has not sent its FIN: `connect_finish` swaps it, as before. + if !self.is_server() && !socket.is_shutdown() { + return None; + } + Some(E::EISCONN) + } + /// Discard a still-live native socket so this wrapper can be reused for a /// fresh connect. `node:net` permits `socket.connect()` on an /// already-connected socket; without this the previous `us_socket_t`'s diff --git a/test/js/node/net/node-net.test.ts b/test/js/node/net/node-net.test.ts index 957a90819872..14e847c10336 100644 --- a/test/js/node/net/node-net.test.ts +++ b/test/js/node/net/node-net.test.ts @@ -516,7 +516,7 @@ describe("net.Socket write", () => { }), ); - it("should allow reconnecting after end()", async () => { + it("should allow reconnecting after end(), once the socket emitted 'close'", async () => { const server = new Server(socket => socket.end()); const port = await new Promise(resolve => { server.once("listening", () => resolve((server.address() as import("node:net").AddressInfo).port)); @@ -532,7 +532,9 @@ describe("net.Socket write", () => { socket.once("connect", (...args) => { socket.write("script\n", err => { if (err) return reject(err); - socket.end(() => setTimeout(resolve, 3)); + // The handle owns its socket until then: a connect() before it is EISCONN. + socket.once("close", resolve); + socket.end(); }); }); socket.connect(port, "127.0.0.1"); diff --git a/test/js/node/net/socket-reconnect-live-fixture.mjs b/test/js/node/net/socket-reconnect-live-fixture.mjs new file mode 100644 index 000000000000..e974642b4269 --- /dev/null +++ b/test/js/node/net/socket-reconnect-live-fixture.mjs @@ -0,0 +1,578 @@ +// connect() on a net.Socket or TLSSocket whose handle still owns a socket. +// The tests call the scenarios and pin what they return. The file has no bun import, so that +// `node ...` prints what node returns for them. +import { once } from "node:events"; +import fs from "node:fs"; +import net from "node:net"; +import path from "node:path"; +import tls from "node:tls"; +import { fileURLToPath } from "node:url"; + +const HOST = "127.0.0.1"; + +// Greets each connection, and answers "count?" with the number of connections it has accepted. +// `at` is a path to listen on. `host` is an address, or null for the default one. +async function listen({ greeting = "hello", at, host = HOST } = {}) { + const accepted = []; + const left = []; + let probes = 0; + const server = net.createServer({ allowHalfOpen: true }, c => { + accepted.push(c); + left.push(new Promise(resolve => c.once("end", resolve).once("error", resolve))); + c.on("error", () => {}); + c.on("data", d => { + if (String(d).includes("count?")) c.write(`accepted=${accepted.length - probes};`); + }); + if (greeting) c.write(greeting); + }); + if (at) server.listen(at); + else if (host === null) server.listen(0); + else server.listen(0, host); + await once(server, "listening"); + const port = at ? undefined : server.address().port; + return { + accepted, + at, + port, + // Settles when connection i saw the FIN or the reset of its peer. + left, + // The connections that were dialed to this server. The server accepts in the order of the dials, + // so it has accepted every earlier one when it answers the probe. + async dialed() { + probes++; + const probe = at ? net.connect({ path: at }) : net.connect(port, host ?? HOST); + let answer = ""; + probe.on("data", d => (answer += d)); + probe.write("count?"); + while (!answer.includes(";")) await once(probe, "data"); + probe.destroy(); + return Number(answer.match(/accepted=(\d+);/)[1]); + }, + close() { + for (const c of accepted) c.destroy(); + server.close(); + }, + }; +} + +async function listenTLS() { + const keys = new URL("../test/fixtures/keys/", import.meta.url); + const accepted = []; + const server = tls.createServer( + { key: fs.readFileSync(new URL("agent1-key.pem", keys)), cert: fs.readFileSync(new URL("agent1-cert.pem", keys)) }, + c => { + accepted.push(c); + c.on("error", () => {}); + c.write("hello"); + }, + ); + server.listen(0, HOST); + await once(server, "listening"); + return { + port: server.address().port, + close() { + for (const c of accepted) c.destroy(); + server.close(); + }, + }; +} + +// The events of `s` in order, its data, and its errors. `names` maps a placeholder to the text that +// differs from run to run (a port, a path). +function record(s, names = {}) { + const events = []; + const errors = []; + let data = ""; + let wake = () => {}; + let isClosed = false; + const scrub = text => { + text = String(text).replace(/Local \([^)]*\)/, "Local (ADDRESS)"); + for (const [name, value] of Object.entries(names)) text = text.replaceAll(String(value), name); + return text; + }; + for (const e of ["connect", "ready", "secureConnect", "end"]) s.on(e, () => events.push(e)); + for (const e of ["connectionAttempt", "connectionAttemptFailed"]) s.on(e, ip => events.push(`${e} ${scrub(ip)}`)); + s.on("error", e => { + events.push(`error ${e.code} ${e.syscall}`); + const address = e.address === undefined ? undefined : scrub(e.address); + errors.push({ message: scrub(e.message), code: e.code, syscall: e.syscall, address }); + }); + s.on("data", d => { + data += d; + wake(); + }); + const closed = new Promise(resolve => { + s.on("close", hadError => { + events.push(`close ${hadError}`); + isClosed = true; + wake(); + resolve(); + }); + }); + return { + // Copies: what a scenario's own cleanup makes the socket report is not part of its result. + get events() { + return [...events]; + }, + get errors() { + return [...errors]; + }, + names, + data: () => data, + closed, + // Settles when `text` arrived, or when the socket closed without it. + async until(text) { + while (!data.includes(text) && !isClosed) await new Promise(resolve => (wake = resolve)); + }, + }; +} + +// What the connect() call came to: "connect" when the socket said it again, "close" when the call destroyed it. +function outcomeOf(s) { + return new Promise(resolve => { + const done = name => () => { + s.removeListener("connect", onConnect); + s.removeListener("close", onClose); + resolve(name); + }; + const onConnect = done("connect"); + const onClose = done("close"); + s.once("connect", onConnect); + s.once("close", onClose); + }); +} + +async function connectedClient(peer, options, names) { + const s = net.connect({ port: peer.port, host: HOST, ...options }); + const r = record(s, { PORT: peer.port, ...names }); + await once(s, "connect"); + await r.until("hello"); + return { s, r }; +} + +async function connectedTLSClient(peer, names) { + const s = tls.connect({ port: peer.port, host: HOST, rejectUnauthorized: false }); + const r = record(s, { PORT: peer.port, ...names }); + await once(s, "secureConnect"); + await r.until("hello"); + return { s, r }; +} + +// Two addresses for one name: the connect takes the path that walks a list of addresses. +// Nothing has to listen on the second one: an attempt on it is in the events. +const twoAddresses = (port, first = HOST, second = "127.0.0.2") => ({ + port, + host: "reconnect-live.invalid", + autoSelectFamily: true, + lookup: (host, options, cb) => + cb(null, [ + { address: first, family: 4 }, + { address: second, family: 4 }, + ]), +}); + +// A second connect() with the path of `unix`, on the handle of `s`. +async function thenPath(s, r, unix) { + s.connect({ path: unix.at }); + const destroyedInCall = s.destroyed; + const outcome = await outcomeOf(s); + s.destroy(); + return { outcome, destroyedInCall, events: r.events, errors: r.errors, dialed: await unix.dialed() }; +} + +// A second connect() with the port of `peer`. +async function thenPort(s, peer) { + s.connect({ port: peer.port, host: HOST, rejectUnauthorized: false }); + const outcome = await outcomeOf(s); + s.destroy(); + return { outcome }; +} + +export const scenarios = { + // ---- Calls that node's handle refuses too. + + async "in-flight"() { + const peer = await listen(); + const s = new net.Socket(); + const r = record(s, { PORT: peer.port }); + s.connect(peer.port, HOST); + s.connect(peer.port, HOST); + const outcome = await outcomeOf(s); + s.destroy(); + peer.close(); + return { outcome, events: r.events, errors: r.errors }; + }, + + async "lookup-twice"() { + const peer = await listen(); + const s = new net.Socket(); + const r = record(s, { PORT: peer.port }); + s.connect({ + port: peer.port, + host: "reconnect-live.invalid", + lookup(host, options, cb) { + const answer = options.all ? [[{ address: HOST, family: 4 }]] : [HOST, 4]; + cb(null, ...answer); + cb(null, ...answer); + }, + }); + const outcome = await outcomeOf(s); + s.destroy(); + peer.close(); + return { outcome, events: r.events, errors: r.errors }; + }, + + // A socket that a server accepted has a connection and no connect request. + async accepted() { + const peer = await listen(); + const other = await listen({ greeting: "OTHER" }); + const client = net.connect(peer.port, HOST); + const c = record(client); + await once(client, "connect"); + await c.until("hello"); + const accepted = peer.accepted[0]; + const r = record(accepted, { PORT: other.port }); + accepted.connect(other.port, HOST); + const outcome = await outcomeOf(accepted); + accepted.destroy(); + await c.closed; + const result = { outcome, events: r.events, errors: r.errors, dialed: await other.dialed(), client: c.events }; + peer.close(); + other.close(); + return result; + }, + + // The socket sent its FIN. Its handle owns the socket until the peer closes too. + async "after-end"() { + const peer = await listen(); + const { s, r } = await connectedClient(peer); + s.end(); + await peer.left[0]; + s.connect(peer.port, HOST); + const outcome = await outcomeOf(s); + s.destroy(); + const result = { outcome, events: r.events, errors: r.errors, dialed: await peer.dialed() }; + peer.close(); + return result; + }, + + // bind(2) of a socket that is connected. + async "local-address"() { + const peer = await listen(); + const { s, r } = await connectedClient(peer); + s.connect({ port: peer.port, host: HOST, localAddress: HOST }); + const outcome = await outcomeOf(s); + s.destroy(); + const result = { outcome, events: r.events, errors: r.errors, dialed: await peer.dialed() }; + peer.close(); + return result; + }, + + // A path on a TCP handle. node hands the path to uv_ip4_addr, and EINVAL is that parse failing: + // https://github.com/nodejs/node/blob/v26.3.0/src/tcp_wrap.cc#L433-L465 + async "tcp-path"(dir) { + const unix = await listen({ at: path.join(dir, "tcp.sock") }); + const peer = await listen(); + const { s, r } = await connectedClient(peer, undefined, { SOCK: unix.at }); + const result = await thenPath(s, r, unix); + unix.close(); + peer.close(); + return result; + }, + + // The same while the connect of the handle is in flight: the parse comes before libuv sees the handle. + async "tcp-path-in-flight"(dir) { + const unix = await listen({ at: path.join(dir, "flight.sock") }); + const peer = await listen(); + const s = new net.Socket(); + const r = record(s, { PORT: peer.port, SOCK: unix.at }); + s.connect(peer.port, HOST); + // The dial of that call starts on the next tick. + await new Promise(resolve => process.nextTick(resolve)); + const result = await thenPath(s, r, unix); + unix.close(); + peer.close(); + return result; + }, + + async "tls-path"(dir) { + const unix = await listen({ at: path.join(dir, "tls.sock") }); + const peer = await listenTLS(); + const { s, r } = await connectedTLSClient(peer, { SOCK: unix.at }); + const result = await thenPath(s, r, unix); + unix.close(); + peer.close(); + return result; + }, + + // A socket accepted on the default address. Where that is dual-stack, its local address is an IPv6 one. + async "accepted-path"(dir) { + const unix = await listen({ at: path.join(dir, "accepted.sock") }); + const peer = await listen({ host: null }); + const client = net.connect(peer.port, HOST); + client.on("error", () => {}); + await once(client, "data"); + const accepted = peer.accepted[0]; + const r = record(accepted, { SOCK: unix.at }); + const result = await thenPath(accepted, r, unix); + client.destroy(); + unix.close(); + peer.close(); + return result; + }, + + // A path that is an IPv4 address is an address to node's TCP handle. + async "accepted-path-ipv4"() { + const peer = await listen(); + const client = net.connect(peer.port, HOST); + client.on("error", () => {}); + await once(client, "data"); + const accepted = peer.accepted[0]; + const r = record(accepted); + accepted.connect({ path: HOST }); + const outcome = await outcomeOf(accepted); + accepted.destroy(); + client.destroy(); + peer.close(); + return { outcome, events: r.events, errors: r.errors }; + }, + + async "tcp6-path"(dir) { + const unix = await listen({ at: path.join(dir, "tcp6.sock") }); + const peer = await listen({ host: "::1" }); + const s = net.connect(peer.port, "::1"); + const r = record(s, { PORT: peer.port, SOCK: unix.at }); + await once(s, "connect"); + await r.until("hello"); + const result = await thenPath(s, r, unix); + unix.close(); + peer.close(); + return result; + }, + + async "tls-in-flight"() { + const peer = await listenTLS(); + const s = tls.connect({ port: peer.port, host: HOST, rejectUnauthorized: false }); + const r = record(s, { PORT: peer.port }); + s.connect(peer.port, HOST); + const outcome = await outcomeOf(s); + s.destroy(); + peer.close(); + return { outcome, events: r.events, errors: r.errors }; + }, + + // ---- Calls that node refuses with another errno, at another time, or after a step more. + + // A unix socket that is connected. node reports what connect(2) says for the new path, with the + // request: EISCONN for a socket, ENOENT for a path that is not there. + async "unix-path"(dir) { + const peer = await listen({ at: path.join(dir, "unix.sock") }); + const s = net.connect({ path: peer.at }); + const r = record(s, { SOCK: peer.at }); + await once(s, "connect"); + await r.until("hello"); + const result = await thenPath(s, r, peer); + peer.close(); + return result; + }, + + // connect(path); connect(path) in one tick. + async "unix-in-flight"(dir) { + const peer = await listen({ at: path.join(dir, "twice.sock") }); + const s = new net.Socket(); + const r = record(s, { SOCK: peer.at }); + s.connect({ path: peer.at }); + s.connect({ path: peer.at }); + const destroyedInCall = s.destroyed; + const outcome = await outcomeOf(s); + s.destroy(); + peer.close(); + return { outcome, destroyedInCall, events: r.events, errors: r.errors }; + }, + + // A name with two addresses, and a bind that the handle refuses: the walk over the addresses stops. + // node goes on to the second address: it closes the connection without an event and dials there. + async "two-addresses-local-port"() { + const peer = await listen(); + const { s, r } = await connectedClient(peer); + r.names.LOCAL = s.localPort; + s.connect({ ...twoAddresses(peer.port), localPort: s.localPort }); + const outcome = await outcomeOf(s); + s.destroy(); + const result = { outcome, events: r.events, errors: r.errors, dialed: await peer.dialed() }; + peer.close(); + return result; + }, + + // The same with a first address that the socket's BlockList blocks. + async "two-addresses-blocked"() { + const peer = await listen(); + const blockList = new net.BlockList(); + const s = new net.Socket({ blockList }); + const r = record(s, { PORT: peer.port }); + s.connect(peer.port, HOST); + await once(s, "connect"); + await r.until("hello"); + blockList.addAddress(HOST); + s.connect(twoAddresses(peer.port)); + const outcome = await outcomeOf(s); + s.destroy(); + const result = { outcome, events: r.events, errors: r.errors, data: r.data(), dialed: await peer.dialed() }; + peer.close(); + return result; + }, + + // connect(); connect() by a name with two addresses (#21524). The socket that said 'close' is closed. + async "two-addresses-in-flight"() { + const peer = await listen(); + const s = new net.Socket(); + const r = record(s, { PORT: peer.port }); + s.connect(twoAddresses(peer.port)); + s.connect(twoAddresses(peer.port)); + const outcome = await outcomeOf(s); + // The server greeted the first dial. A socket that is still connected reads that before the probe returns. + await peer.dialed(); + const result = { + outcome, + events: r.events, + errors: r.errors, + data: r.data(), + state: { destroyed: s.destroyed, connecting: s.connecting, readyState: s.readyState }, + }; + s.destroy(); + peer.close(); + return result; + }, + + // The first address refuses. A 'connectionAttemptFailed' listener calls connect(path), and the walk + // of the first call then finds the handle with the socket of that dial. + async "walk-finds-a-socket"(dir) { + const unix = await listen({ at: path.join(dir, "walk.sock") }); + const peer = await listen(); + const s = new net.Socket(); + const r = record(s, { PORT: peer.port, SOCK: unix.at }); + s.once("connectionAttemptFailed", () => s.connect({ path: unix.at })); + // Nothing listens on 127.0.0.9: the kernel refuses. + s.connect(twoAddresses(peer.port, "127.0.0.9", HOST)); + const outcome = await outcomeOf(s); + // A socket that is still connected reads a greeting before the probes return. + const dialed = { tcp: await peer.dialed(), unix: await unix.dialed() }; + const result = { + outcome, + events: r.events, + data: r.data(), + dialed, + state: { destroyed: s.destroyed, connecting: s.connecting, readyState: s.readyState }, + }; + s.destroy(); + unix.close(); + peer.close(); + return result; + }, + + // Bun only: node's connect() takes no fd. libuv opens an fd only on a handle that has none. + async fd() { + const peer = await listen(); + const { s, r } = await connectedClient(peer); + const fd = fs.openSync(fileURLToPath(import.meta.url), "r"); + try { + s.connect({ fd }); + const outcome = await Promise.race([outcomeOf(s), peer.dialed().then(() => "none")]); + s.destroy(); + const result = { outcome, events: r.events, errors: r.errors, dialed: await peer.dialed() }; + peer.close(); + return result; + } finally { + fs.closeSync(fd); + } + }, + + // Windows: a named pipe that is connected, then its own name, then an address. + async "named-pipe"() { + const results = {}; + for (const call of ["path", "address"]) { + const pipe = await listen({ at: `\\\\.\\pipe\\reconnect-live-${process.pid}-${call}` }); + const peer = await listen(); + const s = net.connect({ path: pipe.at }); + const r = record(s, { PIPE: pipe.at, PORT: peer.port }); + await once(s, "connect"); + await r.until("hello"); + if (call === "path") s.connect({ path: pipe.at }); + else s.connect(peer.port, HOST); + const outcome = await outcomeOf(s); + s.destroy(); + results[call] = { outcome, events: r.events, errors: r.errors, dialed: await peer.dialed() }; + pipe.close(); + peer.close(); + } + return results; + }, + + // ---- A client socket that has not sent its FIN. The connection is swapped for a new one, as before: + // connect(2) answers 0 for such a socket once on Linux, and node says 'connect' again there. + + async connected() { + const peer = await listen(); + const { s } = await connectedClient(peer); + const result = await thenPort(s, peer); + peer.close(); + return result; + }, + + async "other-port"() { + const peer = await listen(); + const other = await listen({ greeting: "OTHER" }); + const { s } = await connectedClient(peer); + const result = await thenPort(s, other); + peer.close(); + other.close(); + return result; + }, + + // The peer sent its FIN. The socket is half-open. + async "after-fin"() { + const peer = await listen(); + const { s } = await connectedClient(peer, { allowHalfOpen: true }); + peer.accepted[0].end(); + await once(s, "end"); + const result = await thenPort(s, peer); + peer.close(); + return result; + }, + + async "two-addresses"() { + const peer = await listen(); + const { s } = await connectedClient(peer); + s.connect(twoAddresses(peer.port)); + const outcome = await outcomeOf(s); + s.destroy(); + peer.close(); + return { outcome }; + }, + + async "tls-established"() { + const peer = await listenTLS(); + const { s } = await connectedTLSClient(peer); + const result = await thenPort(s, peer); + peer.close(); + return result; + }, + + // TCP is up and the handshake is not done. + async "tls-handshake"() { + const peer = await listenTLS(); + const s = tls.connect({ port: peer.port, host: HOST, rejectUnauthorized: false }); + s.on("error", () => {}); + await once(s, "connect"); + const result = await thenPort(s, peer); + peer.close(); + return result; + }, +}; + +if (process.argv[1] === fileURLToPath(import.meta.url)) { + const [dir, ...names] = process.argv.slice(2); + const results = {}; + for (const name of names) results[name] = await scenarios[name](dir); + console.log(JSON.stringify(results, null, 1)); +} diff --git a/test/js/node/net/socket-reconnect-live.test.ts b/test/js/node/net/socket-reconnect-live.test.ts index f83ed929e4e3..8c4a7e7b4b3d 100644 --- a/test/js/node/net/socket-reconnect-live.test.ts +++ b/test/js/node/net/socket-reconnect-live.test.ts @@ -1,9 +1,18 @@ import { describe, expect, it } from "bun:test"; -import { bunEnv, bunExe } from "harness"; +import { bunEnv, bunExe, isIPv6, isLinux, isWindows, nodeExe, tempDir } from "harness"; +import { readFileSync } from "node:fs"; +import { join } from "node:path"; +import { scenarios } from "./socket-reconnect-live-fixture.mjs"; + +// A connect that is in flight on a TCP handle. libuv refuses a second one on POSIX, and ConnectEx does on Windows: +// https://github.com/nodejs/node/blob/v26.3.0/deps/uv/src/unix/tcp.c#L288-L289 +// https://github.com/nodejs/node/blob/v26.3.0/deps/uv/src/win/tcp.c#L900-L918 +const busy = isWindows ? "EISCONN" : "EALREADY"; // connect_finish must tear down a still-live previous native socket before // reusing the wrapper, not alias two native sockets onto one ext slot. -describe.concurrent("socket.connect() on an already-connected socket", () => { +// One at a time: a debug build needs 3 s to start one of these processes. +describe("socket.connect() on an already-connected socket", () => { it("does not crash and emits connect for the new connection", async () => { await using proc = Bun.spawn({ cmd: [ @@ -100,9 +109,15 @@ describe.concurrent("socket.connect() on an already-connected socket", () => { }); it("does not crash when reconnecting while the first connect is still in flight", async () => { + using dir = tempDir("reconnect-live-trace", {}); + const traceFile = join(String(dir), "node_trace.log"); await using proc = Bun.spawn({ cmd: [ bunExe(), + "--trace-event-categories", + "node.net.native", + "--trace-event-file-pattern", + traceFile, "-e", ` const { createServer, connect } = require("node:net"); @@ -129,10 +144,293 @@ describe.concurrent("socket.connect() on an already-connected socket", () => { stderr: "pipe", }); const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); - // Node emits EALREADY here; Bun drops the first in-flight connect and - // completes the second. Either is acceptable so long as the process - // exits cleanly. - expect(["connect", "err:EALREADY"]).toContain(stdout); - expect({ stderr, exitCode }).toEqual({ stderr, exitCode: 0 }); + // The handle refuses the second connect, as node's does. + expect({ stdout, stderr, exitCode }).toEqual({ stdout: "err:" + busy, stderr, exitCode: 0 }); + // The refused connect ends its trace span too. + const phases = JSON.parse(readFileSync(traceFile, "utf8")) + .traceEvents.filter((e: { name: string }) => e.name === "connect") + .map((e: { ph: string }) => e.ph); + expect(phases.sort()).toEqual(["b", "b", "e", "e"]); }); }); + +// A net.Socket keeps one native handle. Where node's handle refuses a connect() because it still owns +// a socket, ours answers with the same errno from the state of that socket: nothing is closed and +// nothing is dialed for the answer, and the net.Socket is destroyed with the error. + +// The scenarios are in a file that node runs too. +const fixture = join(import.meta.dir, "socket-reconnect-live-fixture.mjs"); + +async function inThisProcess(name: string) { + using dir = tempDir("reconnect-live", {}); + return await (scenarios as Record Promise>)[name](String(dir)); +} + +// Runs the scenarios in a process of `exe`, one after the other. +async function inProcessOf(exe: string, names: readonly string[]) { + using dir = tempDir("reconnect-live", {}); + await using proc = Bun.spawn({ + cmd: [exe, fixture, String(dir), ...names], + env: bunEnv, + stdout: "pipe", + stderr: "pipe", + }); + const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + expect(stderr).toBe(""); + // The results are the last thing the fixture prints: no exit 0, nothing to parse. + expect(exitCode).toBe(0); + return JSON.parse(stdout) as Record; +} + +const dial = ["connectionAttempt 127.0.0.1", "connect", "ready"]; +// What a refused connect() adds to the events of a socket. 'close' reports that there was an error. +const refused = (code: string) => ["connectionAttempt 127.0.0.1", `error ${code} connect`, "close true"]; +// https://github.com/nodejs/node/blob/v26.3.0/lib/net.js#L1161-L1170 +const refusal = (code: string) => ({ + message: `connect ${code} 127.0.0.1:PORT - Local (ADDRESS)`, + code, + syscall: "connect", + address: "127.0.0.1", +}); +// A path on a TCP handle: node parses it as an IPv4 address, and that fails. +const pathOnTCP = (before: string[], dialed = 0) => ({ + outcome: "close", + destroyedInCall: true, + events: [...before, "connectionAttempt SOCK", "error EINVAL connect", "close true"], + errors: [{ message: "connect EINVAL SOCK - Local (ADDRESS)", code: "EINVAL", syscall: "connect", address: "SOCK" }], + dialed, +}); + +// These print the same in node v26.3.0 on Linux. libuv's Windows handles are other code: no node row there. +const sameInNode = [ + "in-flight", + "lookup-twice", + "accepted", + "after-end", + "local-address", + "tcp-path", + "tcp-path-in-flight", + "tls-path", + "accepted-path", + "accepted-path-ipv4", + ...(isIPv6() ? ["tcp6-path"] : []), + "tls-in-flight", +]; +const node = nodeExe(); +let fromNode: Promise> | undefined; + +describe.each([ + ["bun", false, inThisProcess], + ["node", !node || isWindows, async (name: string) => (await (fromNode ??= inProcessOf(node!, sameInNode)))[name]], +] as const)("connect() that node's handle refuses (%s)", (_runtime, skip, run) => { + it.skipIf(skip)("while a connect is in flight", async () => { + expect(await run("in-flight")).toEqual({ + outcome: "close", + events: ["connectionAttempt 127.0.0.1", ...refused(busy)], + errors: [refusal(busy)], + }); + }); + + it.skipIf(skip)("from a lookup that calls back twice", async () => { + expect(await run("lookup-twice")).toEqual({ + outcome: "close", + events: ["connectionAttempt 127.0.0.1", ...refused(busy)], + errors: [refusal(busy)], + }); + }); + + it.skipIf(skip)("on a socket that a server accepted: EISCONN", async () => { + expect(await run("accepted")).toEqual({ + outcome: "close", + events: refused("EISCONN"), + errors: [refusal("EISCONN")], + dialed: 0, + // The peer of the accepted socket sees a FIN. + client: [...dial, "end", "close false"], + }); + }); + + it.skipIf(skip)("after end(), until the peer closes: EISCONN", async () => { + expect(await run("after-end")).toEqual({ + outcome: "close", + events: [...dial, ...refused("EISCONN")], + errors: [refusal("EISCONN")], + dialed: 1, + }); + }); + + it.skipIf(skip)("with a localAddress on a connected socket: bind EINVAL", async () => { + expect(await run("local-address")).toEqual({ + outcome: "close", + events: [...dial, "error EINVAL bind", "close true"], + errors: [{ message: "bind EINVAL 127.0.0.1", code: "EINVAL", syscall: "bind", address: "127.0.0.1" }], + dialed: 1, + }); + }); + + it.skipIf(skip)("with a path on a connected TCP socket: EINVAL", async () => { + expect(await run("tcp-path")).toEqual(pathOnTCP(dial)); + }); + + it.skipIf(skip)("with a path while the TCP connect is in flight: EINVAL", async () => { + expect(await run("tcp-path-in-flight")).toEqual(pathOnTCP(["connectionAttempt 127.0.0.1"])); + }); + + it.skipIf(skip)("with a path on an established TLSSocket: EINVAL", async () => { + expect(await run("tls-path")).toEqual(pathOnTCP([...dial, "secureConnect"])); + }); + + it.skipIf(skip)("with a path on a socket accepted on the default address: EINVAL", async () => { + expect(await run("accepted-path")).toEqual(pathOnTCP([])); + }); + + it.skipIf(skip)("with a path that is an IPv4 address, on an accepted socket: EISCONN", async () => { + expect(await run("accepted-path-ipv4")).toEqual({ + outcome: "close", + events: refused("EISCONN"), + errors: [ + { + message: "connect EISCONN 127.0.0.1 - Local (ADDRESS)", + code: "EISCONN", + syscall: "connect", + address: "127.0.0.1", + }, + ], + }); + }); + + it.skipIf(skip || !isIPv6())("with a path on a connected IPv6 socket: EINVAL", async () => { + expect(await run("tcp6-path")).toEqual(pathOnTCP(["connectionAttempt ::1", "connect", "ready"])); + }); + + it.skipIf(skip)("on a TLSSocket whose connect is in flight", async () => { + expect(await run("tls-in-flight")).toEqual({ + outcome: "close", + events: ["connectionAttempt 127.0.0.1", ...refused(busy)], + errors: [refusal(busy)], + }); + }); +}); + +describe("connect() that node refuses by another way", () => { + const onUnix = { message: "connect EISCONN SOCK", code: "EISCONN", syscall: "connect", address: "SOCK" }; + + // node reports what connect(2) says for the new path, with the request, one loop turn later: + // https://github.com/nodejs/node/blob/v26.3.0/deps/uv/src/unix/pipe.c#L300-L330 + it("on a connected unix socket: EISCONN in the call", async () => { + expect(await inThisProcess("unix-path")).toEqual({ + outcome: "close", + destroyedInCall: true, + events: ["connect", "ready", "connectionAttempt SOCK", "error EISCONN connect", "close true"], + errors: [onUnix], + dialed: 1, + }); + }); + + it("on a unix socket whose connect is in flight: EISCONN in the call", async () => { + expect(await inThisProcess("unix-in-flight")).toEqual({ + outcome: "close", + destroyedInCall: true, + events: ["connectionAttempt SOCK", "connectionAttempt SOCK", "error EISCONN connect", "close true"], + errors: [onUnix], + }); + }); + + // Bun only: node's connect() takes no fd. + it("with an fd: EBUSY", async () => { + expect(await inThisProcess("fd")).toEqual({ + outcome: "close", + events: [...dial, "error EBUSY open", "close true"], + errors: [{ message: "open EBUSY", code: "EBUSY", syscall: "open" }], + dialed: 1, + }); + }); + + // libuv reports EBUSY for a named pipe, with the request. + it.skipIf(!isWindows)("on a connected named pipe: EISCONN in the call", async () => { + expect(await inThisProcess("named-pipe")).toEqual({ + path: { + outcome: "close", + events: ["connect", "ready", "connectionAttempt PIPE", "error EISCONN connect", "close true"], + errors: [{ message: "connect EISCONN PIPE", code: "EISCONN", syscall: "connect", address: "PIPE" }], + dialed: 0, + }, + address: { + outcome: "close", + events: ["connect", "ready", "connectionAttempt 127.0.0.1", "error EISCONN connect", "close true"], + errors: [ + { message: "connect EISCONN 127.0.0.1:PORT", code: "EISCONN", syscall: "connect", address: "127.0.0.1" }, + ], + dialed: 0, + }, + }); + }); + + // After such a refusal node goes on to the next address of the name: it closes the socket of the + // handle without an event and dials there. Here the socket is destroyed with the error. + describe("by a name with two addresses, the walk stops", () => { + it("at a bind that the handle refuses", async () => { + expect(await inThisProcess("two-addresses-local-port")).toEqual({ + outcome: "close", + events: [...dial, "error EINVAL bind", "close true"], + errors: [{ message: "bind EINVAL 0.0.0.0:LOCAL", code: "EINVAL", syscall: "bind", address: "0.0.0.0" }], + dialed: 1, + }); + }); + + it("at an address that the BlockList of the socket blocks", async () => { + expect(await inThisProcess("two-addresses-blocked")).toEqual({ + outcome: "close", + events: [...dial, "connectionAttemptFailed 127.0.0.1", "error ERR_IP_BLOCKED undefined", "close true"], + errors: [{ message: "IP(127.0.0.1) is blocked by net.BlockList", code: "ERR_IP_BLOCKED" }], + data: "hello", + dialed: 1, + }); + }); + + // #21524 + it("at a connect that is in flight, and the socket is closed", async () => { + expect(await inThisProcess("two-addresses-in-flight")).toEqual({ + outcome: "close", + events: [ + "connectionAttempt 127.0.0.1", + "connectionAttempt 127.0.0.1", + "connectionAttemptFailed 127.0.0.1", + // The attempt of the first call, which the destroy cancelled. + "connectionAttemptFailed 127.0.0.1", + `error ${busy} connect`, + "close true", + ], + errors: [refusal(busy)], + data: "", + state: { destroyed: true, connecting: false, readyState: "closed" }, + }); + }); + + // 127.0.0.9 refuses at once on Linux only. + it.skipIf(!isLinux)("at the socket that a 'connectionAttemptFailed' listener dialed on the handle", async () => { + expect(await inThisProcess("walk-finds-a-socket")).toEqual({ + outcome: "close", + events: [ + "connectionAttempt 127.0.0.9", + "connectionAttemptFailed 127.0.0.9", + "connectionAttempt SOCK", + "error ECONNREFUSED connect", + "close true", + ], + data: "", + dialed: { tcp: 0, unix: 1 }, + state: { destroyed: true, connecting: false, readyState: "closed" }, + }); + }); + }); +}); + +// connect(2) answers 0 for such a socket once on Linux, and node emits 'connect' again on the same +// connection. Here the connection is swapped for a new one, as before. +it.each(["connected", "other-port", "after-fin", "two-addresses", "tls-established", "tls-handshake"])( + "a client socket that has not sent its FIN still emits 'connect' again (%s)", + async name => { + expect(await inThisProcess(name)).toEqual({ outcome: "connect" }); + }, +);