diff --git a/src/bun.js/api/bun/h2_frame_parser.zig b/src/bun.js/api/bun/h2_frame_parser.zig index 26c0dd44c264..535db8e26fa9 100644 --- a/src/bun.js/api/bun/h2_frame_parser.zig +++ b/src/bun.js/api/bun/h2_frame_parser.zig @@ -1141,6 +1141,7 @@ pub const H2FrameParser = struct { this.signal = null; signal.deinit(); } + JSC.VirtualMachine.get().eventLoop().processGCTimer(); } }; @@ -1611,7 +1612,7 @@ pub const H2FrameParser = struct { // fallback to onWrite non-native callback const output_value = this.handlers.binary_type.toJS(bytes, this.handlers.globalObject); const result = this.call(.onWrite, output_value); - const code = result.to(i32); + const code = if (result.isNumber()) result.to(i32) else -1; switch (code) { -1 => { // dropped @@ -1757,7 +1758,7 @@ pub const H2FrameParser = struct { return data.len; } - pub fn decodeHeaderBlock(this: *H2FrameParser, payload: []const u8, stream: *Stream, flags: u8) *Stream { + pub fn decodeHeaderBlock(this: *H2FrameParser, payload: []const u8, stream: *Stream, flags: u8) ?*Stream { log("decodeHeaderBlock isSever: {}", .{this.isServer}); var offset: usize = 0; @@ -1776,7 +1777,9 @@ pub const H2FrameParser = struct { log("header {s} {s}", .{ header.name, header.value }); if (this.isServer and strings.eqlComptime(header.name, ":status")) { this.sendGoAway(stream_id, ErrorCode.PROTOCOL_ERROR, "Server received :status header", this.lastStreamID, true); - return this.streams.getEntry(stream_id).?.value_ptr; + + if (this.streams.getEntry(stream_id)) |entry| return entry.value_ptr; + return null; } count += 1; if (this.maxHeaderListPairs < count) { @@ -1786,7 +1789,8 @@ pub const H2FrameParser = struct { } else { this.endStream(stream, ErrorCode.ENHANCE_YOUR_CALM); } - return this.streams.getEntry(stream_id).?.value_ptr; + if (this.streams.getEntry(stream_id)) |entry| return entry.value_ptr; + return null; } const output = brk: { @@ -1817,7 +1821,8 @@ pub const H2FrameParser = struct { this.dispatchWith3Extra(.onStreamHeaders, stream.getIdentifier(), headers, sensitiveHeaders, JSC.JSValue.jsNumber(flags)); // callbacks can change the Stream ptr in this case we always return the new one - return this.streams.getEntry(stream_id).?.value_ptr; + if (this.streams.getEntry(stream_id)) |entry| return entry.value_ptr; + return null; } pub fn handleDataFrame(this: *H2FrameParser, frame: FrameHeader, data: []const u8, stream_: ?*Stream) usize { @@ -1882,7 +1887,8 @@ pub const H2FrameParser = struct { this.currentFrame = null; if (emitted) { // we need to revalidate the stream ptr after emitting onStreamData - stream = this.streams.getEntry(frame.streamIdentifier).?.value_ptr; + const entry = this.streams.getEntry(frame.streamIdentifier) orelse return end; + stream = entry.value_ptr; } if (frame.flags & @intFromEnum(DataFrameFlags.END_STREAM) != 0) { const identifier = stream.getIdentifier(); @@ -2029,7 +2035,10 @@ pub const H2FrameParser = struct { } if (handleIncommingPayload(this, data, frame.streamIdentifier)) |content| { const payload = content.data; - stream = this.decodeHeaderBlock(payload[0..payload.len], stream, frame.flags); + stream = this.decodeHeaderBlock(payload[0..payload.len], stream, frame.flags) orelse { + this.readBuffer.reset(); + return content.end; + }; this.readBuffer.reset(); if (frame.flags & @intFromEnum(HeadersFrameFlags.END_HEADERS) != 0) { stream.isWaitingMoreHeaders = false; @@ -2092,7 +2101,10 @@ pub const H2FrameParser = struct { this.sendGoAway(frame.streamIdentifier, ErrorCode.FRAME_SIZE_ERROR, "invalid Headers frame size", this.lastStreamID, true); return data.len; } - stream = this.decodeHeaderBlock(payload[offset..end], stream, frame.flags); + stream = this.decodeHeaderBlock(payload[offset..end], stream, frame.flags) orelse { + this.readBuffer.reset(); + return content.end; + }; this.readBuffer.reset(); stream.isWaitingMoreHeaders = frame.flags & @intFromEnum(HeadersFrameFlags.END_HEADERS) == 0; if (frame.flags & @intFromEnum(HeadersFrameFlags.END_STREAM) != 0) { @@ -3253,7 +3265,26 @@ pub const H2FrameParser = struct { } return array; } - + pub fn emitAbortToAllStreams(this: *H2FrameParser, _: *JSC.JSGlobalObject, _: *JSC.CallFrame) JSC.JSValue { + JSC.markBinding(@src()); + var it = StreamResumableIterator.init(this); + while (it.next()) |stream| { + // this is the oposite logic of emitErrorToallStreams, in this case we wanna to cancel this streams + if (this.isServer) { + if (stream.id % 2 == 0) continue; + } else if (stream.id % 2 != 0) continue; + if (stream.state != .CLOSED) { + const old_state = stream.state; + stream.state = .CLOSED; + stream.rstCode = @intFromEnum(ErrorCode.CANCEL); + const identifier = stream.getIdentifier(); + identifier.ensureStillAlive(); + stream.freeResources(this, false); + this.dispatchWith2Extra(.onAborted, identifier, .undefined, JSC.JSValue.jsNumber(@intFromEnum(old_state))); + } + } + return .undefined; + } pub fn emitErrorToAllStreams(this: *H2FrameParser, globalObject: *JSC.JSGlobalObject, callframe: *JSC.CallFrame) JSC.JSValue { JSC.markBinding(@src()); @@ -3265,6 +3296,9 @@ pub const H2FrameParser = struct { var it = StreamResumableIterator.init(this); while (it.next()) |stream| { + if (this.isServer) { + if (stream.id % 2 != 0) continue; + } else if (stream.id % 2 == 0) continue; if (stream.state != .CLOSED) { stream.state = .CLOSED; stream.rstCode = args_list.ptr[0].to(u32); @@ -3675,6 +3709,7 @@ pub const H2FrameParser = struct { } const socket_js = args_list.ptr[0]; + this.detachNativeSocket(); if (JSTLSSocket.fromJS(socket_js)) |socket| { log("TLSSocket attached", .{}); if (socket.attachNativeCallback(.{ .h2 = this })) { @@ -3859,17 +3894,15 @@ pub const H2FrameParser = struct { } return this; } - - pub fn deinit(this: *H2FrameParser) void { - log("deinit", .{}); - - defer { - if (ENABLE_ALLOCATOR_POOL) { - H2FrameParser.pool.?.put(this); - } else { - this.destroy(); - } - } + pub fn detachFromJS(this: *H2FrameParser, _: *JSC.JSGlobalObject, _: *JSC.CallFrame) JSValue { + JSC.markBinding(@src()); + this.detach(false); + return .undefined; + } + /// be careful when calling detach be sure that the socket is closed and the parser not accesible anymore + /// this function can be called multiple times, it will erase stream info + pub fn detach(this: *H2FrameParser, comptime finalizing: bool) void { + this.flushCorked(); this.detachNativeSocket(); this.strong_ctx.deinit(); this.handlers.deinit(); @@ -3886,9 +3919,24 @@ pub const H2FrameParser = struct { } var it = this.streams.valueIterator(); while (it.next()) |stream| { - stream.freeResources(this, true); + stream.freeResources(this, finalizing); + } + var streams = this.streams; + defer streams.deinit(); + this.streams = bun.U32HashMap(Stream).init(bun.default_allocator); + } + + pub fn deinit(this: *H2FrameParser) void { + log("deinit", .{}); + + defer { + if (ENABLE_ALLOCATOR_POOL) { + H2FrameParser.pool.?.put(this); + } else { + this.destroy(); + } } - this.streams.deinit(); + this.detach(true); } pub fn finalize( diff --git a/src/bun.js/api/h2.classes.ts b/src/bun.js/api/h2.classes.ts index dab1dd2d5ba5..bcad57f64a45 100644 --- a/src/bun.js/api/h2.classes.ts +++ b/src/bun.js/api/h2.classes.ts @@ -37,6 +37,10 @@ export default [ fn: "flushFromJS", length: 0, }, + detach: { + fn: "detachFromJS", + length: 0, + }, rstStream: { fn: "rstStream", length: 1, @@ -93,6 +97,10 @@ export default [ fn: "emitErrorToAllStreams", length: 1, }, + emitAbortToAllStreams: { + fn: "emitAbortToAllStreams", + length: 0, + }, getNextStream: { fn: "getNextStream", length: 0, diff --git a/src/js/node/http2.ts b/src/js/node/http2.ts index 4840bf4d834c..72936d97851a 100644 --- a/src/js/node/http2.ts +++ b/src/js/node/http2.ts @@ -1541,6 +1541,7 @@ function markStreamClosed(stream: Http2Stream) { if ((status & StreamState.Closed) === 0) { stream[bunHTTP2StreamStatus] = status | StreamState.Closed; + markWritableDone(stream); } } @@ -1709,7 +1710,6 @@ class Http2Stream extends Duplex { markStreamClosed(this); session[bunHTTP2Native]?.rstStream(this.#id, code); - this[bunHTTP2Session] = null; } if (typeof callback === "function") { @@ -1717,50 +1717,46 @@ class Http2Stream extends Duplex { } } _destroy(err, callback) { - if ((this[bunHTTP2StreamStatus] & StreamState.Closed) === 0) { - const { ending } = this._writableState; - if (!ending) { - // If the writable side of the Http2Stream is still open, emit the - // 'aborted' event and set the aborted flag. - if (!this.aborted) { - this[kAborted] = true; - this.emit("aborted"); - } - - // at this state destroyed will be true but we need to close the writable side - this._writableState.destroyed = false; - this.end(); - // we now restore the destroyed flag - this._writableState.destroyed = true; + const { ending } = this._writableState; + + if (!ending) { + // If the writable side of the Http2Stream is still open, emit the + // 'aborted' event and set the aborted flag. + if (!this.aborted) { + this[kAborted] = true; + this.emit("aborted"); } + // at this state destroyed will be true but we need to close the writable side + this._writableState.destroyed = false; + this.end(); + // we now restore the destroyed flag + this._writableState.destroyed = true; + } - const session = this[bunHTTP2Session]; - assertSession(session); + const session = this[bunHTTP2Session]; + assertSession(session); - let rstCode = this.rstCode; - if (!rstCode) { - if (err != null) { - if (err.code === "ABORT_ERR") { - // Enables using AbortController to cancel requests with RST code 8. - rstCode = NGHTTP2_CANCEL; - } else { - rstCode = NGHTTP2_INTERNAL_ERROR; - } + let rstCode = this.rstCode; + if (!rstCode) { + if (err != null) { + if (err.code === "ABORT_ERR") { + // Enables using AbortController to cancel requests with RST code 8. + rstCode = NGHTTP2_CANCEL; } else { - rstCode = this.rstCode = 0; + rstCode = NGHTTP2_INTERNAL_ERROR; } + } else { + rstCode = this.rstCode = 0; } + } - if (this.writableFinished) { - markStreamClosed(this); + if (this.writableFinished) { + markStreamClosed(this); - session[bunHTTP2Native]?.rstStream(this.#id, rstCode); - this[bunHTTP2Session] = null; - } else { - this.once("finish", Http2Stream.#rstStream); - } - } else { + session[bunHTTP2Native]?.rstStream(this.#id, rstCode); this[bunHTTP2Session] = null; + } else { + this.once("finish", Http2Stream.#rstStream); } callback(err); @@ -2154,6 +2150,7 @@ function emitStreamErrorNT(self, stream, error, destroy, destroy_self) { } else { error_instance = error; } + if (stream.readable) { stream.resume(); // we have a error we consume and close pushToStream(stream, null); @@ -2163,6 +2160,7 @@ function emitStreamErrorNT(self, stream, error, destroy, destroy_self) { else if (error_instance) { stream.emit("error", error_instance); } + if (destroy_self) self.destroy(); } } @@ -2247,15 +2245,12 @@ class ServerHttp2Session extends Http2Session { }, aborted(self: ServerHttp2Session, stream: ServerHttp2Stream, error: any, old_state: number) { if (!self || typeof stream !== "object") return; - stream.rstCode = constants.NGHTTP2_CANCEL; - markStreamClosed(stream); // if writable and not closed emit aborted if (old_state != 5 && old_state != 7) { stream[kAborted] = true; stream.emit("aborted"); } - self.#connections--; process.nextTick(emitStreamErrorNT, self, stream, error, true, self.#connections === 0 && self.#closed); }, @@ -2343,9 +2338,7 @@ class ServerHttp2Session extends Http2Session { error(self: ServerHttp2Session, errorCode: number, lastStreamId: number, opaqueData: Buffer) { if (!self) return; const error_instance = sessionErrorFromCode(errorCode); - self.emit("error", error_instance); - self[bunHTTP2Socket]?.end(); - self.#parser = null; + self.destroy(error_instance); }, wantTrailers(self: ServerHttp2Session, stream: ServerHttp2Stream) { if (!self || typeof stream !== "object") return; @@ -2366,14 +2359,11 @@ class ServerHttp2Session extends Http2Session { if (errorCode !== 0) { self.#parser.emitErrorToAllStreams(errorCode); } - - self[bunHTTP2Socket]?.end(); - self.#parser = null; + self.close(); }, end(self: ServerHttp2Session, errorCode: number, lastStreamId: number, opaqueData: Buffer) { if (!self) return; - self[bunHTTP2Socket]?.end(); - self.#parser = null; + self.destroy(); }, write(self: ServerHttp2Session, buffer: Buffer) { if (!self) return -1; @@ -2391,7 +2381,12 @@ class ServerHttp2Session extends Http2Session { } #onClose() { - // this.destroy(); + const parser = this.#parser; + if (parser) { + parser.emitAbortToAllStreams(); + parser.detach(); + this.#parser = null; + } this.close(); } @@ -2602,8 +2597,12 @@ class ServerHttp2Session extends Http2Session { this.goaway(code || constants.NGHTTP2_NO_ERROR, 0, Buffer.alloc(0)); socket.end(); } - this.#parser?.emitErrorToAllStreams(code || constants.NGHTTP2_NO_ERROR); - this.#parser = null; + const parser = this.#parser; + if (parser) { + parser.emitErrorToAllStreams(code || constants.NGHTTP2_NO_ERROR); + parser.detach(); + this.#parser = null; + } this[bunHTTP2Socket] = null; if (error) { @@ -2653,8 +2652,6 @@ class ClientHttp2Session extends Http2Session { }, aborted(self: ClientHttp2Session, stream: ClientHttp2Stream, error: any, old_state: number) { if (!self || typeof stream !== "object") return; - - markStreamClosed(stream); stream.rstCode = constants.NGHTTP2_CANCEL; // if writable and not closed emit aborted if (old_state != 5 && old_state != 7) { @@ -2754,9 +2751,7 @@ class ClientHttp2Session extends Http2Session { error(self: ClientHttp2Session, errorCode: number, lastStreamId: number, opaqueData: Buffer) { if (!self) return; const error_instance = sessionErrorFromCode(errorCode); - self.emit("error", error_instance); - self[bunHTTP2Socket]?.destroy(); - self.#parser = null; + self.destroy(error_instance); }, wantTrailers(self: ClientHttp2Session, stream: ClientHttp2Stream) { @@ -2776,13 +2771,11 @@ class ClientHttp2Session extends Http2Session { if (errorCode !== 0) { self.#parser.emitErrorToAllStreams(errorCode); } - self[bunHTTP2Socket]?.end(); - self.#parser = null; + self.close(); }, end(self: ClientHttp2Session, errorCode: number, lastStreamId: number, opaqueData: Buffer) { if (!self) return; - self[bunHTTP2Socket]?.end(); - self.#parser = null; + self.destroy(); }, write(self: ClientHttp2Session, buffer: Buffer) { if (!self) return -1; @@ -2836,9 +2829,17 @@ class ClientHttp2Session extends Http2Session { } #onClose() { + const parser = this.#parser; + if (parser) { + parser.emitAbortToAllStreams(); + parser.detach(); + this.#parser = null; + } this.close(); + this[bunHTTP2Socket] = null; } #onError(error: Error) { + this[bunHTTP2Socket] = null; this.destroy(error); } #onTimeout() { @@ -3055,9 +3056,13 @@ class ClientHttp2Session extends Http2Session { this.goaway(code || constants.NGHTTP2_NO_ERROR, 0, Buffer.alloc(0)); socket.end(); } - this.#parser?.emitErrorToAllStreams(code || constants.NGHTTP2_NO_ERROR); - this[bunHTTP2Socket] = null; + const parser = this.#parser; + if (parser) { + parser.emitErrorToAllStreams(code || constants.NGHTTP2_NO_ERROR); + parser.detach(); + } this.#parser = null; + this[bunHTTP2Socket] = null; if (error) { this.emit("error", error); diff --git a/test/js/node/http2/node-http2.test.js b/test/js/node/http2/node-http2.test.js index c75a0f5cb0cb..6d19fe6dd1e5 100644 --- a/test/js/node/http2/node-http2.test.js +++ b/test/js/node/http2/node-http2.test.js @@ -10,7 +10,7 @@ import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it } from import http2utils from "./helpers"; import { nodeEchoServer, TLS_CERT, TLS_OPTIONS } from "./http2-helpers"; -for (const nodeExecutable of [nodeExe()]) { +for (const nodeExecutable of [nodeExe(), bunExe()]) { describe(`${path.basename(nodeExecutable)}`, () => { let nodeEchoServer_; @@ -665,30 +665,7 @@ for (const nodeExecutable of [nodeExe()]) { expect(req.aborted).toBeTrue(); expect(req.rstCode).toBe(http2.constants.NGHTTP2_CANCEL); }); - it("aborted event should not work when not writable but should emit error", async () => { - const abortController = new AbortController(); - const { promise, resolve, reject } = Promise.withResolvers(); - const client = http2.connect(HTTPS_SERVER, TLS_OPTIONS); - client.on("error", reject); - const req = client.request({ ":path": "/" }, { signal: abortController.signal }); - req.on("aborted", reject); - req.on("error", err => { - if (err.code !== "ABORT_ERR") { - reject(err); - } else { - resolve(); - } - }); - req.on("end", () => { - reject(); - client.close(); - }); - abortController.abort(); - const result = await promise; - expect(result).toBeUndefined(); - expect(req.aborted).toBeFalse(); // will only be true when the request is in a writable state - expect(req.rstCode).toBe(http2.constants.NGHTTP2_CANCEL); - }); + it("aborted event should work with aborted signal", async () => { const { promise, resolve, reject } = Promise.withResolvers(); const client = http2.connect(HTTPS_SERVER, TLS_OPTIONS); diff --git a/test/js/node/test/parallel/http2-connect-tls-with-delay.test.js b/test/js/node/test/parallel/http2-connect-tls-with-delay.test.js index 8e70ca287039..1161272cabe0 100644 --- a/test/js/node/test/parallel/http2-connect-tls-with-delay.test.js +++ b/test/js/node/test/parallel/http2-connect-tls-with-delay.test.js @@ -1,54 +1,54 @@ //#FILE: test-http2-connect-tls-with-delay.js //#SHA1: 8c5489e025ec14c2cc53788b27fde11a11990e42 //----------------- -'use strict'; +"use strict"; -const http2 = require('http2'); -const tls = require('tls'); -const fs = require('fs'); -const path = require('path'); +const http2 = require("http2"); +const tls = require("tls"); +const fs = require("fs"); +const path = require("path"); const serverOptions = { - key: fs.readFileSync(path.join(__dirname, '..', 'fixtures', 'keys', 'agent1-key.pem')), - cert: fs.readFileSync(path.join(__dirname, '..', 'fixtures', 'keys', 'agent1-cert.pem')) + key: fs.readFileSync(path.join(__dirname, "..", "fixtures", "keys", "agent1-key.pem")), + cert: fs.readFileSync(path.join(__dirname, "..", "fixtures", "keys", "agent1-cert.pem")), }; let server; -beforeAll((done) => { +beforeAll(done => { server = http2.createSecureServer(serverOptions, (req, res) => { res.end(); }); - server.listen(0, '127.0.0.1', done); + server.listen(0, "127.0.0.1", done); }); -afterAll((done) => { - server.close(done); +afterAll(() => { + server.close(); }); -test('HTTP/2 connect with TLS and delay', (done) => { +test("HTTP/2 connect with TLS and delay", done => { const options = { - ALPNProtocols: ['h2'], - host: '127.0.0.1', - servername: 'localhost', + ALPNProtocols: ["h2"], + host: "127.0.0.1", + servername: "localhost", port: server.address().port, - rejectUnauthorized: false + rejectUnauthorized: false, }; const socket = tls.connect(options, async () => { - socket.once('readable', () => { - const client = http2.connect( - 'https://localhost:' + server.address().port, - { ...options, createConnection: () => socket } - ); + socket.once("readable", () => { + const client = http2.connect("https://localhost:" + server.address().port, { + ...options, + createConnection: () => socket, + }); - client.once('remoteSettings', () => { + client.once("remoteSettings", () => { const req = client.request({ - ':path': '/' + ":path": "/", }); - req.on('data', () => req.resume()); - req.on('end', () => { + req.on("data", () => req.resume()); + req.on("end", () => { client.close(); req.close(); done();