Skip to content
Open
Show file tree
Hide file tree
Changes from all 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/internal/http1_server_fallback.ts
Original file line number Diff line number Diff line change
Expand Up @@ -665,7 +665,8 @@ function connectionListenerHTTP1(server, socket, options) {
function onHttp1SocketDrain() {
resumeFallbackReadsOnDrain(socket);
// Node's socketOnDrain then emits 'drain' on the response that waits for it.
socket._httpMessage?.[kHttp1ResponseHandle]?.socketDrained();
// The response in flight can be one of a native listener, whose handle has none of the methods called here and below: a CONNECT behind it was handed off, and the socket given to emit("connection").
socket._httpMessage?.[kHttp1ResponseHandle]?.socketDrained?.();
}
socket.on("data", onHttp1SocketData);
socket.on("error", onHttp1SocketErrorListener);
Expand All @@ -674,17 +675,17 @@ function connectionListenerHTTP1(server, socket, options) {
socket.once("finish", () => {
const inflight = socket._httpMessage;
// An ended socket emits no 'drain', and all that the response wrote is out: the write callbacks that wait succeed.
inflight?.[kHttp1ResponseHandle]?.socketEnded(() => inflight._callPendingCallbacks());
inflight?.[kHttp1ResponseHandle]?.socketEnded?.(() => inflight._callPendingCallbacks());
});
// Prepended: like Node's 'finish' after a failed last write, a draining response finishes (and detaches) before the other 'close' listeners run.
socket.prependOnceListener("close", () => {
socket._httpMessage?.[kHttp1ResponseHandle]?.flushed();
socket._httpMessage?.[kHttp1ResponseHandle]?.flushed?.();
});
socket.once("close", () => {
connections.delete(socket);
const inflight = socket._httpMessage;
// The write callbacks that still wait fail, before the response's 'close', as in Node.
inflight?.[kHttp1ResponseHandle]?.socketEnded(() =>
inflight?.[kHttp1ResponseHandle]?.socketEnded?.(() =>
failPendingWriteCallbacks(inflight, socket.errored ?? inflight.errored ?? $ERR_STREAM_DESTROYED("write")),
);
// Like the native socket's close path (Node's socketOnClose ->
Expand Down
62 changes: 62 additions & 0 deletions test/js/node/http/node-http-connect.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -1586,6 +1586,68 @@ describe("CONNECT pipelined behind a pending response", () => {
firstFinished: false,
});
});

// The 'connect' listener gives the socket to a second server with emit("connection"), like a proxy
// that serves some tunnels in its own process. The response to /first never ends, so it still has
// the socket then. The listeners that the second server puts on the socket called methods that
// only its own responses have: an uncaught TypeError ended the process. Node v26.3.0 prints the
// same two lines and exits with code 0.
test.concurrent("should leave the response ahead alone when a second server gets the socket", async () => {
const fixture = /* js */ `
const http = require("node:http");
const net = require("node:net");
// Resolves when the tunnel socket has closed. \`event\` is what happens to it first.
function run(event) {
const { promise, resolve } = Promise.withResolvers();
const second = http.createServer();
const server = http.createServer((req, res) => void res.write("first"));
server.on("connect", (req, socket) => {
socket.once(event, () => {
console.log(event);
socket.destroy();
server.close();
});
socket.once("close", resolve);
socket.write("HTTP/1.1 200 Connection Established\\r\\n\\r\\n");
second.emit("connection", socket);
if (event === "drain") {
// A write that returns false is followed by 'drain' when the client has read it.
const chunk = Buffer.alloc(1 << 20, "x");
while (socket.write(chunk)) {}
}
});
server.listen(0, "127.0.0.1", () => {
const client = net.connect(server.address().port, "127.0.0.1");
let head = "";
client.on("data", chunk => {
if (event !== "close" || head.includes("Established")) return;
head += chunk.toString("latin1");
if (head.includes("Established")) client.destroy();
});
client.on("error", () => {});
client.write(
"GET /first HTTP/1.1\\r\\nHost: example.com\\r\\n\\r\\n" +
"CONNECT example.com:443 HTTP/1.1\\r\\nHost: example.com:443\\r\\n\\r\\n",
);
});
return promise;
}
run("drain").then(() => run("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({ stdout, stderr, exitCode, signalCode: proc.signalCode }).toEqual({
stdout: "drain\nclose\n",
stderr: "",
exitCode: 0,
signalCode: null,
});
});
});

/**
Expand Down
Loading