Skip to content
Open
Show file tree
Hide file tree
Changes from 3 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 5 additions & 4 deletions src/js/node/net.ts
Original file line number Diff line number Diff line change
Expand Up @@ -2122,8 +2122,6 @@ Socket.prototype.connect = function connect(...args) {
}
if (this.destroyed) {
this._handle = null;
this._peername = null;
this._sockname = null;
}

this.connecting = true;
Expand All @@ -2134,8 +2132,10 @@ Socket.prototype.connect = function connect(...args) {

if (!this._handle) {
this._handle = newDetachedSocket(typeof this[bunTlsSymbol] === "function");
initSocketHandle(this);
}
// Also for a reused handle: doConnect replaces the connection it still
// carries, so the stream state of that connection has to go with it.

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

If you need a paragraph-long comment to justify why the workaround is OK, the code is wrong — fix the code

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Trimmed to one line in 5314d9e.

initSocketHandle(this);
Comment thread
claude[bot] marked this conversation as resolved.
Comment on lines +2136 to +2137

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🟡 The hostname variant of this fix has a race window: initSocketHandle(this) resets state synchronously here, but for a non-IP host lookupAndConnect defers internalConnectdetach_for_reconnect to an async DNS callback, so the old still-attached connection can deliver end/close (SocketHandlers2 has no self.connecting guard) between the reset and the detach and re-set kended/self.write=writeAfterFIN/push(null). Not a regression — before this PR the hostname case failed 100% of the time — but every new test uses "127.0.0.1" (the process.nextTick fast path), so the gap is untested. Guarding finishSocketEnd/close on self.connecting, or repeating the reset in internalConnect right before doConnect, would close the window; fine as a follow-up.

Extended reasoning...

What the gap is

Socket.prototype.connect now runs initSocketHandle(this) unconditionally at net.ts:2137, which _undestroy()s the Duplex and clears kended/kclosed/_peername. That happens synchronously. The old native connection, however, is not detached until internalConnectdoConnectdetach_for_reconnect() runs. For an IP-literal host that is scheduled via process.nextTick (net.ts:2896) — no I/O poll happens between a nextTick queue flush and the code that scheduled it, so the window is closed. For a unix path it is synchronous. But for a hostname host, lookupAndConnect defers internalConnect to an asynchronous dns.lookup callback (net.ts:2938-2953), and the event loop does poll I/O in between.

The code path that re-corrupts the reset state

During that DNS window the old (still-attached) connection can deliver its FIN or close to SocketHandlers2:

  • SocketHandlers2.end (net.ts:1304-1309) calls finishSocketEnd(self) with no self.connecting guard. finishSocketEnd (net.ts:589-606) checks only self[kended] — which initSocketHandle just cleared to false — so it runs: self[kended] = true, self.write = writeAfterFIN (default allowHalfOpen: false), self.push(null) (readable side re-ended), and socket.unref().
  • SocketHandlers2.close (net.ts:1326-1368) similarly has no self.connecting guard, sets self[kclosed] = true, and calls finishSocketEnd. A knock-on: with kclosed already true, the new connection's later close hits if (self[kclosed]) return; at line 1330 and is silently dropped.

The this.write restore at net.ts:2119-2122 already ran before this, so it does not undo the re-installed writeAfterFIN. internalConnect and afterConnect do not repeat the reset.

Why the existing tests don't cover it

Every new test in this PR — the rewritten "should allow reconnecting after end()" and all four "connect() while the previous connection is half-closed" cases — connects to "127.0.0.1" (or a unix path). isIP("127.0.0.1") is truthy, so lookupAndConnect takes the process.nextTick fast path at line 2896, and no I/O poll can interleave between initSocketHandle and detach_for_reconnect. A test that reconnects to "localhost" (or via a custom options.lookup that resolves asynchronously) while the previous peer's FIN is still in flight would exercise the window.

Step-by-step trace

  1. socket.end(); server keeps its side open (allowHalfOpen: true), so the client is half-closed and socket.destroyed === false.
  2. socket.connect(port, "localhost") runs. this.write === writeAfterFIN is restored (line 2119). this.connecting = true. initSocketHandle(this) runs _undestroy(), sets kended = false, kclosed = false. lookupAndConnect sees isIP("localhost") === 0 and calls dns.lookup(...).
  3. Control returns to the event loop; the loop polls I/O. The old connection's peer now closes its side (or the FIN that was already on the wire is read). SocketHandlers2.end fires → finishSocketEnd(self): kended = true, self.write = writeAfterFIN, self.push(null).
  4. The DNS callback fires → internalConnectdoConnectdetach_for_reconnect (old connection finally detached) → new connection established → afterConnect emits 'connect'.
  5. socket.write("data") is now writeAfterFIN. push(null) on a !allowHalfOpen Duplex triggered auto-end(), so writableEnded is true again and the write fails — same user-visible symptom this PR set out to fix, just via a narrower race and a different code (EPIPE instead of ERR_STREAM_WRITE_AFTER_END).

Why this is a nit, not a blocker

  • Not a regression. Before this PR initSocketHandle was never called on the reused-handle path, so the hostname case failed 100% of the time with ERR_STREAM_WRITE_AFTER_END. After this PR it fails only when the old FIN/close races into the DNS window. Strict improvement on every path.
  • Bun-specific extension. Node rejects this whole pattern with EISCONN (per the PR description); reconnecting a half-closed net.Socket to a hostname while the old peer's FIN is in flight is a narrow edge of a Bun-only behavior.
  • The proper fix is a bit larger than this PR's scope: either guard finishSocketEnd / the close handler on self.connecting for the reused-handle case (so late old-connection events are ignored), or detach the old connection synchronously in connect(), or repeat initSocketHandle inside internalConnect just before doConnect (after detach_for_reconnect has nulled the old ext slot so no further JS callbacks can arrive from it). Any of these is a reasonable follow-up.

Per REVIEW.md's "Cover the variant matrix, not just the repro" this is worth noting — the hostname variant is an untested gap — but not worth blocking a PR that fixes the flaky test it set out to fix and strictly improves every case.


if (!pipe) {
lookupAndConnect(this, options);
Expand Down Expand Up @@ -4261,10 +4261,11 @@ function normalizeArgs(args: unknown[]): [options: Record<PropertyKey, any>, cb:
return arr;
}

// Called when creating new Socket, or when re-using a closed Socket
// Called on every connect(): a new Socket, or one re-used for another connection.
function initSocketHandle(self) {
self._undestroy();
self._sockname = null;
self._peername = null;
self[kclosed] = false;
self[kended] = false;

Expand Down
228 changes: 208 additions & 20 deletions test/js/node/net/node-net.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -463,33 +463,221 @@ describe("net.Socket write", () => {
}),
);

it("should allow reconnecting after end()", async () => {
const server = new Server(socket => socket.end());
const port = await new Promise(resolve => {
server.once("listening", () => resolve(server.address().port));
server.listen();
function listen(server: Server) {
return new Promise<number>((resolve, reject) => {
server.once("error", reject);
server.listen(0, "127.0.0.1", () => resolve((server.address() as import("node:net").AddressInfo).port));
});
}
Comment thread
claude[bot] marked this conversation as resolved.

it("should allow reconnecting after end()", async () => {
// #7325: the same net.Socket is reconnected from end()'s callback. That
// callback runs before the peer's FIN has been read, so the socket is not
// destroyed yet and connect() reuses the handle of the half-closed
// connection. (This used to reconnect 3ms after the callback instead, which
// only passed when the previous connection had finished closing by then.)
const iterations = 10;
const connections: Socket[] = [];
const server = createServer(c => {
connections.push(c);
c.on("error", () => {});
c.end();
});
const socket = new Socket();
socket.on("data", data => console.log(data.toString()));
socket.on("error", err => console.error(err));

async function run() {
return new Promise((resolve, reject) => {
socket.once("connect", (...args) => {
socket.write("script\n", err => {
if (err) return reject(err);
socket.end(() => setTimeout(resolve, 3));
});
const errors: Error[] = [];
socket.on("error", err => errors.push(err));
socket.resume();
const results: unknown[] = [];
try {
const port = await listen(server);
for (let i = 0; i < iterations; i++) {
socket.connect(port, "127.0.0.1");
await once(socket, "connect");
const readyState = socket.readyState;
const writeError = await new Promise<Error | null | undefined>(resolve => socket.write("script\n", resolve));
const endError = await new Promise<Error | null | undefined>(resolve => socket.end(resolve));
results.push({ readyState, writeError: writeError?.message ?? null, endError: endError?.message ?? null });
}
expect(results).toEqual(Array(iterations).fill({ readyState: "open", writeError: null, endError: null }));
expect(errors).toEqual([]);
} finally {
socket.destroy();
for (const c of connections) c.destroy();
server.close();
}
});

describe("connect() while the previous connection is half-closed", () => {
// connect() on a socket that has not been destroyed replaces the connection
// underneath the same handle; the per-connection stream state has to start
// over with it. One side of each connection stays open (allowHalfOpen on
// the server, or on the client) so that the client socket is still not
// destroyed when it is reconnected.
it("re-opens the writable side after end()", async () => {
const connections: Socket[] = [];
const received: { connection: number; data: string }[] = [];
let onData: ((data: string) => void) | undefined;
const server = createServer({ allowHalfOpen: true }, c => {
connections.push(c);
const connection = connections.length;
c.on("error", () => {});
c.setEncoding("utf8");
c.on("data", (data: string) => {
received.push({ connection, data });
onData?.(data);
});
});
const socket = new Socket();
const errors: Error[] = [];
socket.on("error", err => errors.push(err));
function writeAndWaitForServer(data: string) {
const { promise, resolve } = Promise.withResolvers<string>();
onData = resolve;
socket.write(data);
return promise;
}
try {
const port = await listen(server);

socket.connect(port, "127.0.0.1");
await once(socket, "connect");
await writeAndWaitForServer("first");
socket.end();
await once(socket, "finish");
expect({
destroyed: socket.destroyed,
writableEnded: socket.writableEnded,
readyState: socket.readyState,
}).toEqual({ destroyed: false, writableEnded: true, readyState: "readOnly" });

socket.connect(port, "127.0.0.1");
await once(socket, "connect");
expect({
writableEnded: socket.writableEnded,
writableFinished: socket.writableFinished,
readyState: socket.readyState,
}).toEqual({ writableEnded: false, writableFinished: false, readyState: "open" });
await writeAndWaitForServer("second");
expect(received).toEqual([
{ connection: 1, data: "first" },
{ connection: 2, data: "second" },
]);
expect(errors).toEqual([]);
} finally {
socket.destroy();
for (const c of connections) c.destroy();
server.close();
}
});

it("re-opens the writable side after end() (unix socket path)", async () => {
const connections: Socket[] = [];
const server = createServer({ allowHalfOpen: true }, c => {
connections.push(c);
c.on("error", () => {});
});
}
const socketPath = join(socket_domain, "reconnect.sock");
const socket = new Socket();
const errors: Error[] = [];
socket.on("error", err => errors.push(err));
try {
await new Promise<void>((resolve, reject) => {
server.once("error", reject);
server.listen(socketPath, resolve);
});

for (let i = 0; i < 10; i++) {
await run();
}
server.close();
socket.connect(socketPath);
await once(socket, "connect");
socket.end();
await once(socket, "finish");

socket.connect(socketPath);
await once(socket, "connect");
const writeError = await new Promise<Error | null | undefined>(resolve => socket.write("again", resolve));
expect({ readyState: socket.readyState, writeError: writeError?.message ?? null }).toEqual({
readyState: "open",
writeError: null,
});
expect(errors).toEqual([]);
} finally {
socket.destroy();
for (const c of connections) c.destroy();
server.close();
}
});

it("re-opens the readable side after the peer ended an allowHalfOpen socket", async () => {
const connections: Socket[] = [];
const server = createServer(c => {
connections.push(c);
c.on("error", () => {});
c.end();
});
const socket = new Socket({ allowHalfOpen: true });
const errors: Error[] = [];
socket.on("error", err => errors.push(err));
socket.resume();
try {
const port = await listen(server);

socket.connect(port, "127.0.0.1");
await once(socket, "end");
expect({
destroyed: socket.destroyed,
readableEnded: socket.readableEnded,
readyState: socket.readyState,
}).toEqual({ destroyed: false, readableEnded: true, readyState: "writeOnly" });

const endedAgain = once(socket, "end");
socket.connect(port, "127.0.0.1");
await once(socket, "connect");
expect({ readableEnded: socket.readableEnded, readyState: socket.readyState }).toEqual({
readableEnded: false,
readyState: "open",
});
// The new connection's FIN is reported as a fresh 'end'.
await endedAgain;
expect(connections).toHaveLength(2);
expect(errors).toEqual([]);
} finally {
socket.destroy();
for (const c of connections) c.destroy();
server.close();
}
});

it("reports the address of the new peer", async () => {
const connections: Socket[] = [];
const onConnection = (c: Socket) => {
connections.push(c);
c.on("error", () => {});
};
const server1 = createServer({ allowHalfOpen: true }, onConnection);
const server2 = createServer({ allowHalfOpen: true }, onConnection);
const socket = new Socket();
const errors: Error[] = [];
socket.on("error", err => errors.push(err));
try {
const port1 = await listen(server1);
const port2 = await listen(server2);

socket.connect(port1, "127.0.0.1");
await once(socket, "connect");
expect(socket.remotePort).toBe(port1);
socket.end();
await once(socket, "finish");

socket.connect(port2, "127.0.0.1");
await once(socket, "connect");
expect(socket.remotePort).toBe(port2);
expect(errors).toEqual([]);
} finally {
socket.destroy();
for (const c of connections) c.destroy();
server1.close();
server2.close();
}
});
});

// Client-mode `Handlers.markInactive()` frees the per-connection Handlers
Expand Down
Loading