diff --git a/src/js/node/http2.ts b/src/js/node/http2.ts index bd2cb588e233..9c7747d20737 100644 --- a/src/js/node/http2.ts +++ b/src/js/node/http2.ts @@ -2021,6 +2021,12 @@ function pushToStream(stream, data) { } } +// Like node's onSessionHeaders, a HEADERS frame with END_STREAM ends the readable before its event. +function endInboundHalf(stream: Http2Stream) { + if (!stream.rstCode) stream.rstCode = 0; + pushToStream(stream, null); +} + enum StreamState { EndedCalled = 1 << 0, // 00001 = 1 WantTrailer = 1 << 1, // 00010 = 2 @@ -3266,6 +3272,7 @@ class ServerHttp2Stream extends Http2Stream { if (pushedStream && pushedStream[bunHTTP2Headers] == null) { pushedStream[bunHTTP2Headers] = headers; } + if (pushedStream) endInboundHalf(pushedStream); if (onServerStreamCreatedChannel.hasSubscribers) { onServerStreamCreatedChannel.publish({ stream: pushedStream, headers }); } @@ -4057,10 +4064,7 @@ class ServerHttp2Session extends Http2Session { } if (state == 6 || state == 7) { if (stream.readable) { - if (!stream.rstCode) { - stream.rstCode = 0; - } - pushToStream(stream, null); + endInboundHalf(stream); // If the user hasn't tried to consume the stream then dump the incoming data so the // stream can finish — but at half-close only when nothing is buffered: a consumer may @@ -4149,7 +4153,9 @@ class ServerHttp2Session extends Http2Session { stream[kHeadRequest] = true; } const status = stream[bunHTTP2StreamStatus]; + const endOfStream = (flags & constants.NGHTTP2_FLAG_END_STREAM) !== 0; if ((status & StreamState.StreamResponded) !== 0) { + if (endOfStream) endInboundHalf(stream); stream.emit("trailers", headers, flags, rawheaders); } else { // Set the StreamResponded bit BEFORE dispatching the 'stream' event @@ -4166,6 +4172,7 @@ class ServerHttp2Session extends Http2Session { if (onServerStreamStartChannel.hasSubscribers) { onServerStreamStartChannel.publish({ stream, headers }); } + if (endOfStream) endInboundHalf(stream); // performServerHandshake() sessions have no owning server. self[kServer]?.emit("stream", stream, headers, flags, rawheaders); self.emit("stream", stream, headers, flags, rawheaders); @@ -5056,12 +5063,9 @@ class ClientHttp2Session extends Http2Session { } if (state == 6 || state == 7) { if (stream.readable) { - if (!stream.rstCode) { - stream.rstCode = 0; - } // Push a null so the stream can end whenever the client consumes // it completely. - pushToStream(stream, null); + endInboundHalf(stream); stream.read(0); } } @@ -5129,11 +5133,13 @@ class ClientHttp2Session extends Http2Session { } const status = stream[bunHTTP2StreamStatus]; const header_status = headers[HTTP2_HEADER_STATUS]; + const endOfStream = (flags & constants.NGHTTP2_FLAG_END_STREAM) !== 0; if (header_status === HTTP_STATUS_CONTINUE) { stream.emit("continue"); } if ((status & StreamState.StreamResponded) !== 0) { + if (endOfStream) endInboundHalf(stream); stream.emit("trailers", headers, flags, rawheaders); } else { if (header_status >= 100 && header_status < 200) { @@ -5151,6 +5157,7 @@ class ClientHttp2Session extends Http2Session { if (onClientStreamFinishChannel.hasSubscribers) { onClientStreamFinishChannel.publish({ stream, headers, flags }); } + if (endOfStream) endInboundHalf(stream); if (stream[kPush]) { // A pushed stream delivers its response via 'push'; the session 'stream' event already // fired (with the promised request headers) when the PUSH_PROMISE arrived. diff --git a/test/js/node/http2/node-http2-client-close.test.ts b/test/js/node/http2/node-http2-client-close.test.ts index 537639edc0b4..32e74457da20 100644 --- a/test/js/node/http2/node-http2-client-close.test.ts +++ b/test/js/node/http2/node-http2-client-close.test.ts @@ -1,7 +1,9 @@ /** - * ClientHttp2Stream.close(code) event contract: + * Http2Stream.close(code) event contract, client and server, while the peer's half is still open: * NO_ERROR / CANCEL -> 'end', 'close' (documented 'error' exemption) * any other code -> 'error', 'close' (no 'end': the body was killed by RST_STREAM) + * Once the peer's half has ended (END_STREAM on a HEADERS frame, or a server push, which has no + * inbound half) the readable already has its EOF, so 'end' comes first for every code. * * Works with both: * bun bd test test/js/node/http2/node-http2-client-close.test.ts @@ -183,10 +185,153 @@ describe("ClientHttp2Stream.close(code) while pending", () => { } }); +// The listeners below call close(code) once the peer's half of the stream has ended. Each one is +// the event of a HEADERS frame that carried END_STREAM (or the pushStream() callback: a server push +// has no inbound half). "server 'stream', request open" is the contrast: no END_STREAM yet. +type Site = + | "server 'stream'" + | "server 'stream' after respond()" + | "server 'stream', request open" + | "server 'trailers'" + | "pushStream() callback" + | "client 'response'" + | "client 'trailers'"; + +async function closeIn(site: Site, code: number, defer: boolean): Promise { + const events: string[] = []; + const { promise, resolve, reject } = Promise.withResolvers(); + const watch = (stream: http2.Http2Stream) => { + stream.on("end", () => events.push("end")); + stream.on("error", e => events.push("error:" + (e as NodeJS.ErrnoException).code)); + stream.on("close", () => { + events.push("close:" + stream.rstCode); + resolve(events); + }); + stream.resume(); + }; + const close = (stream: http2.Http2Stream) => { + if (defer) process.nextTick(() => stream.close(code)); + else stream.close(code); + }; + const ignoreErrors = (stream: http2.Http2Stream) => stream.on("error", () => {}); + + const server = http2.createServer(); + server.on("stream", stream => { + switch (site) { + case "server 'stream'": + case "server 'stream', request open": + watch(stream); + close(stream); + break; + case "server 'stream' after respond()": + watch(stream); + stream.respond({ ":status": 200 }, { endStream: true }); + close(stream); + break; + case "server 'trailers'": + watch(stream); + stream.respond({ ":status": 200 }); + stream.on("trailers", () => close(stream)); + break; + case "pushStream() callback": + ignoreErrors(stream); + stream.pushStream({ ":path": "/pushed" }, (err, pushed) => { + if (err) return reject(err); + watch(pushed); + close(pushed); + }); + stream.respond({ ":status": 200 }); + stream.end(); + break; + case "client 'response'": + ignoreErrors(stream); + stream.respond({ ":status": 204 }, { endStream: true }); + break; + case "client 'trailers'": + ignoreErrors(stream); + stream.respond({ ":status": 200 }, { waitForTrailers: true }); + stream.on("wantTrailers", () => stream.sendTrailers({ "x-trailer": "1" })); + stream.end("body"); + break; + } + }); + const port = await listen(server); + const client = http2.connect(`http://127.0.0.1:${port}`); + client.on("error", reject); + client.on("stream", pushed => ignoreErrors(pushed).resume()); + try { + switch (site) { + case "server 'trailers'": { + const req = client.request({ ":path": "/", ":method": "POST" }, { waitForTrailers: true }); + req.on("wantTrailers", () => req.sendTrailers({ "x-trailer": "1" })); + ignoreErrors(req).resume(); + req.end("body"); + break; + } + case "client 'response'": + case "client 'trailers'": { + // The request stays open, so nothing but close(code) can close the stream. + const req = client.request({ ":path": "/", ":method": "POST" }); + watch(req); + req.on(site === "client 'response'" ? "response" : "trailers", () => close(req)); + break; + } + case "server 'stream', request open": + ignoreErrors(client.request({ ":path": "/", ":method": "POST" })).resume(); + break; + default: + ignoreErrors(client.request({ ":path": "/" }, { endStream: true })).resume(); + } + return await promise; + } finally { + client.destroy(); + server.close(); + } +} + +const endThenClose = [ + [NGHTTP2_NO_ERROR, ["end", "close:0"]], + [NGHTTP2_CANCEL, ["end", "close:8"]], + [NGHTTP2_INTERNAL_ERROR, ["end", "error:ERR_HTTP2_STREAM_ERROR", "close:2"]], + [NGHTTP2_ENHANCE_YOUR_CALM, ["end", "error:ERR_HTTP2_STREAM_ERROR", "close:11"]], +] as const; +const noEndForErrorCodes = [ + [NGHTTP2_NO_ERROR, ["end", "close:0"]], + [NGHTTP2_CANCEL, ["end", "close:8"]], + [NGHTTP2_INTERNAL_ERROR, ["error:ERR_HTTP2_STREAM_ERROR", "close:2"]], + [NGHTTP2_ENHANCE_YOUR_CALM, ["error:ERR_HTTP2_STREAM_ERROR", "close:11"]], +] as const; + +for (const [site, table] of [ + ["server 'stream'", endThenClose], + ["server 'stream' after respond()", endThenClose], + ["server 'trailers'", endThenClose], + ["pushStream() callback", endThenClose], + ["client 'response'", endThenClose], + ["client 'trailers'", endThenClose], + ["server 'stream', request open", noEndForErrorCodes], +] as const) { + for (const defer of [false, true]) { + describe(`close(code) ${defer ? "a tick after" : "in"} ${site}`, () => { + for (const [code, expected] of table) { + test(`close(${code})`, async () => { + assert.deepStrictEqual(await closeIn(site, code, defer), expected); + }); + } + }); + } +} + if (typeof Bun !== "undefined") { const node = Bun.which("node"); + // Alpine's node segfaults at a random point of this file (alpine 3.23 aarch64, on main too), and + // the CI runner fails a file for any new core dump, whichever process wrote it. The glibc, macOS + // and Windows lanes keep the cross-check. + const isMusl = + process.platform === "linux" && + !(process.report.getReport() as { header: { glibcVersionRuntime?: string } }).header.glibcVersionRuntime; describe("Node.js compatibility", () => { - test("tests should run on node.js", { skip: !node }, async () => { + test("tests should run on node.js", { skip: !node || isMusl }, async () => { await using proc = Bun.spawn({ cmd: [node as string, "--test", import.meta.filename], stdout: "inherit",