diff --git a/src/js/node/net.ts b/src/js/node/net.ts index 763dcac394d2..af0493e14a6b 100644 --- a/src/js/node/net.ts +++ b/src/js/node/net.ts @@ -122,6 +122,7 @@ function lazyBlockList() { } const newDetachedSocket = $newRustFunction("node_net_binding.rs", "newDetachedSocket", 1); const doConnect = $newRustFunction("node_net_binding.rs", "doConnect", 2); +const drainMicrotasksAndNextTicks = $newRustFunction("node_net_binding.rs", "drainMicrotasksAndNextTicks", 0); const addServerName = $newRustFunction("Listener.rs", "jsAddServerName", 3); const upgradeDuplexToTLS = $newRustFunction("runtime/socket/socket.rs", "jsUpgradeDuplexToTLS", 2); @@ -194,6 +195,8 @@ const kOnreadPendingEnd = Symbol("kOnreadPendingEnd"); // Node's handle.reading: https://github.com/nodejs/node/blob/v26.3.0/lib/net.js#L817-L845 const kOnreadReading = Symbol("kOnreadReading"); const kOnreadEmptyTail = Buffer.alloc(0); +// Not zero while the checkpoint between two onread slices runs: see drainOnreadTailNT. +let onreadCheckpoints = 0; const kwriteCallback = Symbol("writeCallback"); const kSocketClass = Symbol("kSocketClass"); @@ -1775,7 +1778,20 @@ function Socket(options?) { reportError(e); } if (self.destroyed) return; - if (ret === false || !self[kOnreadReading]) { + let stop = ret === false || !self[kOnreadReading]; + // Node drains ticks and microtasks after each read, except in a nested scope (a wrapped stream's 'data' emit): https://github.com/nodejs/node/blob/v26.3.0/src/api/callback.cc#L165-L207 + if (!stop && offset < total && !self[kupgraded]) { + const handle = self._handle; + onreadCheckpoints++; + try { + drainMicrotasksAndNextTicks(); + } finally { + onreadCheckpoints--; + } + if (onreadConnectionGone(self, handle)) return; + stop = !self[kOnreadReading]; + } + if (stop) { const rest = buffer.subarray(offset); self[kOnreadTail] = rest.length !== 0 ? rest : kOnreadEmptyTail; readStop(self, self._handle); @@ -2339,6 +2355,11 @@ function hasUnflushedWrites(connection) { return connection.writableLength > 0 || connection[kwriteCallback] != null; } +// The rest of a chunk, and a FIN that waits behind it, belong to the connection that read them. +function onreadConnectionGone(self, handle) { + return self.destroyed || self.connecting || self._handle !== handle; +} + // The tail is delivered on the next tick. A pause() before then clears kOnreadReading and the tick delivers nothing. function drainOnreadTail(self) { if (self[kOnreadTail] === undefined) return false; @@ -2351,13 +2372,23 @@ function drainOnreadTail(self) { } function drainOnreadTailNT(socket) { + // Inside the checkpoint of another socket this drain would nest, one level per socket with a tail. + if (onreadCheckpoints !== 0) { + setImmediate(drainOnreadTailNT, socket); + return; + } socket[kOnreadDraining] = false; const tail = socket[kOnreadTail]; if (tail === undefined || socket.destroyed) return; if (!socket[kOnreadReading]) return; socket[kOnreadTail] = undefined; + const handle = socket._handle; socket[kOnreadDeliver](tail); - if (socket[kOnreadTail] !== undefined || socket.destroyed) return; + if (onreadConnectionGone(socket, handle)) { + socket[kOnreadPendingEnd] = false; + return; + } + if (socket[kOnreadTail] !== undefined) return; if (socket[kOnreadPendingEnd]) { socket[kOnreadPendingEnd] = false; finishSocketEnd(socket); diff --git a/src/runtime/node/node_net_binding.rs b/src/runtime/node/node_net_binding.rs index e4ebc3d406ab..6f9e81d07451 100644 --- a/src/runtime/node/node_net_binding.rs +++ b/src/runtime/node/node_net_binding.rs @@ -173,3 +173,15 @@ pub(crate) fn do_connect(global: &JSGlobalObject, frame: &CallFrame) -> JsResult let maybe_tls = prev.as_::(); Listener::connect_inner(global, maybe_tcp, maybe_tls, opts) } + +/// net.ts calls this between two `onread` callbacks of one native read, where node ends a callback scope. +#[bun_jsc::host_fn] +pub(crate) fn drain_microtasks_and_next_ticks( + global: &JSGlobalObject, + _frame: &CallFrame, +) -> JsResult { + global + .drain_microtasks_and_next_ticks() + .map_err(|stopped| stopped.throw(global))?; + Ok(JSValue::UNDEFINED) +} diff --git a/test/js/node/net/node-net.test.ts b/test/js/node/net/node-net.test.ts index ea9a76267808..1d6a20d6fea1 100644 --- a/test/js/node/net/node-net.test.ts +++ b/test/js/node/net/node-net.test.ts @@ -30,7 +30,8 @@ import { Stream, } from "node:net"; import { join } from "node:path"; -import { TLSSocket } from "node:tls"; +import { Duplex } from "node:stream"; +import { createServer as createTLSServer, connect as tlsConnect, TLSSocket } from "node:tls"; const socket_domain = tmpdirSync(); @@ -2959,6 +2960,484 @@ it("onread: a swallowed throw does not drop the rest of the current native read" expect(exitCode).toBe(0); }); +// Node reads one onread buffer per callback scope (MakeCallback), and a scope +// ends with the tick queue and the microtasks drained. Bun slices one larger +// native read into several callbacks, so it runs that checkpoint between two +// slices. Every expectation below is what node v26.3.0 does. +// https://github.com/nodejs/node/blob/v26.3.0/src/api/callback.cc#L165-L207 +describe("net.Socket onread: the callbacks of one native read are separate callback scopes", () => { + // One 12-byte write is one native read; a 4-byte onread buffer makes three callbacks out of it. + const payload = "abcdefghijkl"; + const slices = ["abcd", "efgh", "ijkl"]; + const transports = ["tcp", "tls"] as const; + + async function startServer(transport: (typeof transports)[number]) { + const onConnection = (c: Socket) => { + c.on("error", () => {}); + c.end(payload); + }; + const server = + transport === "tls" + ? createTLSServer({ key: tlsCert.key, cert: tlsCert.cert }, onConnection) + : createServer(onConnection); + const listening = Promise.withResolvers(); + server.once("error", listening.reject); + server.listen(0, "127.0.0.1", () => listening.resolve()); + await listening.promise; + server.off("error", listening.reject); + return server; + } + + function dial( + transport: (typeof transports)[number], + server: Server, + callback: (this: unknown, nread: number, buf: Buffer) => boolean | void, + ): Socket { + const options = { + port: (server.address() as import("node:net").AddressInfo).port, + host: "127.0.0.1", + onread: { buffer: Buffer.alloc(4), callback }, + }; + return transport === "tls" ? tlsConnect({ ...options, rejectUnauthorized: false }) : createConnection(options); + } + + const immediate = () => new Promise(resolve => setImmediate(resolve)); + + it.each(transports)( + "%s: the ticks and microtasks a callback queued run before the next callback", + async transport => { + const server = await startServer(transport); + const log: string[] = []; + const closed = Promise.withResolvers(); + let socket: Socket | undefined; + try { + socket = dial(transport, server, (n, buf) => { + log.push(`callback ${buf.toString("latin1", 0, n)}`); + // The tick and the microtask read the static buffer: it still holds the bytes of their own callback. + process.nextTick(() => log.push(`tick ${buf.toString("latin1", 0, n)}`)); + queueMicrotask(() => log.push(`microtask ${buf.toString("latin1", 0, n)}`)); + }); + socket.on("error", closed.reject); + socket.on("close", () => closed.resolve()); + await closed.promise; + expect(log).toEqual(slices.flatMap(slice => [`callback ${slice}`, `tick ${slice}`, `microtask ${slice}`])); + } finally { + socket?.destroy(); + server.close(); + } + }, + ); + + it.each([ + ["queueMicrotask", (fn: () => void) => queueMicrotask(fn)], + ["process.nextTick", (fn: () => void) => process.nextTick(fn)], + ["a promise reaction", (fn: () => void) => void Promise.resolve().then(fn)], + ])("a pause() deferred with %s stops the rest of the read until resume()", async (_name, defer) => { + const server = await startServer("tcp"); + const received: string[] = []; + const paused = Promise.withResolvers(); + const done = Promise.withResolvers(); + let socket: Socket | undefined; + try { + socket = dial("tcp", server, (n, buf) => { + received.push(buf.toString("latin1", 0, n)); + if (received.length === 1) { + defer(() => { + socket!.pause(); + paused.resolve([...received]); + }); + } + if (received.length === slices.length) done.resolve(); + }); + socket.on("error", done.reject); + socket.on("close", () => done.reject(new Error(`closed before all data was delivered: ${received.join("|")}`))); + expect(await paused.promise).toEqual(["abcd"]); + // The rest of the read waits for resume(), it is not delivered on a later turn. + for (let i = 0; i < 4; i++) await immediate(); + expect(received).toEqual(["abcd"]); + socket.resume(); + await done.promise; + expect(received).toEqual(slices); + } finally { + socket?.destroy(); + server.close(); + } + }); + + it("a deferred destroy() drops the rest of the read", async () => { + const server = await startServer("tcp"); + const received: string[] = []; + const closed = Promise.withResolvers(); + let socket: Socket | undefined; + try { + socket = dial("tcp", server, (n, buf) => { + received.push(buf.toString("latin1", 0, n)); + queueMicrotask(() => socket!.destroy()); + }); + socket.on("error", closed.reject); + socket.on("close", () => closed.resolve()); + await closed.promise; + expect(received).toEqual(["abcd"]); + } finally { + socket?.destroy(); + server.close(); + } + }); + + it("a deferred destroy() then connect() does not hand the rest of the old read to the new connection", async () => { + const first = await startServer("tcp"); + const second = createServer(c => { + c.on("error", () => {}); + c.end("mnopqrst"); + }); + const listening = Promise.withResolvers(); + second.once("error", listening.reject); + second.listen(0, "127.0.0.1", () => listening.resolve()); + await listening.promise; + const received: [number, string][] = []; + const done = Promise.withResolvers(); + let generation = 1; + const socket = new Socket({ + onread: { + buffer: Buffer.alloc(4), + callback(n: number, buf: Buffer) { + const slice = buf.toString("latin1", 0, n); + received.push([generation, slice]); + if (received.length === 1) { + process.nextTick(() => { + socket.destroy(); + generation = 2; + socket.connect((second.address() as import("node:net").AddressInfo).port, "127.0.0.1"); + }); + } + if (slice === "qrst") done.resolve(); + }, + }, + }); + try { + socket.on("error", done.reject); + socket.connect((first.address() as import("node:net").AddressInfo).port, "127.0.0.1"); + await done.promise; + expect(received).toEqual([ + [1, "abcd"], + [2, "mnop"], + [2, "qrst"], + ]); + } finally { + socket.destroy(); + first.close(); + second.close(); + } + }); + + // The peer resets the first connection while the declined tail waits. With no 'error' listener bun keeps that close + // behind the tail, like a FIN (kOnreadPendingEnd). A connection that replaces the first one must not inherit it. + it.each([ + ["destroy() then connect() in one tick", true], + ["destroy(), then connect() after 'close'", false], + ])("inside a redelivered tail, a deferred %s does not end the next connection", async (_name, sameTick) => { + let peer: Socket | undefined; + const first = createServer(c => { + peer = c; + c.on("error", () => {}); + c.write(payload); + }); + const second = createServer(c => { + c.on("error", () => {}); + c.write("mnopqrst"); + }); + const listening = [Promise.withResolvers(), Promise.withResolvers()]; + [first, second].forEach((server, i) => { + server.once("error", listening[i].reject); + server.listen(0, "127.0.0.1", () => listening[i].resolve()); + }); + await Promise.all(listening.map(l => l.promise)); + const port = (server: Server) => (server.address() as import("node:net").AddressInfo).port; + const events: string[] = []; + const declined = [Promise.withResolvers(), Promise.withResolvers()]; + const delivered = Promise.withResolvers(); + let generation = 1; + let endedAtConnect: boolean | undefined; + const reconnect = () => { + generation = 2; + socket.on("error", delivered.reject); + socket.connect(port(second), "127.0.0.1", () => { + endedAtConnect = (socket as any)._readableState.ended; + }); + }; + const socket = new Socket({ + onread: { + buffer: Buffer.alloc(4), + callback(n: number, buf: Buffer) { + const slice = buf.toString("latin1", 0, n); + events.push(`${generation} ${slice}`); + if (slice === "abcd" || slice === "mnop") { + declined[generation - 1].resolve(); + return false; + } + if (generation === 2) return void delivered.resolve(); + process.nextTick(() => { + socket.destroy(); + if (sameTick) reconnect(); + else socket.once("close", reconnect); + }); + }, + }, + }); + try { + socket.on("end", () => generation === 2 && events.push("end 2")); + socket.connect(port(first), "127.0.0.1"); + await declined[0].promise; + peer!.resetAndDestroy(); + while ((socket as any)._handle?.readyState === 1) await immediate(); + socket.resume(); + await declined[1].promise; + socket.resume(); + await delivered.promise; + for (let i = 0; i < 4; i++) await immediate(); + expect({ events, endedAtConnect }).toEqual({ + events: ["1 abcd", "1 efgh", "2 mnop", "2 qrst"], + endedAtConnect: false, + }); + } finally { + socket.destroy(); + first.close(); + second.close(); + } + }); + + it("the tail that resume() redelivers gets the same checkpoints", async () => { + const server = await startServer("tcp"); + const log: string[] = []; + const declined = Promise.withResolvers(); + const closed = Promise.withResolvers(); + let socket: Socket | undefined; + try { + socket = dial("tcp", server, (n, buf) => { + const slice = buf.toString("latin1", 0, n); + log.push(`callback ${slice}`); + process.nextTick(() => log.push(`tick ${slice}`)); + queueMicrotask(() => log.push(`microtask ${slice}`)); + if (slice === "abcd") { + declined.resolve(); + return false; + } + }); + socket.on("error", closed.reject); + socket.on("close", () => closed.resolve()); + await declined.promise; + await immediate(); + log.push("resume"); + socket.resume(); + await closed.promise; + expect(log).toEqual([ + "callback abcd", + "tick abcd", + "microtask abcd", + "resume", + ...slices.slice(1).flatMap(slice => [`callback ${slice}`, `tick ${slice}`, `microtask ${slice}`]), + ]); + } finally { + socket?.destroy(); + server.close(); + } + }); + + // The checkpoint runs user code while the loop still holds the rest of the read. + it.each([ + ["pause() then resume() in one tick", (socket: Socket) => process.nextTick(() => socket.pause().resume())], + [ + "pause() in a tick, resume() in the microtask after it", + (socket: Socket) => + process.nextTick(() => { + socket.pause(); + queueMicrotask(() => socket.resume()); + }), + ], + ["read() in a microtask", (socket: Socket) => queueMicrotask(() => socket.read())], + // pause() stops the handle and read() starts it again: https://github.com/nodejs/node/blob/v26.3.0/lib/net.js#L817-L845 + ["pause() then read() in one tick", (socket: Socket) => process.nextTick(() => void socket.pause().read())], + ["pause() then read() in one microtask", (socket: Socket) => queueMicrotask(() => void socket.pause().read())], + [ + "pause() then read() in one promise reaction", + (socket: Socket) => void Promise.resolve().then(() => void socket.pause().read()), + ], + ])("%s keeps the read going, each slice once and in order", async (_name, reenter) => { + const server = await startServer("tcp"); + const received: string[] = []; + const closed = Promise.withResolvers(); + let socket: Socket | undefined; + try { + socket = dial("tcp", server, (n, buf) => { + received.push(buf.toString("latin1", 0, n)); + reenter(socket!); + }); + socket.on("error", closed.reject); + socket.on("close", () => closed.resolve()); + await closed.promise; + expect(received).toEqual(slices); + } finally { + socket?.destroy(); + server.close(); + } + }); + + // A tail drain is a tick, so the checkpoint of one socket would run the tail drain of the next one inside it. + it("tail drains of several sockets that come due in one turn do not nest", async () => { + const server = await startServer("tcp"); + const count = 3; + const received: string[][] = []; + const sockets: Socket[] = []; + const closed: Promise[] = []; + const allDeclined = Promise.withResolvers(); + let declined = 0; + // Callbacks whose tick did not run yet: 1 when every callback is followed by its own checkpoint. + let pending = 0; + let maxPending = 0; + try { + for (let i = 0; i < count; i++) { + const mine: string[] = []; + received.push(mine); + const { promise, resolve, reject } = Promise.withResolvers(); + closed.push(promise); + const socket = dial("tcp", server, (n, buf) => { + mine.push(buf.toString("latin1", 0, n)); + if (mine.length === 1) { + if (++declined === count) allDeclined.resolve(); + return false; + } + maxPending = Math.max(maxPending, ++pending); + process.nextTick(() => pending--); + }); + socket.on("error", reject); + socket.on("close", () => resolve()); + sockets.push(socket); + } + await allDeclined.promise; + await immediate(); + for (const socket of sockets) socket.resume(); + await Promise.all(closed); + expect(received).toEqual([slices, slices, slices]); + expect(maxPending).toBe(1); + } finally { + for (const socket of sockets) socket.destroy(); + server.close(); + } + }); + + // A socket that wraps another stream reads from inside that stream's 'data' emit: a nested scope, which does not + // drain. Holds the same if `onread` is ignored for a wrapped socket, as in node. + it("a TLS socket over a generic Duplex does not run the Duplex's ticks inside its 'data' emit", async () => { + const server = createTLSServer({ key: tlsCert.key, cert: tlsCert.cert }, c => { + c.on("error", () => {}); + c.once("data", () => c.end(payload)); + }); + const listening = Promise.withResolvers(); + server.once("error", listening.reject); + server.listen(0, "127.0.0.1", () => listening.resolve()); + await listening.promise; + const log: string[] = []; + let received = ""; + let secure = false; + const closed = Promise.withResolvers(); + const raw = createConnection({ + port: (server.address() as import("node:net").AddressInfo).port, + host: "127.0.0.1", + }); + const transport = new Duplex({ + // A TLSSocket takes allowHalfOpen from the stream it wraps, and with `true` node emits 'end' but never 'close': + // https://github.com/nodejs/node/blob/v26.3.0/lib/internal/tls/wrap.js#L592 + allowHalfOpen: false, + read() {}, + write(chunk, _encoding, callback) { + raw.write(chunk, callback); + }, + final(callback) { + raw.end(callback); + }, + }); + let client: Socket | undefined; + try { + raw.on("error", closed.reject); + raw.on("end", () => transport.push(null)); + raw.on("data", chunk => { + if (!secure) return void transport.push(chunk); + // One tick emits 'data' on the transport, a second one is queued right behind it. + process.nextTick(() => { + log.push("push begin"); + transport.push(chunk); + log.push("push end"); + }); + process.nextTick(() => log.push("next tick")); + }); + client = tlsConnect({ + socket: transport, + rejectUnauthorized: false, + onread: { + buffer: Buffer.alloc(4), + callback(n: number, buf: Buffer) { + received += buf.toString("latin1", 0, n); + }, + }, + }); + client.on("data", chunk => (received += chunk.toString("latin1"))); + client.on("secureConnect", () => { + secure = true; + client!.write("go"); + }); + client.on("error", closed.reject); + client.on("close", () => closed.resolve()); + await closed.promise; + expect(received).toBe(payload); + expect(log.slice(0, 3)).toEqual(["push begin", "push end", "next tick"]); + expect(log.join(",")).not.toContain("push begin,next tick"); + } finally { + client?.destroy(); + raw.destroy(); + server.close(); + } + }); + + it("a tick that throws during the checkpoint is an uncaught exception and the read goes on", async () => { + const fixture = /* js */ ` + const log = []; + process.on("uncaughtException", e => log.push("uncaught " + e.message)); + const net = require("net"); + const server = net.createServer(c => c.end("abcdefghijkl")); + server.listen(0, "127.0.0.1", () => { + const client = net.connect({ + port: server.address().port, + host: "127.0.0.1", + onread: { + buffer: Buffer.alloc(4), + callback(n, buf) { + const slice = buf.toString("latin1", 0, n); + log.push("callback " + slice); + process.nextTick(() => { throw new Error("tick-boom " + slice); }); + }, + }, + }); + client.on("error", e => log.push("socket-error " + e.message)); + client.on("close", () => { console.log(JSON.stringify(log)); server.close(); }); + }); + `; + await using proc = Bun.spawn({ + cmd: [bunExe(), "-e", fixture], + env: bunEnv, + stdout: "pipe", + stderr: "pipe", + }); + const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + expect({ log: stdout.trim() && JSON.parse(stdout), exitCode, failureDetail: exitCode === 0 ? "" : stderr }).toEqual( + { + log: slices.flatMap(slice => [`callback ${slice}`, `uncaught tick-boom ${slice}`]), + exitCode: 0, + failureDetail: "", + }, + ); + }); +}); + // On Windows the native layer does not report fatal send errors yet (the WSA // error translation is a follow-up), so the write error never surfaces there. it.skipIf(isWindows)("a write after the peer reset the connection fails with a write error", async () => {