Skip to content
38 changes: 26 additions & 12 deletions src/js/node/net.ts
Original file line number Diff line number Diff line change
Expand Up @@ -1565,15 +1565,20 @@ function Socket(options?) {
if (keepAliveInitialDelay < 0) keepAliveInitialDelay = 0;
}

if (options?.fd !== undefined) {
validateInt32(options.fd, "options.fd", 0);
// Like node's `options = { ...options }`: `fd`, `readable` and `writable` are
// read from own properties only, the same view the Duplex gets below.
// https://github.com/nodejs/node/blob/v26.3.0/lib/net.js#L410-L419
Comment thread
robobun marked this conversation as resolved.
Resolved
const fd = opts.fd;
if (fd !== undefined) {
validateInt32(fd, "options.fd", 0);
}

// `readable` / `writable` pass through to the Duplex like node's do: a
// `readable: false` socket starts with its readable side finished (no 'end',
// so no allowHalfOpen teardown of the writable side).
Comment thread
robobun marked this conversation as resolved.
Resolved
Duplex.$call(this, {
...opts,
allowHalfOpen,
readable: true,
writable: true,
//For node.js compat do not emit close on destroy.
emitClose: false,
autoDestroy: true,
Expand Down Expand Up @@ -1619,9 +1624,7 @@ function Socket(options?) {
// Shut down the socket when we're finished with it.
this.on("end", onSocketEnd);

if (options?.fd !== undefined) {
const { fd } = options;
validateInt32(fd, "fd", 0);
if (fd !== undefined) {
// Adopt pipe/character-device/file fds with synchronous writes. Matches
// node's effective semantics for stdio-style sockets: writes to a pipe
// complete inline, so data survives an immediate process.exit().
Expand All @@ -1632,7 +1635,7 @@ function Socket(options?) {
// here would end its readable side and stomp its write path.
// Network-socket fds are not supported (handle adoption needs native
// support); those keep the previous validated-but-inert behavior.
if (options.writable === true) {
if (opts.writable === true) {
let stats;
try {
stats = require("node:fs").fstatSync(fd);
Expand All @@ -1644,7 +1647,7 @@ function Socket(options?) {
// isSocket() covers stdio handed to a child as a socketpair (how spawn
// implements pipes on unix); writable-only adoption with sync write(2)
// is correct there too.
const optionsReadable = options.readable;
const optionsReadable = opts.readable;
if (
stats.isFIFO() ||
stats.isCharacterDevice() ||
Expand All @@ -1654,7 +1657,10 @@ function Socket(options?) {
this[kSyncWriteFd] = fd;
this._write = fdSyncWrite;
this._writev = fdSyncWritev;
if (optionsReadable !== true) {
// `readable: false` finished the readable side in the Duplex already.
// Reads on this path need a native pipe handle, so an unspecified
// `readable` still ends it here.
Comment thread
robobun marked this conversation as resolved.
Resolved
if (optionsReadable !== true && optionsReadable !== false) {
this.push(null);
this.read(0);
}
Expand Down Expand Up @@ -2337,6 +2343,9 @@ Socket.prototype.resume = function resume() {
// kOnreadDraining is still set and does not queue a second drain: Node's
// override sets handle.reading synchronously for the same reason.
const ret = Duplex.prototype.resume.$call(this);
// Nothing restarts the handle once the readable side has ended (EOF emitted,
// or built with `readable: false`): node only reaches readStart from _read.
Comment thread
robobun marked this conversation as resolved.
Resolved
if (this.readableEnded) return ret;
if (!this.connecting && !drainOnreadTail(this)) {
this._handle?.resume?.();
}
Expand Down Expand Up @@ -2436,7 +2445,8 @@ Socket.prototype[Symbol.for("::bunUpgradeServerTLS::")] = function (connection,
};

Socket.prototype.read = function read(size) {
if (!this.connecting && !drainOnreadTail(this, true)) {
// See resume(): an ended readable side never restarts the handle.
if (!this.readableEnded && !this.connecting && !drainOnreadTail(this, true)) {
this._handle?.resume?.();
restorePausedHold(this, this._handle);
}
Expand Down Expand Up @@ -2533,6 +2543,8 @@ Object.defineProperty(Socket.prototype, "remoteFamily", {

function fdSyncWrite(chunk, encoding, callback) {
const fs = require("node:fs");
// node's _writeGeneric restarts the idle timer on every write.
this._unrefTimer();
try {
const buf = typeof chunk === "string" ? Buffer.from(chunk, encoding) : chunk;
let offset = 0;
Expand All @@ -2550,6 +2562,7 @@ function fdSyncWrite(chunk, encoding, callback) {

function fdSyncWritev(data, callback) {
const fs = require("node:fs");
this._unrefTimer();
try {
let total = 0;
for (let i = 0; i < data.length; i++) {
Expand Down Expand Up @@ -3378,7 +3391,8 @@ function afterConnect(status, handle, req, readable, writable) {
}

// Ours already reads, Node's starts at read(): stop a paused plain socket now, and after the listeners unless one asked for a read.
const pausedBeforeConnect = self.isPaused();
// A socket built with `readable: false` never reads: read() cannot reach _read once the readable side has ended.
const pausedBeforeConnect = self.isPaused() || self.readableEnded;
if (pausedBeforeConnect && !self.encrypted) readStop(self, self._handle);

self.emit("connect");
Expand Down
10 changes: 9 additions & 1 deletion src/js/node/tls.ts
Original file line number Diff line number Diff line change
Expand Up @@ -744,7 +744,15 @@ function TLSSocket(socket?, options?) {

this._rejectUnauthorized = !!options.rejectUnauthorized;

NetSocket.$call(this, options);
// A TLS socket is always a full duplex: node builds the net.Socket options
// itself and never forwards the caller's `readable` / `writable`.
// https://github.com/nodejs/node/blob/v26.3.0/lib/internal/tls/wrap.js#L590-L600
Comment thread
robobun marked this conversation as resolved.
Resolved
NetSocket.$call(
this,
options.readable === undefined && options.writable === undefined
? options
: { ...options, readable: undefined, writable: undefined },
);

// Node's _init installs this as the first 'error' listener and removes it in
// _releaseControl: until control is handed to the user it routes errors
Expand Down
42 changes: 42 additions & 0 deletions test/js/bun/http/proxy.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,7 @@ import { afterAll, beforeAll, describe, expect, test } from "bun:test";
import { bunEnv, bunExe, isASAN, tls as tlsCert } from "harness";
import { HttpsProxyAgent } from "https-proxy-agent";
import { once } from "node:events";
import http from "node:http";
import net from "node:net";
import tls from "node:tls";
async function createProxyServer(is_tls: boolean) {
Expand Down Expand Up @@ -747,6 +748,47 @@ test("axios with https-proxy-agent", async () => {
expect(httpProxyServer.log).toEqual([`CONNECT localhost:${httpsServer.port}`]);
});

// For a refused CONNECT, https-proxy-agent hands node:http a detached
// `new net.Socket({ writable: false })` and replays the proxy's reply into it.
// node:http only parses from a socket that is not writable and never writes the
// request to it, so the caller sees the proxy's status. With `writable` forced
// to true the request was written to the handle-less socket and failed with
// ERR_SOCKET_CLOSED instead.
test("https-proxy-agent reports a refused CONNECT as the proxy's response, like node", async () => {
const proxy = net.createServer(socket => {
socket.on("error", () => {});
socket.once("data", () => {
socket.end(
'HTTP/1.1 407 Proxy Authentication Required\r\nProxy-Authenticate: Basic realm="test"\r\nContent-Length: 0\r\n\r\n',
);
});
});
await once(proxy.listen(0, "127.0.0.1"), "listening");
const agent = new HttpsProxyAgent(`http://127.0.0.1:${(proxy.address() as net.AddressInfo).port}`);
try {
const events: string[] = [];
const closed = Promise.withResolvers<void>();
const ended = Promise.withResolvers<void>();
const req = http.request({ host: "example.test", port: 80, path: "/", agent }, res => {
events.push(`response ${res.statusCode} ${res.headers["proxy-authenticate"]}`);
res.on("error", ended.reject);
res.on("end", () => {
events.push("res end");
ended.resolve();
});
res.resume();
});
req.on("error", err => events.push(`req error ${(err as NodeJS.ErrnoException).code}`));
req.on("close", () => closed.resolve());
req.end();
await Promise.all([closed.promise, ended.promise]);
Comment thread
robobun marked this conversation as resolved.
Resolved
expect(events).toEqual(['response 407 Basic realm="test"', "res end"]);
} finally {
agent.destroy();
proxy.close();
}
});

test("HTTPS proxy tunnel keep-alive reuses CONNECT across sequential requests", async () => {
httpProxyServer.log.length = 0;

Expand Down
185 changes: 185 additions & 0 deletions test/js/node/net/node-net.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@ import {
Stream,
} from "node:net";
import { join } from "node:path";
import { TLSSocket } from "node:tls";

const socket_domain = tmpdirSync();

Expand Down Expand Up @@ -1252,6 +1253,74 @@ it.skipIf(isWindows)(
60_000,
);

// node only reaches readStart() from _read(), and read(0) never calls _read once
// the readable side has ended, so a `readable: false` client leaves the peer's
// bytes unread. Bun's handle reads unless stopped, and pushing those bytes into
// the ended Readable raised ERR_STREAM_PUSH_AFTER_EOF.
it("a client dialed with readable: false never reads and keeps writing", async () => {
const done = Promise.withResolvers<string>();
const server = createServer(socket => {
socket.on("error", done.reject);
socket.setEncoding("utf8");
let got = "";
socket.on("data", chunk => {
got += chunk;
if (got.endsWith("second\n")) done.resolve(got);
});
socket.write("banner\n");
});
await once(server.listen(0, "127.0.0.1"), "listening");
try {
const client = connect({
port: (server.address() as import("node:net").AddressInfo).port,
host: "127.0.0.1",
readable: false,
});
client.on("error", done.reject);
const events: string[] = [];
for (const name of ["data", "end", "close"]) client.on(name, () => events.push(name));
await once(client, "connect");
client.write("first\n");
// The banner is in flight the whole time; give the loop several turns to
// (wrongly) read it before the second write proves the socket still works.
for (let i = 0; i < 5; i++) await new Promise<void>(resolve => setImmediate(resolve));
client.resume();
client.write("second\n");
expect(await done.promise).toBe("first\nsecond\n");
expect({
events,
destroyed: client.destroyed,
readable: client.readable,
readableEnded: client.readableEnded,
}).toEqual({
events: [],
destroyed: false,
readable: false,
readableEnded: true,
});
client.destroy();
await once(client, "close");
} finally {
server.close();
}
});

it("passes readable / writable through to the Duplex like node (a TLSSocket is always a full duplex)", () => {
// Values observed under node v26.3.0.
const a = new Socket({ readable: false });
const b = new Socket({ writable: false });
const c = new TLSSocket(undefined, { readable: false, writable: false });
expect({
a: [a.readable, a.writable, a.readableEnded],
b: [b.readable, b.writable, b.writableEnded, b.writableFinished],
c: [c.readable, c.writable],
}).toEqual({
a: [false, true, true],
b: [true, false, true, true],
c: [true, true],
});
});

describe("Socket fd adoption", () => {
it("writes synchronously to an adopted fd and closes it (> 2) on destroy", async () => {
const path = join(tmpdirSync(), "adopted-fd.txt");
Expand All @@ -1270,6 +1339,72 @@ describe("Socket fd adoption", () => {
expect(() => fs.fstatSync(fd)).toThrow();
});

// node wraps a piped stdout/stderr exactly this way (new Socket({ fd, readable:
// false, writable: true })). The readable side must start out finished without
// an 'end' event; emitting one let allowHalfOpen=false end the writable side,
// destroy the socket and close the fd one tick after construction.
it("readable: false leaves the writable side and the adopted fd open", async () => {
const path = join(tmpdirSync(), "adopted-write-only.txt");
const fd = fs.openSync(path, "w");
const socket = new Socket({ fd, readable: false, writable: true });
const events: string[] = [];
for (const name of ["end", "finish", "close"]) socket.on(name, () => events.push(name));
socket.on("error", err => events.push(`error:${(err as NodeJS.ErrnoException).code}`));

// The teardown being guarded against ran purely on process.nextTick
// (end -> finish -> destroy -> close), so two immediate turns are past it.
await new Promise<void>(resolve => setImmediate(() => setImmediate(resolve)));
expect(events).toEqual([]);
expect(socket.destroyed).toBe(false);
expect(fs.fstatSync(fd).isFile()).toBe(true);
expect([socket.readable, socket.readableEnded, socket.writable]).toEqual([false, true, true]);

await new Promise<void>((resolve, reject) => socket.write("late", err => (err ? reject(err) : resolve())));
const closed = once(socket, "close");
socket.end();
await closed;
expect(events).toEqual(["finish", "close"]);
expect(fs.readFileSync(path, "utf8")).toBe("late");
let closeError: NodeJS.ErrnoException | undefined;
try {
fs.fstatSync(fd);
} catch (e) {
closeError = e as NodeJS.ErrnoException;
}
expect(closeError?.code).toBe("EBADF");
});

it("a write-only socket over the process's piped stdout keeps delivering later writes", async () => {
await using proc = Bun.spawn({
cmd: [
bunExe(),
"-e",
`
const net = require("node:net");
Comment thread
coderabbitai[bot] marked this conversation as resolved.
Resolved
const s = new net.Socket({ fd: 1, readable: false, writable: true });
const events = [];
for (const n of ["end", "finish", "close"]) s.on(n, () => events.push(n));
s.on("error", e => events.push("error:" + e.code));
s.write("now\\n");
setImmediate(() => setImmediate(() => {
s.write("late\\n", err => {
console.error(JSON.stringify({ events, destroyed: s.destroyed, cb: err ? err.code : null }));
});
}));
`,
],
env: bunEnv,
stdout: "pipe",
stderr: "pipe",
});
const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]);
expect({ stdout, report: JSON.parse(stderr.trim().split("\n").pop()!), exitCode }).toEqual({
stdout: "now\nlate\n",
report: { events: [], destroyed: false, cb: null },
exitCode: 0,
});
});

it("throws ERR_INVALID_FD_TYPE for a writable fd that cannot be fstat'ed", () => {
let error: any;
try {
Expand All @@ -1286,6 +1421,56 @@ describe("Socket fd adoption", () => {
// extra stdio relies on this path (connect({ fd }) attaches natively).
expect(() => new Socket({ fd: 0x7ffff })).not.toThrow();
});

// node's _writeGeneric restarts the idle timer before every write; the
// synchronous fd write path has no handle doing that for it.
it("writes to an adopted fd restart the setTimeout() idle timer", async () => {
const path = join(tmpdirSync(), "adopted-timeout.txt");
const fd = fs.openSync(path, "w");
const socket = new Socket({ fd, readable: false, writable: true });
try {
const timeouts: number[] = [];
const start = performance.now();
socket.on("timeout", () => timeouts.push(performance.now() - start));
socket.setTimeout(500);
// Two full timeout periods of steady writes: never 500ms idle.
for (let i = 0; i < 20; i++) {
socket.write("x");
await Bun.sleep(50);
}
expect(timeouts).toEqual([]);
Comment thread
robobun marked this conversation as resolved.
Resolved
// The timer is still armed from the last write and fires once it goes idle.
await once(socket, "timeout");
expect(fs.readFileSync(path, "utf8")).toBe("x".repeat(20));
Comment thread
coderabbitai[bot] marked this conversation as resolved.
Resolved
} finally {
socket.destroy();
}
});

// node copies the options ({ ...options }) before reading fd / readable /
// writable, so inherited properties never adopt anything; the Duplex flags and
// the adoption decision must come from the same view.
it("ignores fd / readable / writable that are not own properties of the options", async () => {
const path = join(tmpdirSync(), "inherited-fd.txt");
const fd = fs.openSync(path, "w");
try {
const socket = new Socket(Object.create({ fd, readable: false, writable: true }));
const readableAfterConstruct = socket.readable;
socket.on("error", () => {});
const writeError = await new Promise<string | undefined>(resolve =>
socket.write("x", err => resolve((err as NodeJS.ErrnoException | null)?.code)),
);
// No handle and nothing adopted: the write fails the way node's does.
expect({ writeError, readableAfterConstruct, file: fs.readFileSync(path, "utf8") }).toEqual({
writeError: "ERR_SOCKET_CLOSED",
readableAfterConstruct: true,
file: "",
});
expect(fs.fstatSync(fd).isFile()).toBe(true);
} finally {
fs.closeSync(fd);
}
});
});

describe.concurrent("socket that already sent FIN and is paused with unread data", () => {
Expand Down
Loading