diff --git a/src/js/node/http2.ts b/src/js/node/http2.ts index f41ce68605fc..8b6148d98a74 100644 --- a/src/js/node/http2.ts +++ b/src/js/node/http2.ts @@ -27,6 +27,7 @@ * Modifications were made to the original code. */ const { isTypedArray } = require("node:util/types"); +const { aggregateTwoErrors } = require("internal/errors"); const { hideFromStack, hasObserver, enqueueNodeEntry, PerformanceNodeEntry } = require("internal/shared"); const { STATUS_CODES, utcDate } = require("internal/http"); const { kTimeout, getTimerDuration } = require("internal/timers"); @@ -2060,6 +2061,8 @@ enum StreamState { // callback). Until then no 'error' listener can exist, so stream errors must not be emitted: // node never constructs the JS stream object before a complete header block arrives. Delivered = 1 << 8, // 100000000 = 256 + // close(code) stored its code, before its listeners ran. _destroy keeps that code, NGHTTP2_NO_ERROR included. + CloseRecorded = 1 << 9, // 1000000000 = 512 } // native.writeStream() return-value flag (mirrors WRITE_FLUSHED_WITHOUT_CALLBACK in // h2_frame_parser.rs): the chunk was handed to the socket without queueing and the engine did @@ -2226,6 +2229,10 @@ function markStreamClosed(stream: Http2Stream) { markWritableDone(stream); } } +// True while close() runs its listeners and end(): the code it stored is then not the code of a reset that happened. +function isInsideClose(status: number) { + return (status & (StreamState.Closed | StreamState.CloseRecorded)) === StreamState.CloseRecorded; +} function rstNextTick(id: number, rstCode: number) { const session = this as Http2Session; session[bunHTTP2Native]?.rstStream(id, rstCode); @@ -2273,7 +2280,9 @@ function setupRequestEndAndSignal(req: Http2Stream, options: any, signal: AbortS // an 'error' event), and stream teardown must not wait a tick. function destroyStreamForSessionDestroy(error: Error | undefined, rstCode: number, stream: Http2Stream) { if (stream.destroyed) return; - if (rstCode && !stream.rstCode) stream.rstCode = rstCode; + if (rstCode && !stream.rstCode && (stream[bunHTTP2StreamStatus] & StreamState.CloseRecorded) === 0) { + stream.rstCode = rstCode; + } // A clean session teardown can reach here while a stream's already-received request data is // still being drained by its consumer: the native side is done with the stream (state 7, // which is what allowed the session to start destroying), but the JS side is not. Destroying @@ -2570,9 +2579,22 @@ class Http2Stream extends (Duplex as Http2StreamBase) { validateFunction(callback, "callback"); this.once("close", callback); } + // A listener of a close() that is still running called close(): the first code stays, as in node. + if ((this[bunHTTP2StreamStatus] & StreamState.CloseRecorded) !== 0) return; + // Stored before 'aborted', like https://github.com/nodejs/node/blob/v26.3.0/lib/internal/http2/core.js#L2013 + this.rstCode = code; + this[bunHTTP2StreamStatus] |= StreamState.CloseRecorded; + let thrown: unknown; + let threw = false; if (this.pending || code === NGHTTP2_NO_ERROR || code === NGHTTP2_CANCEL) { // For other rstCodes _destroy ends the readable so 'end' is suppressed. - this.push(null); + try { + this.push(null); + } catch (err) { + // A listener that throws does not stop the close: node main has submitted its reset by then (nodejs/node#66314). + thrown = err; + threw = true; + } } const { ending } = this._writableState; if (!ending) { @@ -2580,7 +2602,12 @@ class Http2Stream extends (Duplex as Http2StreamBase) { // 'aborted' event and set the aborted flag. if (!this.aborted) { this[kAborted] = true; - this.emit("aborted"); + try { + this.emit("aborted"); + } catch (err) { + thrown = threw ? aggregateTwoErrors(err as Error, thrown as Error) : err; + threw = true; + } } this.end(); } @@ -2591,7 +2618,8 @@ class Http2Stream extends (Duplex as Http2StreamBase) { // RST_STREAM has to be sent after the HEADERS frame, once the id is assigned. this.once("ready", sendRstOnReady.bind(this, session, code)); } else if (this.writableFinished || code) { - setImmediate(rstNextTick.bind(session, this.#id, code)); + // A listener that destroyed the stream made _destroy queue this reset, with the code stored above. + if (this[bunHTTP2Session] !== null) setImmediate(rstNextTick.bind(session, this.#id, code)); } else { this.once("finish", rstNextTick.bind(session, this.#id, code)); } @@ -2602,6 +2630,7 @@ class Http2Stream extends (Duplex as Http2StreamBase) { } else { this.once("finish", scheduleDestroyIfNotDestroyed.bind(null, this)); } + if (threw) throw thrown; } } _destroy(err, callback) { @@ -2609,6 +2638,8 @@ class Http2Stream extends (Duplex as Http2StreamBase) { // push(null)) and a throwing listener would otherwise skip the clear and // leave a retained stream pinning the store. this[bunHTTP2AsyncContextFrame] = undefined; + // Read before 'aborted' below: a close(0) from that listener is not a stored NGHTTP2_NO_ERROR for this destroy. + const closeRecorded = (this[bunHTTP2StreamStatus] & StreamState.CloseRecorded) !== 0; const { ending } = this._writableState; this.push(null); // A pushed stream's request was synthesized by the server, so its local (writable) half is @@ -2631,7 +2662,8 @@ class Http2Stream extends (Duplex as Http2StreamBase) { assertSession(session); let rstCode = this.rstCode; - if (!rstCode) { + // A pushed stream does not keep NGHTTP2_NO_ERROR from close(): native rstStream(id, 0) writes nothing for its id. + if (rstCode === undefined || (rstCode === 0 && (!closeRecorded || this[kPush]))) { if (err != null) { if (err.code === "ABORT_ERR") { // Enables using AbortController to cancel requests with RST code 8. @@ -4111,11 +4143,12 @@ class ServerHttp2Session extends Http2Session { } // 7 = closed, in this case we already send everything and received everything if (state === 7) { - stream[bunHTTP2StreamStatus] |= StreamState.NativeClosed; + const status = stream[bunHTTP2StreamStatus]; + stream[bunHTTP2StreamStatus] = status | StreamState.NativeClosed; markStreamClosed(stream); self.#connections--; if (stream.id % 2 === 1) self.#peerInitiatedStreams--; - if (stream.readable && !stream.rstCode) { + if (stream.readable && (!stream.rstCode || isInsideClose(status))) { // Clean close while data is still buffered on the readable side (e.g. the response // ended before the request body was consumed): node defers the destroy until the // consumer drains it ('end'), so the buffered request body is not lost. @@ -5112,10 +5145,11 @@ class ClientHttp2Session extends Http2Session { // 7 = closed, in this case we already send everything and received everything if (state === 7) { - stream[bunHTTP2StreamStatus] |= StreamState.NativeClosed; + const status = stream[bunHTTP2StreamStatus]; + stream[bunHTTP2StreamStatus] = status | StreamState.NativeClosed; markStreamClosed(stream); self.#connections--; - if (stream.readable && !stream.rstCode) { + if (stream.readable && (!stream.rstCode || isInsideClose(status))) { // Clean close while data is still buffered on the readable side: node defers the // destroy until the consumer drains it ('end'), so a late-attaching reader does not // lose data. @@ -5159,7 +5193,8 @@ class ClientHttp2Session extends Http2Session { headersTuple: [string[], Record, string[] | undefined], flags: number, ) => { - if (!self || typeof stream !== "object" || stream.rstCode) return; + if (!self || typeof stream !== "object") return; + if (stream.rstCode && !isInsideClose(stream[bunHTTP2StreamStatus])) return; let rawheaders = headersTuple[0]; let headers = headersTuple[1]; if (self.#strictFieldWhitespaceValidation) { diff --git a/test/js/node/http2/h2-conformance.test.ts b/test/js/node/http2/h2-conformance.test.ts index 4a5b6719ea1a..7305156b5076 100644 --- a/test/js/node/http2/h2-conformance.test.ts +++ b/test/js/node/http2/h2-conformance.test.ts @@ -12,7 +12,7 @@ import { bunEnv, bunExe, gcTick, normalizeBunSnapshot } from "harness"; import { once } from "node:events"; import http2 from "node:http2"; import net from "node:net"; -import { Writable } from "node:stream"; +import { PassThrough, Writable, duplexPair, finished, pipeline, promises } from "node:stream"; const PREFACE = Buffer.from("PRI * HTTP/2.0\r\n\r\nSM\r\n\r\n", "latin1"); @@ -2166,3 +2166,588 @@ describe("stream-reset floods (CVE-2023-44487 rapid reset, CVE-2025-8671 MadeYou }); }); }); + +// The RST_STREAM of close(code) carries the given code, also when a listener of close(code) +// destroys the stream. That is the result of node main (nodejs/node#66314, its test is +// test/parallel/test-http2-close-aborted-destroy.js). node v26.3.0 writes no RST_STREAM there. +describe("close(code) reset code", () => { + // One connection per test carries three streams. Stream id -> the code given to close(code). + const codeOf: Record = { 1: ErrorCode.NO_ERROR, 3: ErrorCode.CANCEL, 5: ErrorCode.REFUSED_STREAM }; + const sids = [1, 3, 5]; + const theGivenCode = { "close(0)": [0], "close(8)": [8], "close(7)": [7] }; + + const discard = () => + new Writable({ + write(_chunk, _encoding, callback) { + callback(); + }, + }); + + // Attached before close(code). Each one destroys the stream, or calls close(3), from inside + // close(code): from 'aborted', or for the last one from the 'readable' that close(0) and close(8) + // emit when they end the readable of a stream that was read. + const readers: Record unknown> = { + "pipeline(stream, sink, cb)": stream => pipeline(stream, discard(), () => {}), + "promises.pipeline(stream, sink)": stream => promises.pipeline(stream, discard()).catch(() => {}), + "pipeline(stream, PassThrough, stream, cb)": stream => pipeline(stream, new PassThrough(), stream, () => {}), + "finished(stream, () => stream.destroy())": stream => finished(stream, () => stream.destroy()), + "on('aborted', () => stream.destroy(err))": stream => stream.on("aborted", () => stream.destroy(new Error("boom"))), + "on('aborted', () => stream.destroy())": stream => stream.on("aborted", () => stream.destroy()), + "on('aborted', () => stream.close(3))": stream => stream.on("aborted", () => stream.close(3)), + "read() and on('readable', () => stream.destroy(err))": stream => { + stream.on("readable", () => stream.destroy(new Error("boom"))); + stream.read(); + }, + }; + + const isHeaders = (sid: number) => (f: Frame) => f.type === FrameType.HEADERS && f.streamId === sid; + const isReset = (sid: number) => (f: Frame) => f.type === FrameType.RST_STREAM && f.streamId === sid; + const isPingAck = (opaque: number) => (f: Frame) => + f.type === FrameType.PING && (f.flags & 0x1) !== 0 && f.payload[0] === opaque; + // HEADERS with END_HEADERS and without END_STREAM. + const upload = (sid: number) => encodeFrame(FrameType.HEADERS, 0x4, sid, requestHeaderBlock("POST")); + + // ping(raw, 0) follows close(code) in the same tick, so the session reads a frame while its + // reset is in flight. + const ping = (raw: RawH2 | RawH2Server, opaque: number) => + raw.sendFrame(FrameType.PING, 0, 0, Buffer.alloc(8, opaque)); + + /** The distinct RST_STREAM codes on each of the three streams. */ + function resetCodes(frames: Frame[]) { + return Object.fromEntries( + sids.map(sid => [ + `close(${codeOf[sid]})`, + [...new Set(frames.filter(isReset(sid)).map(f => f.payload.readUInt32BE(0)))], + ]), + ); + } + + /** + * Two PING round trips: every frame the peer wrote before the second ACK has been parsed by + * then. The second round trip is for a frame that a deferred callback writes after the first ACK. + */ + async function twoPings(raw: RawH2 | RawH2Server) { + for (const round of [1, 2]) { + ping(raw, round); + await raw.waitFor(isPingAck(round)); + } + } + + /** + * Waits for the RST_STREAM of each stream in `ids`, then for two PING round trips. A reset that + * never arrives shows in the caller's assertion, not as a bare timeout. + */ + async function resetsThenTwoPings(raw: RawH2 | RawH2Server, ids = sids) { + await Promise.allSettled(ids.map(sid => raw.waitFor(isReset(sid)))); + await twoPings(raw); + } + + /** Runs `body` with a raw client that is connected to `listener`, after the SETTINGS handshake. */ + async function withRawClient(listener: net.Server | http2.Http2Server, body: (c: RawH2) => Promise) { + listener.listen(0); + await once(listener, "listening"); + const c = await RawH2.connect((listener.address() as net.AddressInfo).port); + try { + c.sendPreface(); + c.sendEmptySettings(); + await c.waitFor(f => f.type === FrameType.SETTINGS && (f.flags & 0x1) === 0); + c.sendSettingsAck(); + await body(c); + } finally { + c.destroy(); + listener.close(); + } + } + + /** Runs `body` with a client session that is connected to a raw server. */ + async function withRawServer(body: (client: http2.ClientHttp2Session, raw: RawH2Server) => Promise) { + const raw = await RawH2Server.listen(); + const client = http2.connect(`http://127.0.0.1:${raw.port}`); + client.on("error", () => {}); + try { + await body(client, raw); + } finally { + client.destroy(); + raw.close(); + } + } + + /** stream.rstCode of each stream at its 'close' event, in the shape of resetCodes(). */ + function rstCodesAtClose() { + const codes: Record = {}; + const closed: Promise[] = []; + return { + watch(stream: http2.Http2Stream) { + const key = `close(${codeOf[stream.id!]})`; + const { promise, resolve } = Promise.withResolvers(); + stream.on("close", () => { + codes[key] = [stream.rstCode]; + resolve(); + }); + closed.push(promise); + }, + async all() { + await Promise.all(closed); + return codes; + }, + }; + } + + /** An 'aborted' listener throws out of close(code). The close is complete, and a second close() changes nothing. */ + function closeUnderThrowingListener(stream: http2.Http2Stream) { + stream.once("aborted", () => { + throw new Error("boom"); + }); + let thrown: unknown; + try { + stream.close(codeOf[stream.id!]); + } catch (err) { + thrown = (err as Error).message; + } + stream.close(ErrorCode.ENHANCE_YOUR_CALM); + return { thrown, closed: stream.closed, writableEnded: stream.writableEnded }; + } + const closedInSpiteOfThrow = { thrown: "boom", closed: true, writableEnded: true }; + + const megabyte = Buffer.alloc(1024 * 1024, "a"); + + describe("server stream", () => { + /** + * Runs `handler` on each of the three uploads. Each reset on the wire and each rstCode at + * 'close' must be the code given to close(code). + */ + async function expectTheGivenCodes(handler: (stream: http2.ServerHttp2Stream) => void) { + const server = http2.createServer(); + await withRawClient(server, async c => { + const atClose = rstCodesAtClose(); + server.on("stream", (stream: http2.ServerHttp2Stream) => { + stream.on("error", () => {}); + atClose.watch(stream); + handler(stream); + ping(c, 0); + }); + c.send(Buffer.concat(sids.map(upload))); + await resetsThenTwoPings(c); + expect(resetCodes(c.frames)).toEqual(theGivenCode); + expect(await atClose.all()).toEqual(theGivenCode); + }); + } + + for (const [name, attach] of Object.entries(readers)) { + test(`close(code) under ${name}`, async () => { + await expectTheGivenCodes(stream => { + attach(stream); + stream.close(codeOf[stream.id!]); + }); + }); + } + + const states: Record void> = { + "after respond()": stream => stream.respond({ ":status": 200 }), + "with a write pending": stream => { + stream.respond({ ":status": 200 }); + stream.write("x"); + }, + "with the writable ended": stream => { + stream.respond({ ":status": 200 }); + stream.end("x"); + }, + }; + for (const [state, prepare] of Object.entries(states)) { + test(`close(code) under pipeline(stream, sink, cb) ${state}`, async () => { + await expectTheGivenCodes(stream => { + prepare(stream); + pipeline(stream, discard(), () => {}); + stream.close(codeOf[stream.id!]); + }); + }); + } + + // The session's transport is one side of a duplexPair(), not a socket. The other side is + // piped to a TCP socket so that the raw client can drive it. + test("close(7) under pipeline(stream, sink, cb) on an in-memory Duplex transport", async () => { + const server = http2.createServer(); + const relay = net.createServer(socket => { + const [near, far] = duplexPair(); + pipeline(socket, near, socket, () => {}); + server.emit("connection", far); + }); + await withRawClient(relay, async c => { + server.on("stream", stream => { + stream.on("error", () => {}); + pipeline(stream, discard(), () => {}); + stream.close(ErrorCode.REFUSED_STREAM); + ping(c, 0); + }); + c.send(upload(1)); + await resetsThenTwoPings(c, [1]); + expect([...new Set(c.frames.filter(isReset(1)).map(f => f.payload.readUInt32BE(0)))]).toEqual([7]); + }); + }); + + // When pipeline() destroyed the stream inside close(7), _destroy has queued the reset. A second + // call from close() would also count against maxSessionRejectedStreams. + test("close(7) under pipeline(stream, sink, cb) makes one rstStream call for each stream", async () => { + const server = http2.createServer(); + await withRawClient(server, async c => { + const calls: number[] = []; + let counted: object | undefined; + server.on("stream", stream => { + stream.on("error", () => {}); + const native = ( + stream.session as unknown as Record + )[Symbol.for("::bunhttp2native::")]; + if (counted !== native) { + counted = native; + const rstStream = native.rstStream; + native.rstStream = function (id, code) { + calls.push(id); + return rstStream.call(this, id, code); + }; + } + pipeline(stream, discard(), () => {}); + stream.close(ErrorCode.REFUSED_STREAM); + ping(c, 0); + }); + c.send(Buffer.concat(sids.map(upload))); + await resetsThenTwoPings(c); + expect(calls).toEqual(sids); + }); + }); + + test("close(code) when an 'aborted' listener throws: one reset with that code, then 'close'", async () => { + const afterThrow: unknown[] = []; + await expectTheGivenCodes(stream => { + afterThrow.push(closeUnderThrowingListener(stream)); + }); + expect(afterThrow).toEqual(sids.map(() => closedInSpiteOfThrow)); + }); + + // close(0) ends the readable first, and that emits 'readable' for a stream that was read. The + // listeners of 'aborted' still run after that listener threw. Stream id -> the events whose + // listener throws. + test("close(0) when listeners of 'readable' and 'aborted' throw: the close is complete, and it throws each error", async () => { + const throwing: Record = { 1: ["readable"], 3: ["aborted"], 5: ["readable", "aborted"] }; + const server = http2.createServer(); + await withRawClient(server, async c => { + const seen: Record = {}; + const closed: Promise[] = []; + server.on("stream", stream => { + const events: unknown[] = (seen[stream.id!] = []); + const { promise, resolve } = Promise.withResolvers(); + closed.push(promise); + stream.on("error", () => {}); + stream.on("close", () => { + events.push("close:" + stream.rstCode); + resolve(); + }); + stream.on("aborted", () => events.push("aborted")); + const thrownBy = new Map(); + // A close() that stops at the first error leaves a listener. It must not throw in a later test. + let insideClose = true; + for (const event of throwing[stream.id!]) { + const error = new Error(`${event} failed`); + thrownBy.set(error, event); + stream.once(event, () => { + if (insideClose) throw error; + }); + } + stream.read(); + try { + stream.close(ErrorCode.NO_ERROR); + } catch (err) { + const errors: unknown[] = err instanceof AggregateError ? err.errors : [err]; + events.push({ threw: errors.map(error => thrownBy.get(error)), message: (err as Error).message }); + } + insideClose = false; + ping(c, 0); + }); + c.send(Buffer.concat(sids.map(upload))); + await resetsThenTwoPings(c); + expect( + sids.map(sid => [...new Set(c.frames.filter(isReset(sid)).map(f => f.payload.readUInt32BE(0)))]), + ).toEqual(sids.map(() => [0])); + await Promise.all(closed); + expect(seen).toEqual({ + 1: ["aborted", { threw: ["readable"], message: "readable failed" }, "close:0"], + 3: ["aborted", { threw: ["aborted"], message: "aborted failed" }, "close:0"], + 5: ["aborted", { threw: ["readable", "aborted"], message: "readable failed" }, "close:0"], + }); + }); + }); + + // HEADERS and DATA with END_STREAM arrive in one read. close(2) in the handler has returned + // when that END_STREAM closes the stream, so the stream is destroyed at once, with no 'end'. + test("close(2) in the 'stream' handler with END_STREAM in the same read: no 'end'", async () => { + const server = http2.createServer(); + await withRawClient(server, async c => { + const events: string[] = []; + const { promise: closed, resolve } = Promise.withResolvers(); + server.on("stream", stream => { + stream.on("error", err => events.push("error:" + (err as NodeJS.ErrnoException).code)); + for (const name of ["aborted", "end"]) stream.on(name, () => events.push(name)); + stream.on("close", () => { + events.push("close:" + stream.rstCode); + resolve(); + }); + stream.close(ErrorCode.INTERNAL_ERROR); + }); + c.send(Buffer.concat([upload(1), encodeFrame(FrameType.DATA, 0x1, 1, Buffer.from("hello"))])); + await closed; + expect(events).toEqual(["aborted", "error:ERR_HTTP2_STREAM_ERROR", "close:2"]); + }); + }); + + // The raw client never opens its window, so most of the body stays queued. The reset that + // drops it carries the code of close(), like the reset that node's close(0) sends at once. + const dropQueuedBody: Record void> = { + "end(1 MiB) then close(0) then destroy(err)": stream => { + stream.end(megabyte); + stream.close(ErrorCode.NO_ERROR); + stream.destroy(new Error("boom")); + }, + "write(1 MiB) then close(0) with destroy(err) inside 'aborted'": stream => { + stream.write(megabyte); + stream.on("aborted", () => stream.destroy(new Error("boom"))); + stream.close(ErrorCode.NO_ERROR); + }, + }; + for (const [name, act] of Object.entries(dropQueuedBody)) { + test(`${name} sends RST_STREAM(0)`, async () => { + const server = http2.createServer(); + await withRawClient(server, async c => { + server.on("stream", (stream: http2.ServerHttp2Stream) => { + stream.on("error", () => {}); + stream.respond({ ":status": 200 }); + act(stream); + ping(c, 0); + }); + c.send(upload(1)); + await resetsThenTwoPings(c, [1]); + expect([...new Set(c.frames.filter(isReset(1)).map(f => f.payload.readUInt32BE(0)))]).toEqual([0]); + }); + }); + } + + // Today the stream writes an empty DATA frame with END_STREAM and no RST_STREAM. + test.todo("close(7) on a request that is already complete sends RST_STREAM(7) (#33380)"); + }); + + describe("client stream", () => { + /** Opens `count` POSTs and resolves once the raw server has the HEADERS of each. */ + async function openUploads( + client: http2.ClientHttp2Session, + raw: RawH2Server, + count: number, + options?: http2.ClientSessionRequestOptions, + ) { + const requests = Array.from({ length: count }, () => { + const req = client.request({ ":method": "POST", ":path": "/" }, options); + req.on("error", () => {}); + return req; + }); + await Promise.all(requests.map((_, i) => raw.waitFor(isHeaders(1 + 2 * i)))); + raw.sendFrame(FrameType.SETTINGS, 0, 0); + raw.sendFrame(FrameType.SETTINGS, 0x1, 0); + return requests; + } + + /** + * Runs `handler` on each of three uploads. Each reset on the wire and each rstCode at 'close' + * must be the code given to close(code). + */ + async function expectTheGivenCodes(handler: (req: http2.ClientHttp2Stream) => void) { + await withRawServer(async (client, raw) => { + const atClose = rstCodesAtClose(); + for (const req of await openUploads(client, raw, sids.length)) { + atClose.watch(req); + handler(req); + ping(raw, 0); + } + await resetsThenTwoPings(raw); + expect(resetCodes(raw.frames)).toEqual(theGivenCode); + expect(await atClose.all()).toEqual(theGivenCode); + }); + } + + for (const [name, attach] of Object.entries(readers)) { + test(`close(code) under ${name}`, async () => { + await expectTheGivenCodes(req => { + attach(req); + req.close(codeOf[req.id!]); + }); + }); + } + + const states: Record void> = { + "with a write pending": req => void req.write("x"), + "with the writable ended": req => void req.end("x"), + }; + for (const [state, prepare] of Object.entries(states)) { + test(`close(code) under pipeline(stream, sink, cb) ${state}`, async () => { + await expectTheGivenCodes(req => { + prepare(req); + pipeline(req, discard(), () => {}); + req.close(codeOf[req.id!]); + }); + }); + } + + test("close(code) when an 'aborted' listener throws: one reset with that code, then 'close'", async () => { + const afterThrow: unknown[] = []; + await expectTheGivenCodes(req => { + afterThrow.push(closeUnderThrowingListener(req)); + }); + expect(afterThrow).toEqual(sids.map(() => closedInSpiteOfThrow)); + }); + + // The 'close' of a request submits the next request that waits for a stream slot. The reset of + // close(0) is written at 'finish', so the peer gets it before the HEADERS of that request. + test("close(0) under on('aborted', () => stream.destroy()): the reset precedes a queued request", async () => { + await withRawServer(async (client, raw) => { + const remoteSettings = once(client, "remoteSettings"); + await raw.waitFor(f => f.type === FrameType.SETTINGS); + // SETTINGS_MAX_CONCURRENT_STREAMS = 1 + raw.sendFrame(FrameType.SETTINGS, 0, 0, Buffer.from([0, 3, 0, 0, 0, 1])); + raw.sendFrame(FrameType.SETTINGS, 0x1, 0); + await remoteSettings; + const first = client.request({ ":method": "POST", ":path": "/" }); + first.on("error", () => {}); + await raw.waitFor(isHeaders(1)); + const second = client.request({ ":path": "/" }); + second.on("error", () => {}); + const queued = second.pending; + first.on("aborted", () => first.destroy()); + first.close(ErrorCode.NO_ERROR); + await raw.waitFor(isHeaders(3)); + await twoPings(raw); + const order = raw.frames.filter(f => isReset(1)(f) || isHeaders(3)(f)).map(f => f.streamId); + expect({ queued, order }).toEqual({ queued: true, order: [1, 3] }); + }); + }); + + // Native rstStream(id, NGHTTP2_NO_ERROR) writes nothing for the id of a pushed stream, so + // destroy(err) after close(0) keeps its own code there. + test("pushed stream: close(0) then destroy(err) sends RST_STREAM(INTERNAL_ERROR)", async () => { + await withRawServer(async (client, raw) => { + const pushedStream = once(client, "stream"); + client.request({ ":path": "/" }).on("error", () => {}); + await raw.waitFor(isHeaders(1)); + raw.sendFrame(FrameType.SETTINGS, 0, 0); + raw.sendFrame(FrameType.SETTINGS, 0x1, 0); + const promised = Buffer.alloc(4); + promised.writeUInt32BE(2, 0); + raw.sendFrame(FrameType.PUSH_PROMISE, 0x4, 1, Buffer.concat([promised, requestHeaderBlock("GET")])); + const [pushed] = (await pushedStream) as [http2.ClientHttp2Stream]; + pushed.on("error", () => {}); + pushed.close(ErrorCode.NO_ERROR); + pushed.destroy(new Error("boom")); + await resetsThenTwoPings(raw, [2]); + expect([...new Set(raw.frames.filter(isReset(2)).map(f => f.payload.readUInt32BE(0)))]).toEqual([2]); + }); + }); + + // Both sessions run in this process on a duplexPair(). The server answers when it reads the + // END_STREAM of close(code), so its response is parsed before close(code) returns. + test("close(code) on an in-process transport: the response that arrives inside close() is delivered", async () => { + const [clientSide, serverSide] = duplexPair(); + const server = http2.createServer(); + server.on("stream", (stream: http2.ServerHttp2Stream) => { + stream.on("error", () => {}); + stream.on("readable", () => { + while (stream.read() !== null); + if (stream.headersSent || !(stream as unknown as { _readableState: { ended: boolean } })._readableState.ended) + return; + stream.respond({ ":status": 200 }); + stream.end("ok"); + }); + }); + server.emit("connection", serverSide); + const client = http2.connect("http://localhost", { createConnection: () => clientSide }); + client.on("error", () => {}); + try { + const seen: Record = {}; + for (const code of [ErrorCode.INTERNAL_ERROR, ErrorCode.REFUSED_STREAM]) { + const streamEvent = once(server, "stream"); + const req = client.request({ ":method": "POST", ":path": "/" }); + await streamEvent; + const events: string[] = (seen[code] = []); + const { promise: closed, resolve } = Promise.withResolvers(); + req.on("error", err => events.push("error:" + (err as NodeJS.ErrnoException).code)); + req.on("response", headers => events.push("response:" + headers[":status"])); + req.on("data", chunk => events.push("data:" + chunk)); + for (const name of ["aborted", "end"]) req.on(name, () => events.push(name)); + req.on("close", () => { + events.push("close:" + req.rstCode); + resolve(); + }); + req.close(code); + await closed; + } + const delivered = ["aborted", "response:200", "data:ok", "end", "error:ERR_HTTP2_STREAM_ERROR"]; + expect(seen).toEqual({ 2: [...delivered, "close:2"], 7: [...delivered, "close:7"] }); + } finally { + client.destroy(); + server.close(); + } + }); + + // When its socket closes, the session closes each open stream with close(NGHTTP2_CANCEL). That + // close is complete for stream "a" too, whose 'aborted' listener throws. + test("socket close with a throwing 'aborted' listener: the stream closes with CANCEL", async () => { + const fixture = ` + const http2 = require("node:http2"); + const net = require("node:net"); + const out = []; + process.on("uncaughtException", err => out.push("uncaught:" + err.message)); + const server = http2.createServer(); + server.on("sessionError", () => {}); + let socket; + let streams = 0; + server.on("stream", stream => { + stream.on("error", () => {}); + if (++streams === 2) socket.destroy(); + }); + server.listen(0, "127.0.0.1", () => { + const port = server.address().port; + const client = http2.connect("http://127.0.0.1:" + port, { + createConnection: () => (socket = net.connect(port, "127.0.0.1")), + }); + client.on("error", () => {}); + let left = 3; + const done = () => { + if (--left === 0) { + console.log(JSON.stringify(out.sort())); + server.close(); + } + }; + client.on("close", done); + for (const name of ["a", "b"]) { + const req = client.request({ ":method": "POST", ":path": "/" }); + req.on("error", () => {}); + req.on("close", () => { + out.push(name + "-close:" + req.rstCode); + done(); + }); + if (name === "a") { + req.on("aborted", () => { + throw new Error("boom"); + }); + } + } + }); + `; + 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]); + // stderr is in the diff for debugging. The fixture handles the exception, so it must not be there. + expect({ stdout, stderr, exitCode }).toEqual({ + stdout: JSON.stringify(["a-close:8", "b-close:8", "uncaught:boom"]) + "\n", + stderr: expect.not.stringContaining("boom"), + exitCode: 0, + }); + }, 30_000); + + // Today the native AbortSignal listener writes RST_STREAM(CANCEL) first, and 'close' shows 7. + test.todo("close(7) with the AbortSignal of the request aborted inside 'aborted' sends RST_STREAM(7) (#42763)"); + }); +}); 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 642de0da42de..56a77831deec 100644 --- a/test/js/node/http2/node-http2-client-close.test.ts +++ b/test/js/node/http2/node-http2-client-close.test.ts @@ -4,17 +4,24 @@ * 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. + * close(code) stores its code before it emits 'aborted'. A destroy() or close() made inside that + * event (stream.pipeline() destroys the stream there) does not change the code, and a listener that + * throws does not stop the close. * * Works with both: * bun bd test test/js/node/http2/node-http2-client-close.test.ts * node --test test/js/node/http2/node-http2-client-close.test.ts */ import assert from "node:assert"; +import dc from "node:diagnostics_channel"; +import { once } from "node:events"; import http2 from "node:http2"; import net from "node:net"; -import { describe, test } from "node:test"; +import { PassThrough, Writable, compose, finished, pipeline, promises } from "node:stream"; +import { after, before, describe, test } from "node:test"; -const { NGHTTP2_NO_ERROR, NGHTTP2_CANCEL, NGHTTP2_INTERNAL_ERROR, NGHTTP2_ENHANCE_YOUR_CALM } = http2.constants; +const { NGHTTP2_NO_ERROR, NGHTTP2_CANCEL, NGHTTP2_INTERNAL_ERROR, NGHTTP2_REFUSED_STREAM, NGHTTP2_ENHANCE_YOUR_CALM } = + http2.constants; // Raw h2c server: replies 200 + one DATA frame and never sends END_STREAM, so the only way the // stream ends is via the client's close(code). @@ -322,6 +329,391 @@ for (const [site, table] of [ } } +// close(code) stores its code before it hands control to a listener. What the listener then does +// to the stream does not change the code: stream.pipeline() and its relatives destroy the stream +// from inside 'aborted' when end-of-stream reports a premature close, and a second close(code) is +// a no-op. +const codes = [NGHTTP2_NO_ERROR, NGHTTP2_CANCEL, NGHTTP2_REFUSED_STREAM]; +const discard = () => + new Writable({ + write(chunk, encoding, callback) { + callback(); + }, + }); + +// Attached before close(code). Each one acts on the stream from inside 'aborted': destroy(), +// close(3) or end(). +const readers: [name: string, attach: (stream: http2.Http2Stream) => unknown][] = [ + ["pipeline(stream, sink, cb)", stream => pipeline(stream, discard(), () => {})], + ["promises.pipeline(stream, sink)", stream => promises.pipeline(stream, discard()).catch(() => {})], + ["compose(stream, PassThrough)", stream => compose(stream, new PassThrough()).on("error", () => {})], + ["pipeline(stream, PassThrough, stream, cb)", stream => pipeline(stream, new PassThrough(), stream, () => {})], + ["finished(stream, () => stream.destroy())", stream => finished(stream, () => stream.destroy())], + ["on('aborted', () => stream.destroy(err))", stream => stream.on("aborted", () => stream.destroy(new Error("boom")))], + ["on('aborted', () => stream.destroy())", stream => stream.on("aborted", () => stream.destroy())], + ["on('aborted', () => stream.close(3))", stream => stream.on("aborted", () => stream.close(3))], + ["on('aborted', () => stream.end())", stream => stream.on("aborted", () => stream.end())], +]; + +// The stream shows the code given to close(code) inside 'aborted', at 'close' and one turn later. +const codeIsKept = { + [NGHTTP2_NO_ERROR]: ["aborted:0", "close:0", "immediate:0"], + [NGHTTP2_CANCEL]: ["aborted:8", "close:8", "immediate:8"], + [NGHTTP2_REFUSED_STREAM]: ["aborted:7", "close:7", "immediate:7"], +}; + +// Records the 'error' and 'close' events of a stream and resolves at 'close'. +function errorAndClose(stream: http2.Http2Stream): Promise { + const events: string[] = []; + const { promise, resolve } = Promise.withResolvers(); + stream.on("error", err => events.push("error:" + ((err as NodeJS.ErrnoException).code ?? err.message))); + stream.on("close", () => { + events.push("close:" + stream.rstCode); + resolve(events); + }); + return promise; +} + +for (const side of ["server", "client"] as const) { + describe(`${side} stream keeps the code given to close(code)`, () => { + let server: http2.Http2Server; + let client: http2.ClientHttp2Session; + let sessionFailed: Promise; + before(async () => { + server = http2.createServer(); + server.on("stream", stream => stream.on("error", () => {})); + client = http2.connect(`http://127.0.0.1:${await listen(server)}`); + sessionFailed = new Promise((_, reject) => client.on("error", reject)); + sessionFailed.catch(() => {}); + }); + after(() => { + client.destroy(); + server.close(); + }); + + // Opens a POST whose body stays open, so both halves of the stream are open, and calls + // run(stream, peer) with this side's end first. The server side runs inside the server's + // 'stream' event. The client side runs once the server has the stream: node v26.3.0 spins + // when a request is destroyed inside 'aborted' in the same tick as request(). + type Run = (stream: http2.Http2Stream, peer: http2.Http2Stream) => Promise; + async function open(run: Run): Promise { + const req = client.request({ ":method": "POST", ":path": "/" }); + req.on("error", () => {}); + req.on("response", () => {}); + if (side === "client") { + const [stream] = await once(server, "stream"); + return run(req, stream); + } + const { promise, resolve } = Promise.withResolvers(); + server.once("stream", stream => resolve(run(stream, req))); + return promise; + } + const withStream = (run: Run) => Promise.race([open(run), sessionFailed]); + + // The peer sends 3000 bytes and ends its half. Resolves once this side has all of it: one ping + // round trip after the peer's 'finish'. This side has not read it. + async function peerEnds(stream: http2.Http2Stream, peer: http2.Http2Stream) { + if (side === "client") { + // A server stream that reads its request stays open after its response. + peer.resume(); + (peer as http2.ServerHttp2Stream).respond({ ":status": 200 }); + } + peer.end(Buffer.alloc(3000, "a")); + await once(peer, "finish"); + const { promise: pinged, resolve: onPing } = Promise.withResolvers(); + stream.session!.ping(onPing); + assert.strictEqual(await pinged, null); + } + + // For each code: attaches the reader, calls close(code) and records rstCode inside 'aborted' + // (after the reader's own 'aborted' listener), at 'close' and one setImmediate turn after + // 'close'. With peerEnded, the peer's half has ended first and its data is unread. + async function closeEachCode(attach: (stream: http2.Http2Stream) => unknown, peerEnded = false) { + const seen: Record = {}; + for (const code of codes) { + seen[code] = await withStream(async (stream, peer) => { + if (peerEnded) await peerEnds(stream, peer); + const events: string[] = []; + const { promise, resolve } = Promise.withResolvers(); + attach(stream); + // node never closes a client stream whose data nothing reads. + if (peerEnded && stream.readableFlowing === null) stream.resume(); + stream.on("aborted", () => events.push("aborted:" + stream.rstCode)); + stream.on("close", () => { + events.push("close:" + stream.rstCode); + setImmediate(() => { + events.push("immediate:" + stream.rstCode); + resolve(events); + }); + }); + stream.close(code); + return promise; + }); + } + return seen; + } + + for (const peerEnded of [false, true]) { + const state = peerEnded ? ", peer's half ended" : ""; + test(`close(code): 'aborted' sees rstCode${state}`, async () => { + assert.deepStrictEqual(await closeEachCode(() => {}, peerEnded), codeIsKept); + }); + + for (const [name, attach] of readers) { + test(`close(code) under ${name}${state}`, async () => { + assert.deepStrictEqual(await closeEachCode(attach, peerEnded), codeIsKept); + }); + } + } + + test("close(code) then destroy() inside 'aborted': 'error' for an error code only", async () => { + const seen: Record = {}; + for (const code of codes) { + seen[code] = await withStream(stream => { + const atClose = errorAndClose(stream); + stream.on("aborted", () => stream.destroy()); + stream.close(code); + return atClose; + }); + } + assert.deepStrictEqual(seen, { + [NGHTTP2_NO_ERROR]: ["close:0"], + [NGHTTP2_CANCEL]: ["close:8"], + [NGHTTP2_REFUSED_STREAM]: ["error:ERR_HTTP2_STREAM_ERROR", "close:7"], + }); + }); + + test("close(0) then destroy(err)", async () => { + const events = await withStream(stream => { + const atClose = errorAndClose(stream); + stream.close(NGHTTP2_NO_ERROR); + stream.destroy(new Error("boom")); + return atClose; + }); + assert.deepStrictEqual(events, ["error:boom", "close:0"]); + }); + + // With a body still queued, the reset that drops it carries the code of close() too, so the + // peer does not learn from the code that the body is cut short. + for (const [name, error] of [ + ["destroy(err)", () => new Error("boom")], + ["destroy(AbortError)", () => Object.assign(new Error("aborted"), { name: "AbortError", code: "ABORT_ERR" })], + ] as const) { + test(`write(1 MiB) then close() then ${name}: the peer's stream closes with code 0`, async () => { + const events = await withStream((stream, peer) => { + const atClose = errorAndClose(peer); + peer.resume(); + if (side === "server") (stream as http2.ServerHttp2Stream).respond({ ":status": 200 }); + stream.write(Buffer.alloc(1024 * 1024, "a")); + stream.close(); + stream.destroy(error()); + return atClose; + }); + assert.strictEqual(events.at(-1), "close:0"); + }); + } + + test("destroy(err) with close(0) inside its 'aborted'", async () => { + const events = await withStream(stream => { + const atClose = errorAndClose(stream); + stream.on("aborted", () => stream.close(NGHTTP2_NO_ERROR)); + stream.destroy(new Error("boom")); + return atClose; + }); + assert.deepStrictEqual(events, ["error:boom", "close:2"]); + }); + + // node v26.3.0 shows the same state and then never ends or resets the stream. node main + // (nodejs/node#66314) and bun close it with the first code. + test("close(code) when an 'aborted' listener throws: the stream is closed with that code", async () => { + const seen: Record = {}; + for (const code of codes) { + seen[code] = await withStream(async stream => { + stream.once("aborted", () => { + throw new Error("boom"); + }); + let thrown: unknown; + try { + stream.close(code); + } catch (err) { + thrown = (err as Error).message; + } + const state = [thrown, stream.closed, stream.rstCode]; + stream.close(NGHTTP2_ENHANCE_YOUR_CALM); + return [...state, stream.rstCode]; + }); + } + assert.deepStrictEqual(seen, { + [NGHTTP2_NO_ERROR]: ["boom", true, 0, 0], + [NGHTTP2_CANCEL]: ["boom", true, 8, 8], + [NGHTTP2_REFUSED_STREAM]: ["boom", true, 7, 7], + }); + }); + + // Inside 'aborted', node reports closed === true and ignores the call. bun reports false there, + // so it keeps the callback: code that closes a stream that is not closed yet still completes. + test("close() inside 'aborted' of close(7): bad arguments throw, a callback is not lost", async () => { + const seen = await withStream(stream => { + const { promise, resolve } = Promise.withResolvers>(); + let bad: string[] = []; + let closedInside = false; + let called = false; + stream.on("aborted", () => { + bad = [["x"], [-1], [0, "notfn"]].map(args => { + try { + Reflect.apply(stream.close, stream, args); + return "returned"; + } catch (err) { + return (err as NodeJS.ErrnoException).code!; + } + }); + closedInside = stream.closed; + stream.close(NGHTTP2_CANCEL, () => (called = true)); + }); + stream.on("close", () => + setImmediate(() => resolve({ bad, completes: closedInside || called, rstCode: stream.rstCode })), + ); + stream.close(NGHTTP2_REFUSED_STREAM); + return promise; + }); + assert.deepStrictEqual(seen, { + bad: ["ERR_INVALID_ARG_TYPE", "ERR_OUT_OF_RANGE", "ERR_INVALID_ARG_TYPE"], + completes: true, + rstCode: NGHTTP2_REFUSED_STREAM, + }); + }); + + // The peer's half has ended and its 3000 bytes are unread. close(code) ends the writable, and + // that closes the stream while close(code) is still running. + test("close(code) in the tick a consumer attaches: the consumer gets the data and 'end'", async () => { + const seen: Record = {}; + for (const code of [...codes, NGHTTP2_INTERNAL_ERROR]) { + seen[code] = await withStream(async (stream, peer) => { + await peerEnds(stream, peer); + const events: string[] = []; + const { promise, resolve } = Promise.withResolvers(); + let bytes = 0; + stream.on("data", chunk => (bytes += chunk.length)); + stream.on("aborted", () => events.push("aborted")); + stream.on("end", () => events.push("end:" + bytes)); + stream.on("error", err => events.push("error:" + (err as NodeJS.ErrnoException).code)); + stream.on("close", () => { + events.push("close:" + stream.rstCode); + resolve(events); + }); + stream.close(code); + return promise; + }); + } + assert.deepStrictEqual(seen, { + [NGHTTP2_NO_ERROR]: ["aborted", "end:3000", "close:0"], + [NGHTTP2_CANCEL]: ["aborted", "end:3000", "close:8"], + [NGHTTP2_REFUSED_STREAM]: ["aborted", "end:3000", "error:ERR_HTTP2_STREAM_ERROR", "close:7"], + [NGHTTP2_INTERNAL_ERROR]: ["aborted", "end:3000", "error:ERR_HTTP2_STREAM_ERROR", "close:2"], + }); + }); + + test("close(7) under pipeline(stream, sink, cb): diagnostics channel", async () => { + const channel = `http2.${side}.stream.close`; + const published: number[] = []; + let closing: http2.Http2Stream | undefined; + const onClose = (message: unknown) => { + const { stream } = message as { stream: http2.Http2Stream }; + if (stream === closing) published.push(stream.rstCode); + }; + dc.subscribe(channel, onClose); + try { + await withStream(stream => { + closing = stream; + const atClose = errorAndClose(stream); + pipeline(stream, discard(), () => {}); + stream.close(NGHTTP2_REFUSED_STREAM); + return atClose; + }); + } finally { + dc.unsubscribe(channel, onClose); + } + assert.deepStrictEqual(published, [NGHTTP2_REFUSED_STREAM]); + }); + }); +} + +// A listener of close(code) can run another closer of the stream: a session teardown, or the +// AbortSignal of the request. 'close' still shows the code given to close(code). +describe("close(code) keeps its code when a listener runs another closer", () => { + let server: http2.Http2Server; + let port: number; + before(async () => { + server = http2.createServer(); + server.on("sessionError", () => {}); + server.on("stream", stream => stream.on("error", () => {})); + port = await listen(server); + }); + after(() => { + server.close(); + }); + const connect = () => http2.connect(`http://127.0.0.1:${port}`).on("error", () => {}); + + const teardowns: [name: string, teardown: (session: http2.Http2Session) => void][] = [ + ["session.destroy()", session => session.destroy()], + ["session.destroy(err)", session => session.destroy(new Error("s"))], + ["session.destroy(err, 11)", session => session.destroy(new Error("s"), NGHTTP2_ENHANCE_YOUR_CALM)], + ]; + + for (const [name, teardown] of teardowns) { + test(`pending request: close(7) with ${name} inside 'aborted'`, async () => { + const client = connect(); + const req = client.request({ ":method": "POST", ":path": "/" }); + assert.strictEqual(req.pending, true); + req.once("aborted", () => teardown(client)); + const atClose = errorAndClose(req); + req.close(NGHTTP2_REFUSED_STREAM); + assert.strictEqual((await atClose).at(-1), "close:7"); + }); + } + + for (const code of [NGHTTP2_NO_ERROR, NGHTTP2_REFUSED_STREAM]) { + test(`close(${code}) with the AbortSignal of the request aborted inside 'aborted'`, async () => { + const client = connect(); + try { + const ac = new AbortController(); + const streamEvent = once(server, "stream"); + const req = client.request({ ":method": "POST", ":path": "/" }, { signal: ac.signal }); + await streamEvent; + req.once("aborted", () => ac.abort()); + const atClose = errorAndClose(req); + req.close(code); + assert.strictEqual((await atClose).at(-1), "close:" + code); + } finally { + client.destroy(); + } + }); + } + + for (const [name, teardown] of teardowns.slice(1)) { + for (const when of ["inside 'aborted'", "after close(0) returned"]) { + test(`server stream: close(0) with ${name} ${when}`, async () => { + const client = connect(); + try { + const { promise, resolve } = Promise.withResolvers(); + server.once("stream", stream => { + const session = stream.session!; + session.on("error", () => {}); + if (when === "inside 'aborted'") stream.once("aborted", () => teardown(session)); + resolve(errorAndClose(stream)); + stream.close(NGHTTP2_NO_ERROR); + if (when !== "inside 'aborted'") teardown(session); + }); + client.request({ ":method": "POST", ":path": "/" }).on("error", () => {}); + assert.strictEqual((await promise).at(-1), "close:0"); + } finally { + client.destroy(); + } + }); + } + } +}); + 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