diff --git a/src/js/node/http2.ts b/src/js/node/http2.ts index 1ec56d7470bc..c1ad14207fce 100644 --- a/src/js/node/http2.ts +++ b/src/js/node/http2.ts @@ -372,6 +372,7 @@ const kSessionDestroyError = Symbol("sessionDestroyError"); // responsible for closing it exactly once. respondWithFD() descriptors belong to the caller. const kOwnsFd = Symbol("ownsFd"); const kRequestHeaders = Symbol("requestHeaders"); +const kEndAfterHeaders = Symbol("endAfterHeaders"); // True only for the synchronous extent of Http2Stream#sendTrailers()'s native call. node submits // trailers on a later turn (finishSendTrailers via setImmediate), so a stream never observes its // own close from inside sendTrailers() — the streamEnd(closed) dispatch is re-queued when this is @@ -1694,6 +1695,7 @@ const constants = { const { NGHTTP2_SESSION_SERVER, NGHTTP2_SESSION_CLIENT, + NGHTTP2_FLAG_END_STREAM, NGHTTP2_NO_ERROR, NGHTTP2_INTERNAL_ERROR, NGHTTP2_CANCEL, @@ -2296,7 +2298,10 @@ class Http2Stream extends Duplex { // async-context tracking when no AsyncLocalStorage is in use. [bunHTTP2AsyncContextFrame] = $getInternalField($asyncContext, 0); - rstCode: number | undefined = undefined; + rstCode: number = 0; + // END_STREAM flag of the header block that opened the stream, as in node's onSessionHeaders. + [kEndAfterHeaders]: boolean = false; + headersSent: boolean = false; [bunHTTP2Headers]: any; [kInfoHeaders]: any; #sentTrailers: any; @@ -2495,11 +2500,7 @@ class Http2Stream extends Duplex { } get endAfterHeaders() { - const session = this[bunHTTP2Session]; - if (session) { - return session[bunHTTP2Native]?.getEndAfterHeaders(this.#id) || false; - } - return false; + return this[kEndAfterHeaders]; } get aborted() { @@ -2941,7 +2942,10 @@ class Http2Stream extends Duplex { } } } -class ClientHttp2Stream extends Http2Stream {} +class ClientHttp2Stream extends Http2Stream { + // node sets STREAM_FLAGS_HEADERS_SENT in its ClientHttp2Stream constructor. + headersSent: boolean = true; +} // Wrap a native→JS #Handlers callback so its body runs inside the target // stream's captured async-context frame — the JS-side equivalent of Node's @@ -3183,7 +3187,6 @@ function callStreamClose(stream: ServerHttp2Stream) { if (!stream.destroyed && !stream.closed) stream.close(); } class ServerHttp2Stream extends Http2Stream { - headersSent = false; constructor(streamId, session, headers) { super(streamId, session, headers); } @@ -4164,6 +4167,7 @@ class ServerHttp2Session extends Http2Session { // user handler — in particular, losing WantTrailer/FinalCalled breaks // any later `sendTrailers()` with ERR_HTTP2_TRAILERS_NOT_READY. stream[bunHTTP2StreamStatus] |= StreamState.StreamResponded | StreamState.Delivered; + stream[kEndAfterHeaders] = (flags & NGHTTP2_FLAG_END_STREAM) !== 0; if (onServerStreamCreatedChannel.hasSubscribers) { onServerStreamCreatedChannel.publish({ stream, headers }); } @@ -5660,9 +5664,11 @@ class ClientHttp2Session extends Http2Session { process.nextTick(onConnect.bind(this)); return; } + // Captured before #onConnect(), which drops the socket if close() already ran. + const socket = this[bunHTTP2Socket]; try { this.#onConnect(arguments); - listener?.$call(this, this); + listener?.$call(this, this, socket); } catch (e) { this.destroy(e); } diff --git a/src/runtime/api/bun/h2_frame_parser.rs b/src/runtime/api/bun/h2_frame_parser.rs index 06c2b66811a3..770fadd19df8 100644 --- a/src/runtime/api/bun/h2_frame_parser.rs +++ b/src/runtime/api/bun/h2_frame_parser.rs @@ -1308,7 +1308,6 @@ pub struct Stream { state: StreamState, js_context: StrongOptional, // jsc.Strong.Optional wait_for_trailers: bool, - end_after_headers: bool, padding_strategy: PaddingStrategy, rst_code: u32, stream_dependency: u32, @@ -1823,7 +1822,6 @@ impl Stream { state: StreamState::OPEN, js_context: StrongOptional::empty(), wait_for_trailers: false, - end_after_headers: false, padding_strategy, rst_code: 0, stream_dependency: 0, @@ -4021,12 +4019,7 @@ impl crate::api::h2::connection::Sink for H2FrameParser { }); } - fn on_headers_complete(&self, stream_id: u32, end_stream: bool, flags: u8) { - // Bridge: the JS endAfterHeaders getter reads the legacy stream's end_after_headers flag. - if let Some(stream) = self.streams.get().get(&stream_id).copied() { - // SAFETY: stream is *mut Stream from self.streams; valid while the map entry exists - unsafe { (*stream).end_after_headers = end_stream }; - } + fn on_headers_complete(&self, stream_id: u32, _end_stream: bool, flags: u8) { // Materialize the accumulated block in a single native pass: the raw array, the // node-shaped headers object, and the sensitive list (a zero-field block yields an // empty array + object, matching the legacy decoder's upfront array). @@ -4890,34 +4883,6 @@ impl H2FrameParser { Ok(JSValue::UNDEFINED) } - #[bun_jsc::host_fn(method)] - pub(crate) fn get_end_after_headers( - this: &Self, - global_object: &JSGlobalObject, - callframe: &CallFrame, - ) -> JsResult { - let [stream_arg] = callframe.arguments_as_array::<1>(); - if callframe.arguments_count() < 1 { - return Err(global_object.throw(format_args!("Expected stream argument"))); - } - - if !stream_arg.is_number() { - return Err(global_object.throw(format_args!("Invalid stream id"))); - } - - let stream_id = stream_arg.to_u32(); - if stream_id == 0 { - return Err(global_object.throw(format_args!("Invalid stream id"))); - } - - let Some(stream) = this.streams.get().get(&stream_id).copied() else { - return Err(global_object.throw(format_args!("Invalid stream id"))); - }; - - // SAFETY: stream is *mut Stream from self.streams; valid while the map entry exists - Ok(JSValue::from(unsafe { (*stream).end_after_headers })) - } - #[bun_jsc::host_fn(method)] pub(crate) fn is_stream_aborted( this: &Self, @@ -7387,8 +7352,6 @@ impl H2FrameParser { } if end_stream { - stream.end_after_headers = true; - if wait_for_trailers { stream.state = StreamState::HALF_CLOSED_LOCAL; this.dispatch(JSH2FrameParser::Gc::onWantTrailers, stream.get_identifier()); diff --git a/src/runtime/api/h2.classes.ts b/src/runtime/api/h2.classes.ts index 8b4749d6fb05..b06741ac4298 100644 --- a/src/runtime/api/h2.classes.ts +++ b/src/runtime/api/h2.classes.ts @@ -89,10 +89,6 @@ export default [ fn: "setStreamContext", length: 2, }, - getEndAfterHeaders: { - fn: "getEndAfterHeaders", - length: 1, - }, isStreamAborted: { fn: "isStreamAborted", length: 1, diff --git a/test/js/node/http2/node-http2.test.js b/test/js/node/http2/node-http2.test.js index f73bebcc9863..cd386ff42adf 100644 --- a/test/js/node/http2/node-http2.test.js +++ b/test/js/node/http2/node-http2.test.js @@ -6009,3 +6009,148 @@ describe("frames issued from inside a user-supplied Duplex transport's _write", }, ); }); + +// Getters must be total over a stream's lifecycle: node answers the same way before the stream +// has an id, while it is open, and after it closed. Bun used to throw or report undefined. +describe("stream getters over the lifecycle", () => { + const read = stream => ({ + endAfterHeaders: stream.endAfterHeaders, + rstCode: stream.rstCode, + headersSent: stream.headersSent, + }); + + it("client: a pending stream answers like node, and the values survive close", async () => { + const server = http2.createServer((req, res) => res.end("ok")); + await new Promise(resolve => server.listen(0, "127.0.0.1", resolve)); + try { + const client = http2.connect(`http://127.0.0.1:${server.address().port}`); + try { + // Issued before 'connect', so the stream has no id yet. + const req = client.request({ ":path": "/" }); + const pending = read(req); + const { promise, resolve, reject } = Promise.withResolvers(); + let open; + req.on("response", () => (open = read(req))); + req.on("error", reject); + req.resume(); + req.on("close", resolve); + await promise; + expect(pending).toEqual({ endAfterHeaders: false, rstCode: 0, headersSent: true }); + expect(open).toEqual({ endAfterHeaders: false, rstCode: 0, headersSent: true }); + expect(read(req)).toEqual({ endAfterHeaders: false, rstCode: 0, headersSent: true }); + } finally { + client.destroy(); + } + } finally { + server.close(); + } + }); + + it("server: endAfterHeaders stays true after the stream closes", async () => { + const server = http2.createServer(); + const states = []; + const serverStreamClosed = Promise.withResolvers(); + server.on("stream", stream => { + states.push(read(stream)); + stream.on("close", () => { + states.push(read(stream)); + serverStreamClosed.resolve(); + }); + stream.respond({ ":status": 200 }); + stream.end("ok"); + }); + await new Promise(resolve => server.listen(0, "127.0.0.1", resolve)); + try { + const client = http2.connect(`http://127.0.0.1:${server.address().port}`); + try { + const req = client.request({ ":path": "/" }); + const { promise, resolve, reject } = Promise.withResolvers(); + req.on("error", reject); + req.resume(); + req.on("close", resolve); + await Promise.all([promise, serverStreamClosed.promise]); + // The GET request block carried END_STREAM, so endAfterHeaders is true for its whole life. + expect(states[0]).toEqual({ endAfterHeaders: true, rstCode: 0, headersSent: false }); + expect(states[1]).toEqual({ endAfterHeaders: true, rstCode: 0, headersSent: true }); + } finally { + client.destroy(); + } + } finally { + server.close(); + } + }); + + // node's onSessionHeaders sets endAfterHeaders only from the header block that creates the + // stream: a later trailers block (END_STREAM) does not flip it on the server, and a client + // request stream never gets it, even when the response HEADERS frame carries END_STREAM. + it("endAfterHeaders comes only from the block that opened the stream", async () => { + const server = http2.createServer(); + const serverSide = {}; + const serverStreamClosed = Promise.withResolvers(); + server.on("stream", stream => { + serverSide.atStream = stream.endAfterHeaders; + stream.on("trailers", () => (serverSide.atTrailers = stream.endAfterHeaders)); + stream.on("end", () => { + serverSide.atEnd = stream.endAfterHeaders; + stream.respond({ ":status": 204 }, { endStream: true }); + }); + stream.on("close", () => { + serverSide.atClose = stream.endAfterHeaders; + serverStreamClosed.resolve(); + }); + stream.resume(); + }); + await new Promise(resolve => server.listen(0, "127.0.0.1", resolve)); + try { + const client = http2.connect(`http://127.0.0.1:${server.address().port}`); + try { + const req = client.request({ ":method": "POST", ":path": "/" }, { waitForTrailers: true }); + const clientSide = {}; + const { promise, resolve, reject } = Promise.withResolvers(); + req.on("wantTrailers", () => req.sendTrailers({ "x-t": "1" })); + req.on("response", (headers, flags) => { + clientSide.atResponse = req.endAfterHeaders; + clientSide.responseEndStream = (flags & http2.constants.NGHTTP2_FLAG_END_STREAM) !== 0; + }); + req.on("error", reject); + req.on("close", () => { + clientSide.atClose = req.endAfterHeaders; + resolve(); + }); + req.resume(); + req.end("body"); + await Promise.all([promise, serverStreamClosed.promise]); + expect(serverSide).toEqual({ atStream: false, atTrailers: false, atEnd: false, atClose: false }); + expect(clientSide).toEqual({ atResponse: false, responseEndStream: true, atClose: false }); + } finally { + client.destroy(); + } + } finally { + server.close(); + } + }); + + it("the http2.connect() listener receives the session and the socket", async () => { + const server = http2.createServer(); + await new Promise(resolve => server.listen(0, "127.0.0.1", resolve)); + try { + const { promise, resolve, reject } = Promise.withResolvers(); + const client = http2.connect(`http://127.0.0.1:${server.address().port}`, {}, (session, socket) => + resolve({ sameSession: session === client, socket }), + ); + client.on("error", reject); + client.on("close", () => reject(new Error("session closed before the connect listener ran"))); + try { + const args = await promise; + expect(args.sameSession).toBe(true); + // node hands over the raw socket, not the session's socket proxy. + expect(args.socket).toBeInstanceOf(net.Socket); + expect(args.socket.remotePort).toBe(server.address().port); + } finally { + client.destroy(); + } + } finally { + server.close(); + } + }); +});