diff --git a/src/runtime/api/bun/h2_frame_parser.rs b/src/runtime/api/bun/h2_frame_parser.rs index 129dfc02496e..901de23f50ad 100644 --- a/src/runtime/api/bun/h2_frame_parser.rs +++ b/src/runtime/api/bun/h2_frame_parser.rs @@ -1138,6 +1138,63 @@ impl Drop for DispatchGuard<'_> { } } +/// Follows the byte stream `write()` emits over a JS-backed transport and reports when it +/// sits at a point where another frame may legally begin: between frames, and not inside a +/// header block (HEADERS / PUSH_PROMISE without END_HEADERS up to the CONTINUATION that carries +/// it, RFC 9113 §4.3). See `write_to_js_transport`. +#[derive(Clone, Copy, Default)] +struct TxFrameTracker { + /// Payload bytes still owed on the current frame. + remaining: u32, + /// A frame header split across chunks is collected here until all 9 bytes are known. + header: [u8; FrameHeader::BYTE_SIZE], + header_len: u8, + /// A HEADERS/PUSH_PROMISE/CONTINUATION without END_HEADERS went out; the block is open. + header_block_open: bool, +} + +impl TxFrameTracker { + fn at_boundary(&self) -> bool { + self.remaining == 0 && self.header_len == 0 && !self.header_block_open + } + + fn advance(&mut self, mut chunk: &[u8]) { + const CONNECTION_PREFACE: &[u8] = crate::api::h2::wire::CONNECTION_PREFACE; + while !chunk.is_empty() { + if self.remaining > 0 { + let take = (self.remaining as usize).min(chunk.len()); + self.remaining -= take as u32; + chunk = &chunk[take..]; + continue; + } + if self.header_len == 0 && chunk.starts_with(CONNECTION_PREFACE) { + // The client magic precedes the first SETTINGS frame; it is not a frame. + chunk = &chunk[CONNECTION_PREFACE.len()..]; + continue; + } + let have = self.header_len as usize; + let take = (FrameHeader::BYTE_SIZE - have).min(chunk.len()); + self.header[have..have + take].copy_from_slice(&chunk[..take]); + self.header_len += take as u8; + chunk = &chunk[take..]; + if self.header_len as usize == FrameHeader::BYTE_SIZE { + let header = FrameHeader::decode(&self.header); + self.header_len = 0; + self.remaining = header.length; + // PUSH_PROMISE is not a FrameType variant (the inbound path matches it raw too). + const PUSH_PROMISE: u8 = 0x05; + if header.type_ == FrameType::HTTP_FRAME_HEADERS as u8 + || header.type_ == PUSH_PROMISE + || header.type_ == FrameType::HTTP_FRAME_CONTINUATION as u8 + { + self.header_block_open = + header.flags & HeadersFrameFlags::END_HEADERS as u8 == 0; + } + } + } + } +} + /// The `+1` a native frame holds on the parser while it runs code that can free it (an inbound /// dispatch, a write that re-enters JS). Live guards are counted in /// `H2FrameParser::native_keepalives` so `finalize` can release the ones whose frame will never @@ -1308,6 +1365,9 @@ pub struct H2FrameParser { /// never contends with the engine borrow. engine_frames_received: Cell, engine_frames_sent: Cell, + /// Where the bytes emitted through `write()` over a JS-backed transport stand relative to + /// frame and header-block boundaries. + tx_tracker: Cell, ref_count: bun_ptr::RefCount, // intrusive — bun.ptr.RefCount(@This(), "ref_count", deinit, .{}) /// Number of live `Keepalive` guards: the `+1`s held by native frames currently on the stack. /// Read only by `release_refs_stranded_by_exit()`. @@ -2997,6 +3057,12 @@ impl H2FrameParser { if self.js_socket_flushing.get() { return 0; } + if !self.tx_tracker.get().at_boundary() { + // Mid-frame or mid-header-block (see write_to_js_transport): flushing now would + // put the cork or write_buffer inside that unit. It completes synchronously and + // the cork's auto-flush is already registered. + return 0; + } // Keep `self` alive across the re-entrant JS calls below. let _keepalive = self.keepalive(); @@ -3489,6 +3555,9 @@ impl H2FrameParser { return self._write(bytes); } self.cork(); + if matches!(self.native_socket.get(), BunSocket::None) { + return self.write_to_js_transport(bytes); + } let mut ok = true; loop { let off = CORK_OFFSET.with(|c| c.get()) as usize; @@ -3521,6 +3590,63 @@ impl H2FrameParser { bytes = &bytes[avail..]; } } + + /// `write()` for a session with no native socket, whose bytes reach the wire through the + /// `onWrite` handler (`socket.write()` on a JS stream). That call runs the transport's + /// `_write` synchronously, and user code there can serialize another frame (ping(), + /// settings(), goaway(), request()) or flush before it returns. Bytes are therefore only + /// handed over where another frame may legally follow: at a frame boundary outside a header + /// block. A unit that overflows the cork is assembled in the (empty at this point) batch + /// scratch and written whole once its last chunk arrives; the producers emit those chunks + /// back to back, so no JS can run while the scratch holds a partial unit, and a frame + /// serialized re-entrantly corks up behind the unit instead of landing inside it. + fn write_to_js_transport(&self, bytes: &[u8]) -> bool { + let mut tracker = self.tx_tracker.get(); + tracker.advance(bytes); + self.tx_tracker.set(tracker); + let at_boundary = tracker.at_boundary(); + // A non-empty batch scratch here is the partial unit from this write's earlier chunks + // (send_data's multi-frame batching never re-enters write()). + if BATCH_BUFFER.with_borrow(|batch| batch.is_empty()) { + let off = CORK_OFFSET.with(|c| c.get()) as usize; + if bytes.len() <= H2_CORK_BUFFER_SIZE - off { + CORK_OFFSET.with(|c| c.set((off + bytes.len()) as u16)); + CORK_BUFFER.with_borrow_mut(|buf| { + buf[off..off + bytes.len()].copy_from_slice(bytes); + }); + return true; + } + if off == 0 && at_boundary { + // Nothing corked and the chunk is whole frames: send it directly. + return self._write(bytes); + } + } + BATCH_BUFFER.with_borrow_mut(|batch| { + if batch.is_empty() { + // The corked prefix precedes this unit on the wire. + self.drain_cork_into(batch); + } + batch.extend_from_slice(bytes); + }); + if !at_boundary { + return true; + } + // The unit is complete: hand it over whole. Take the scratch out first, _write + // re-enters JS and a nested send_data must find the batch empty. + let mut data = BATCH_BUFFER.with_borrow_mut(core::mem::take); + let ok = self._write(&data); + data.clear(); + const BATCH_CAPACITY_CAP: usize = 1 << 20; + if data.capacity() > BATCH_CAPACITY_CAP { + data.shrink_to(BATCH_CAPACITY_CAP); + } + BATCH_BUFFER.with_borrow_mut(|b| { + if b.capacity() == 0 { + *b = data; + } + }); + ok + } } // Note: raw-ptr slice — the payload may alias `this.readBuffer` across @@ -9251,6 +9377,17 @@ impl H2FrameParser { if padding != 0 { flags |= HeadersFrameFlags::PADDED as u8; + // Grow before any frame byte is written: failing after the header went out + // would abandon the frame mid-serialization (the JS-transport tracker would + // hold the stream mid-frame and the wire would owe a payload). + if encoded_headers + .try_reserve(encoded_size + padding_overhead - encoded_headers.len()) + .is_err() + { + return Err( + global_object.throw(format_args!("Failed to allocate padding buffer")) + ); + } } let frame = FrameHeader { @@ -9274,16 +9411,9 @@ impl H2FrameParser { // Handle padding if padding != 0 { - if encoded_headers - .try_reserve(encoded_size + padding_overhead - encoded_headers.len()) - .is_err() - { - return Err( - global_object.throw(format_args!("Failed to allocate padding buffer")) - ); - } // Zero-fill the padding region (RFC 7540 §6.2: padding octets MUST be zero) and - // ensure the slice we hand to writer covers only initialized bytes. + // ensure the slice we hand to writer covers only initialized bytes. Cannot + // allocate: the capacity was reserved above, before the frame header went out. encoded_headers.resize(encoded_size + padding_overhead, 0); let buffer = encoded_headers.as_mut_slice(); // memmove: shift right by 1 to make room for the pad-length byte @@ -9621,6 +9751,7 @@ impl H2FrameParser { frames_sent_legacy: Cell::new(0), engine_frames_received: Cell::new(0), engine_frames_sent: Cell::new(0), + tx_tracker: Cell::new(TxFrameTracker::default()), auto_flusher: JsCell::new(AutoFlusher::default()), padding_strategy: Cell::new(PaddingStrategy::None), engine: core::cell::RefCell::new(None), @@ -9842,6 +9973,7 @@ impl H2FrameParser { // capacity must be released here. Drop-and-replace = free. self.read_buffer.set(MutableString::default()); self.write_buffer.with_mut(|wb| wb.clear_and_free()); + self.tx_tracker.set(TxFrameTracker::default()); // Drop every per-stream JS context root; the parser is detaching. self.sctx.with_mut(|m| m.clear()); self.write_buffer_offset.set(0); diff --git a/test/js/node/http2/node-http2.test.js b/test/js/node/http2/node-http2.test.js index 8be58511e215..1bb518a8e698 100644 --- a/test/js/node/http2/node-http2.test.js +++ b/test/js/node/http2/node-http2.test.js @@ -4098,3 +4098,177 @@ it("remoteSettings/localSettings are never null before the peer's SETTINGS arriv server.close(); } }); + +describe("frames issued from inside a user-supplied Duplex transport's _write", () => { + // A transport wrapper that sends a frame of its own from its _write (a keepalive ping, a + // settings update, a goaway, another request) must see that frame sequenced AFTER the unit + // whose bytes it is currently being handed: never spliced between a frame's header and payload, + // never between a HEADERS frame and its CONTINUATIONs, and no byte may be handed to it twice. + // The session used to flush its 16 KiB cork mid-frame (re-entering _write) and cork the frame's + // remainder behind whatever the nested call serialized, so the nested frame landed inside the + // DATA payload and the body's tail spilled past the declared length (peer framing desync). + const PREFACE = Buffer.from("PRI * HTTP/2.0\r\n\r\nSM\r\n\r\n"); + const FRAME = { DATA: 0, HEADERS: 1, SETTINGS: 4, PUSH_PROMISE: 5, PING: 6, GOAWAY: 7, CONTINUATION: 9 }; + const END_HEADERS = 0x4; + function parseFrames(buf) { + let i = buf.subarray(0, PREFACE.length).equals(PREFACE) ? PREFACE.length : 0; + const frames = []; + while (i + 9 <= buf.length) { + const len = buf.readUIntBE(i, 3); + const end = i + 9 + len; + if (end > buf.length) break; + frames.push({ + type: buf[i + 3], + flags: buf[i + 4], + len, + streamId: buf.readUInt32BE(i + 5) & 0x7fffffff, + payload: buf.subarray(i + 9, end), + }); + i = end; + } + // RFC 9113 4.3: from a HEADERS/PUSH_PROMISE without END_HEADERS to the CONTINUATION that + // carries it, no other frame may appear on the connection. + let headerBlocksContiguous = true; + for (let k = 0; k < frames.length; k++) { + const f = frames[k]; + if ([FRAME.HEADERS, FRAME.PUSH_PROMISE, FRAME.CONTINUATION].includes(f.type) && !(f.flags & END_HEADERS)) { + const next = frames[k + 1]; + if (next && !(next.type === FRAME.CONTINUATION && next.streamId === f.streamId)) headerBlocksContiguous = false; + } + } + return { frames, complete: i === buf.length, headerBlocksContiguous }; + } + const nestedBody = Buffer.alloc(40000, 0x42); + const nested = { + ping: { type: FRAME.PING, issue: sess => sess.ping(Buffer.from("NESTPING"), () => {}) }, + settings: { type: FRAME.SETTINGS, issue: sess => sess.settings({ enablePush: false }) }, + goaway: { type: FRAME.GOAWAY, issue: sess => sess.goaway(0, 0, Buffer.from("NESTGOAWAY")) }, + request: { + type: FRAME.HEADERS, + issue: sess => sess.request({ ":method": "GET", ":path": "/nested" }, { endStream: true }).on("error", () => {}), + }, + // A request whose body spans several DATA frames (the batched multi-frame send path). + data: { + type: FRAME.HEADERS, + issue: sess => { + const r = sess.request({ ":method": "POST", ":path": "/nested" }, { endStream: false }); + r.on("error", () => {}); + r.end(nestedBody); + }, + }, + }; + // outer "data": the nested call fires while the 16374-byte body's DATA frame overflows the cork + // behind the corked ~11 KiB HEADERS. outer "continuation": it fires while a header block larger + // than one frame (HEADERS + CONTINUATION) is being handed to the transport. + const cases = []; + for (const kind of Object.keys(nested)) { + cases.push( + [kind, "data", false], + [kind, "data", true], + [kind, "continuation", false], + [kind, "continuation", true], + ); + } + + it.each(cases)( + "nested %s() issued during the outer %s write (transport backpressured: %p)", + async (kind, outer, backpressured) => { + const chunks = []; + let armed = false; + let issued = false; + let sess; + const transport = new Duplex({ + // A 1-byte highWaterMark makes every socket.write() report backpressure, which routes the + // session's follow-up bytes through its native pending buffer instead of straight to JS. + writableHighWaterMark: backpressured ? 1 : undefined, + read() {}, + write(chunk, enc, cb) { + chunks.push(Buffer.from(chunk)); + if (armed && !issued) { + issued = true; + nested[kind].issue(sess); + } + cb(); + }, + }); + + sess = http2.connect("http://localhost:1", { createConnection: () => transport }); + sess.on("error", () => {}); + try { + await new Promise(resolve => sess.once("connect", resolve)); + + const body = Buffer.alloc(16374, 0x41); + // 15000 'p's HPACK-encode to ~11 KiB (one HEADERS frame); 30000 to ~22 KiB, which needs a + // CONTINUATION frame after a full 16384-byte HEADERS frame. + const pad = Buffer.alloc(outer === "continuation" ? 30000 : 15000, 0x70).toString(); + if (outer === "continuation") armed = true; + const req = sess.request({ ":method": "POST", ":path": "/", "x-pad": pad }, { endStream: false }); + req.on("error", () => {}); + armed = true; + req.write(body); + + // Wait (by condition, not time) until the outer frames and the nested frames are all out. + const done = parsed => { + if (!parsed.complete) return false; + const di = parsed.frames.findIndex(f => f.type === FRAME.DATA && f.streamId === 1); + if (di === -1) return false; + if (!parsed.frames.some(f => f.type === nested[kind].type && (f.type !== FRAME.HEADERS || f.streamId === 3))) + return false; + if (kind === "data") { + const got = parsed.frames + .filter(f => f.type === FRAME.DATA && f.streamId === 3) + .reduce((n, f) => n + f.len, 0); + if (got < nestedBody.length) return false; + } + return true; + }; + let parsed; + for (let tick = 0; tick < 400; tick++) { + await new Promise(resolve => setImmediate(resolve)); + parsed = parseFrames(Buffer.concat(chunks)); + if (done(parsed)) break; + } + + expect(issued).toBe(true); + // Every byte the transport received belongs to exactly one complete frame, and header + // blocks are never interleaved with other frames. + expect(parsed.complete).toBe(true); + expect(parsed.headerBlocksContiguous).toBe(true); + // One connection preface + SETTINGS (a second one mid-stream would not even parse), one + // request header block on stream 1, one DATA frame carrying exactly the body. + expect(parsed.frames.filter(f => f.type === FRAME.SETTINGS).length).toBe(kind === "settings" ? 2 : 1); + expect(parsed.frames.filter(f => f.type === FRAME.HEADERS && f.streamId === 1).length).toBe(1); + expect(parsed.frames.some(f => f.type === FRAME.CONTINUATION && f.streamId === 1)).toBe( + outer === "continuation", + ); + const dataFrames = parsed.frames.filter(f => f.type === FRAME.DATA && f.streamId === 1); + expect(dataFrames.length).toBe(1); + expect(dataFrames[0].payload.equals(body)).toBe(true); + // The nested frames are well-formed frames of their own, after the unit they were issued + // from: the stream-1 header block, and (when issued during the body write) its DATA frame. + const blockEnd = parsed.frames.findIndex( + f => [FRAME.HEADERS, FRAME.CONTINUATION].includes(f.type) && f.streamId === 1 && f.flags & END_HEADERS, + ); + expect(blockEnd).toBeGreaterThanOrEqual(0); + const issuedAfter = outer === "continuation" ? blockEnd : parsed.frames.indexOf(dataFrames[0]); + const after = parsed.frames.slice(issuedAfter + 1); + const before = parsed.frames.slice(0, issuedAfter + 1); + expect(before.filter(f => f.streamId === 3).length).toBe(0); + if (kind === "request" || kind === "data") { + expect(after.filter(f => f.type === FRAME.HEADERS && f.streamId === 3).length).toBe(1); + } else { + expect(before.slice(1).filter(f => f.type === nested[kind].type).length).toBe(0); + expect(after.filter(f => f.type === nested[kind].type).length).toBe(1); + } + if (kind === "data") { + const nestedData = Buffer.concat( + after.filter(f => f.type === FRAME.DATA && f.streamId === 3).map(f => f.payload), + ); + expect(nestedData.equals(nestedBody)).toBe(true); + } + } finally { + sess.destroy(); + } + }, + ); +});