diff --git a/src/event_loop/DeferredTaskQueue.rs b/src/event_loop/DeferredTaskQueue.rs index 0889c596ef81..64d3460f25f5 100644 --- a/src/event_loop/DeferredTaskQueue.rs +++ b/src/event_loop/DeferredTaskQueue.rs @@ -50,6 +50,8 @@ pub type DeferredRepeatingTask = unsafe extern "C" fn(*mut c_void) -> bool; #[derive(Default)] pub struct DeferredTaskQueue { pub(crate) map: ArrayHashMap>, DeferredRepeatingTask>, + /// An entry was posted since the last pass started. + unrun: bool, } impl DeferredTaskQueue { @@ -60,11 +62,19 @@ impl DeferredTaskQueue { bun_collections::hash_map::Entry::Occupied(_) => true, bun_collections::hash_map::Entry::Vacant(v) => { v.insert(task); + self.unrun = true; false } } } + /// Whether an entry may still be waiting for its first run. Clears the mark. + pub fn take_unrun(&mut self) -> bool { + let unrun = self.unrun && !self.map.is_empty(); + self.unrun = false; + unrun + } + pub fn unregister_task(&mut self, ctx: Option>) -> bool { // Order is irrelevant for this map's contract (see file doc), so // swap-remove is fine and O(1). @@ -72,6 +82,7 @@ impl DeferredTaskQueue { } pub fn run(&mut self) { + self.unrun = false; // Callbacks may re-entrantly mutate `self.map` (see the re-entrancy // note in the file doc), so re-read `len()` every iteration and // re-check slot `i` after each callback. `remaining` is a livelock bound. diff --git a/src/js/node/http2.ts b/src/js/node/http2.ts index abfd52512f03..557c8fc537e5 100644 --- a/src/js/node/http2.ts +++ b/src/js/node/http2.ts @@ -2719,11 +2719,6 @@ class Http2Stream extends Duplex { } else { this.emit("wantTrailers"); } - // Hand the END_STREAM (or trailer) frame to the socket now: with deferred write - // completion, _final can run on the program's last live turn and a frame left in - // the cork for the auto-flusher would strand a generic-streams (duplexPair) peer - // waiting for 'end'. - native.flush(); if (this[bunHTTP2StreamFinal] !== callback) { // markWritableDone already consumed and invoked the callback during the // synchronous streamEnd dispatch above. @@ -2735,9 +2730,6 @@ class Http2Stream extends Duplex { return; } const settled = native.writeStream(this.#id, "", "ascii", true, callback); - // Same as above: don't leave the END_STREAM frame in the cork on what may be the - // program's last live turn. - native.flush(); if (settled === 5) { // HALF_CLOSED_LOCAL settled synchronously; the dispatch was suppressed. markWritableDone(this); diff --git a/src/jsc/bindings/ZigGlobalObject.cpp b/src/jsc/bindings/ZigGlobalObject.cpp index e99812a33747..ac454120182e 100644 --- a/src/jsc/bindings/ZigGlobalObject.cpp +++ b/src/jsc/bindings/ZigGlobalObject.cpp @@ -3163,6 +3163,15 @@ extern "C" uint8_t JSC__JSGlobalObject__drainMicrotasks(Zig::GlobalObject* globa return globalObject->drainMicrotasks(); } +extern "C" bool JSC__JSGlobalObject__hasPendingMicrotasks(Zig::GlobalObject* globalObject) +{ + auto& vm = globalObject->vm(); + if (!vm.defaultMicrotaskQueue().isEmpty()) + return true; + auto* nextTickQueue = globalObject->m_nextTickQueue.get(); + return nextTickQueue && !nextTickQueue->isEmpty(); +} + template static void visitGlobalObjectMember(Visitor& visitor, T& anything) { anything.visit(visitor); diff --git a/src/jsc/event_loop.rs b/src/jsc/event_loop.rs index a44ba7575b59..0b3c518d739a 100644 --- a/src/jsc/event_loop.rs +++ b/src/jsc/event_loop.rs @@ -154,6 +154,12 @@ mod drain_result { // the microtask queue through it is interior mutation invisible to Rust. unsafe extern "C" { safe fn JSC__JSGlobalObject__drainMicrotasks(global: &JSGlobalObject) -> u8; + safe fn JSC__JSGlobalObject__hasPendingMicrotasks(global: &JSGlobalObject) -> bool; +} + +/// No-op yield task: the tick that runs it ends in a checkpoint, which runs the deferred queue. +fn deferred_task_checkpoint(_queue: *mut DeferredTaskQueue::DeferredTaskQueue) -> JsResult<()> { + Ok(()) } impl JSGlobalObject { @@ -439,6 +445,25 @@ impl EventLoop { self.deferred_tasks.run(); vm.is_inside_deferred_task_queue.set(false); + // A deferred flush into a JS transport runs user code: drain what it queued. + if JSC__JSGlobalObject__hasPendingMicrotasks(global_object) { + match JSC__JSGlobalObject__drainMicrotasks(global_object) { + drain_result::SUCCESS => {} + drain_result::STOPPED => return Err(Stopped), + drain_result::PENDING_EXCEPTION => return Ok(()), + _ => unreachable!(), + } + } + + // A deferred task posted too late for the pass: keep the loop alive for one more. + if self.deferred_tasks.take_unrun() { + let task = ManagedTask::ManagedTask::new( + core::ptr::from_mut(&mut self.deferred_tasks), + deferred_task_checkpoint, + ); + self.enqueue_task_after_yield(task); + } + // Guard on `event_loop_handle` being set, but drain via `uws_loop_mut()`: // on Windows the uSockets loop (`uws::Loop::get()`) is NOT // `event_loop_handle` (which is the libuv loop). diff --git a/src/runtime/api/bun/h2_frame_parser.rs b/src/runtime/api/bun/h2_frame_parser.rs index d0e9f4267aee..a05cfcefa4d3 100644 --- a/src/runtime/api/bun/h2_frame_parser.rs +++ b/src/runtime/api/bun/h2_frame_parser.rs @@ -872,6 +872,8 @@ thread_local! { static CORK_BUFFER: RefCell> = RefCell::new(Box::new([0u8; H2_CORK_BUFFER_SIZE])); static CORK_OFFSET: Cell = const { Cell::new(0) }; + // Bumped when the cork is emptied or changes owner; a `CorkMark` is valid while it matches. + static CORK_GENERATION: Cell = const { Cell::new(0) }; // Multi-frame DATA batches (send_data): all frame headers + payload slices of one // write are serialized here and hit the socket in a single _write, instead of one // cork flush per 16 KB frame. Reused across calls; capacity is capped after use. @@ -897,6 +899,24 @@ thread_local! { const { RefCell::new(None) }; } +/// Where a frame starts in `CORK_BUFFER`, valid while `CORK_GENERATION` still matches. +#[derive(Clone, Copy)] +struct CorkMark { + offset: u16, + generation: u32, +} + +/// Corked DATA frames cut out behind a mark, with the tail offset of each END_STREAM frame. +struct CorkTail { + bytes: Vec, + end_streams: Vec<(u32, u16)>, +} + +fn reset_cork_offset() { + CORK_OFFSET.with(|c| c.set(0)); + CORK_GENERATION.with(|c| c.set(c.get().wrapping_add(1))); +} + /// One wire-order piece of a multi-frame send_data batch (see BATCH_SEGMENTS). #[derive(Clone, Copy)] enum BatchSegment { @@ -1334,6 +1354,8 @@ pub struct Stream { // The JS readable for this stream is paused (setStreamReading(id, false)): the engine defers // replenishing the stream's receive window until reading resumes, backpressuring the peer. reading_paused: bool, + // Cork position of this stream's END_STREAM DATA frame; `push_promise` writes ahead of it. + end_stream_cork: Option, // when we have backpressure we queue the data e round robin the Streams data_frame_queue: PendingQueue, @@ -1839,6 +1861,7 @@ impl Stream { remote_used_window_size: 0, signal: None, reading_paused: false, + end_stream_cork: None, data_frame_queue: PendingQueue::default(), } } @@ -2507,7 +2530,7 @@ impl H2FrameParser { Self::set_corked(Some(self.ref_guard())); self.register_auto_flush(); bun_output::scoped_log!(H2FrameParser, "cork {:p}", self); - CORK_OFFSET.with(|c| c.set(0)); + reset_cork_offset(); } pub(crate) fn generic_flush(&self, mut socket: S) -> usize { @@ -2969,6 +2992,10 @@ impl H2FrameParser { } } + fn holds_cork(&self) -> bool { + Self::corked().is_some_and(|corked| std::ptr::eq(corked, self.as_ctx_ptr())) + } + /// Runs from the deferred tick (never under a write): closes the native socket so the /// normal socket-close teardown runs (native callback detach, JS 'close', session /// destroy) - the same path a peer disconnect takes. Closes WITHOUT detaching: a @@ -3000,6 +3027,10 @@ impl H2FrameParser { pub(crate) fn on_auto_flush(&self) -> bool { let _keepalive = self.keepalive(); + if self.transport_write_fatal.get() && !self.has_backpressure() { + // A later write drained the buffer: the transport recovered, so this tick flushes. + self.transport_write_fatal.set(false); + } if self.transport_write_fatal.get() { // Returning `false` makes DeferredTaskQueue::run remove the entry // itself, so only the registration's flag and ref are released here @@ -3009,14 +3040,7 @@ impl H2FrameParser { // early-return instead of removing a map entry run() still owns. self.auto_flusher.get().registered.set(false); self.deref(); - // An empty write buffer here means a later write in the same flush() - // cycle already drained the bytes the failing send left behind (racy - // one-off errnos, e.g. macOS EPROTOTYPE) - the transport recovered. - if self.has_backpressure() { - self.close_transport_after_fatal_write(); - } else { - self.transport_write_fatal.set(false); - } + self.close_transport_after_fatal_write(); return false; } if self.pending_header_compression_error.get() { @@ -3041,10 +3065,84 @@ impl H2FrameParser { return false; } let _ = self.flush(); - // we will unregister ourselves when the buffer is empty + let still_owed = self.holds_cork() + || self.transport_write_fatal.get() + || self.pending_header_compression_error.get(); + if self.auto_flusher.get().registered.get() && !still_owed { + // Only uncork() releases a registration, and nothing is corked. + self.auto_flusher.get().registered.set(false); + self.deref(); + return false; + } true } + /// The cork position the next `write()` lands at. + fn cork_mark(&self) -> CorkMark { + self.cork(); + CorkMark { + offset: CORK_OFFSET.with(|c| c.get()), + generation: CORK_GENERATION.with(|c| c.get()), + } + } + + /// Cut the corked bytes from a still-valid `mark` to the end of the cork. + fn take_cork_tail(&self, mark: CorkMark) -> Option { + if !ENABLE_AUTO_CORK || Self::corked() != Some(self.as_ctx_ptr()) { + return None; + } + if CORK_GENERATION.with(|c| c.get()) != mark.generation { + return None; + } + let off = CORK_OFFSET.with(|c| c.get()); + if mark.offset >= off { + return None; + } + let tail = CORK_BUFFER.with_borrow(|buf| buf[mark.offset as usize..off as usize].to_vec()); + // Only DATA may move behind a header block: HPACK must decode header blocks in encode order. + let mut end_streams: Vec<(u32, u16)> = Vec::new(); + let mut pos = 0usize; + while pos < tail.len() { + let raw = tail.get(pos..pos + FrameHeader::BYTE_SIZE)?; + let header = FrameHeader::decode(raw.try_into().expect("frame header size")); + if header.type_ != FrameType::HTTP_FRAME_DATA as u8 { + return None; + } + if header.flags & DataFrameFlags::END_STREAM as u8 != 0 { + end_streams.push((header.stream_identifier, pos as u16)); + } + pos += FrameHeader::BYTE_SIZE + header.length as usize; + } + if pos != tail.len() { + return None; + } + CORK_OFFSET.with(|c| c.set(mark.offset)); + // Every mark past `mark.offset` now points at shifted bytes. + CORK_GENERATION.with(|c| c.set(c.get().wrapping_add(1))); + Some(CorkTail { + bytes: tail, + end_streams, + }) + } + + /// Write `tail` back to the cork and mark its END_STREAM frames again. + fn restore_cork_tail(&self, tail: CorkTail) { + let base = self.cork_mark(); + let _ = self.write(&tail.bytes); + for (stream_id, offset) in tail.end_streams { + // `write` can run JS over a JS-backed transport: look each stream up again. + let Some(stream) = self.streams.get().get(&stream_id).copied() else { + continue; + }; + let mark = CorkMark { + offset: base.offset + offset, + generation: base.generation, + }; + // SAFETY: stream is a *mut Stream from self.streams; valid while the map entry exists + unsafe { (*stream).end_stream_cork = Some(mark) }; + } + } + /// Move the cork buffer's current contents to the front of `out` without flushing /// to the socket, so a multi-frame batch carries the already-corked bytes (e.g. the /// response HEADERS frame) in the same write. @@ -3062,7 +3160,7 @@ impl H2FrameParser { if off == 0 { return; } - CORK_OFFSET.with(|c| c.set(0)); + reset_cork_offset(); CORK_BUFFER.with_borrow(|buf| out.extend_from_slice(&buf[0..off])); } @@ -5161,8 +5259,10 @@ impl H2FrameParser { enqueued = true; stream.queue_frame(self, b"", callback, close); } else { + let mark = can_close.then(|| self.cork_mark()); let mut writer = self.to_writer(); let _ = data_header.write(&mut writer, &self.frames_sent_legacy); + stream.end_stream_cork = mark; } } else { let mut offset: usize = 0; @@ -5249,8 +5349,12 @@ impl H2FrameParser { stream_identifier: stream_id, length: u32::try_from(payload_size).expect("int cast"), }; - if payload.len() <= MAX_PAYLOAD_SIZE_WITHOUT_FRAME { - // Single-frame payload: the cork coalesces it with neighbors. + if payload.len() <= MAX_PAYLOAD_SIZE_WITHOUT_FRAME || end_stream { + // The cork coalesces it with neighbors; a body's last frame is corked too. + if payload.len() > MAX_PAYLOAD_SIZE_WITHOUT_FRAME { + self.flush_batch_buffer(); + } + let mark = end_stream.then(|| self.cork_mark()); let mut writer = self.to_writer(); let _ = data_header.write(&mut writer, &self.frames_sent_legacy); if padding != 0 { @@ -5258,6 +5362,7 @@ impl H2FrameParser { } else { let _ = writer.write_all(slice); } + stream.end_stream_cork = mark; } else { // Multi-frame payload: accumulate header + slice in the batch so // the whole write reaches the socket in one syscall (and, on TLS, @@ -6203,61 +6308,67 @@ impl H2FrameParser { .map(|s| s.max_frame_size) .unwrap_or_else(|| this.local_settings.get().max_frame_size) as usize; let payload_size = 4 + encoded_headers.len(); + let mut frames: Vec = Vec::with_capacity( + payload_size + FrameHeader::BYTE_SIZE * (1 + payload_size / max_frame.max(1)), + ); + let promised_be = (promised_id & 0x7fff_ffff).swap_bytes(); if payload_size <= max_frame { // PUSH_PROMISE frame: 9-byte header + 4-byte promised stream id + the header block. - let mut hdr_buf = [0u8; FrameHeader::BYTE_SIZE + 4]; - let mut ws = FixedBufferStream::new(&mut hdr_buf); let frame = FrameHeader { type_: 0x05, flags: 0x04, // END_HEADERS stream_identifier: parent_id, length: payload_size as u32, }; - let _ = frame.write(&mut ws, &this.frames_sent_legacy); - let promised_be = (promised_id & 0x7fff_ffff).swap_bytes(); - let _ = ws.write_all(&promised_be.to_ne_bytes()); - let _ = this.write(&hdr_buf); - let _ = this.write(&encoded_headers); + let _ = frame.write(&mut frames, &this.frames_sent_legacy); + frames.extend_from_slice(&promised_be.to_ne_bytes()); + frames.extend_from_slice(&encoded_headers); } else { // §6.6/§6.10: an oversized block is split - the PUSH_PROMISE (without END_HEADERS) // carries the promised id + the first fragment, then CONTINUATION frames on the // parent stream carry the rest; the last one sets END_HEADERS. Mirrors the // send_trailers()/request() splitting. let first_chunk = max_frame - 4; - let mut hdr_buf = [0u8; FrameHeader::BYTE_SIZE + 4]; - let mut ws = FixedBufferStream::new(&mut hdr_buf); let frame = FrameHeader { type_: 0x05, flags: 0, // continued below stream_identifier: parent_id, length: max_frame as u32, }; - let _ = frame.write(&mut ws, &this.frames_sent_legacy); - let promised_be = (promised_id & 0x7fff_ffff).swap_bytes(); - let _ = ws.write_all(&promised_be.to_ne_bytes()); - let _ = this.write(&hdr_buf); - let _ = this.write(&encoded_headers[..first_chunk]); + let _ = frame.write(&mut frames, &this.frames_sent_legacy); + frames.extend_from_slice(&promised_be.to_ne_bytes()); + frames.extend_from_slice(&encoded_headers[..first_chunk]); let mut offset = first_chunk; while offset < encoded_headers.len() { let chunk = (encoded_headers.len() - offset).min(max_frame); let is_last = offset + chunk >= encoded_headers.len(); - let mut cont_buf = [0u8; FrameHeader::BYTE_SIZE]; - let mut cs = FixedBufferStream::new(&mut cont_buf); let cont = FrameHeader { type_: FrameType::HTTP_FRAME_CONTINUATION as u8, flags: if is_last { 0x04 } else { 0 }, // END_HEADERS on the final fragment stream_identifier: parent_id, length: chunk as u32, }; - let _ = cont.write(&mut cs, &this.frames_sent_legacy); - let _ = this.write(&cont_buf); - let _ = this.write(&encoded_headers[offset..offset + chunk]); + let _ = cont.write(&mut frames, &this.frames_sent_legacy); + frames.extend_from_slice(&encoded_headers[offset..offset + chunk]); offset += chunk; } } - let _ = this.flush(); + // RFC 9113 §6.6: the PUSH_PROMISE must precede the parent's END_STREAM, like nghttp2. + let corked_end_stream = this + .streams + .get() + .get(&parent_id) + .copied() + // SAFETY: stream is a *mut Stream from self.streams; valid while the map entry exists + .and_then(|stream| unsafe { (*stream).end_stream_cork.take() }) + .and_then(|mark| this.take_cork_tail(mark)); + let _ = this.write(&frames); + if let Some(tail) = corked_end_stream { + this.restore_cork_tail(tail); + } + // No flush: a further pushStream() in this tick must also land ahead of the END_STREAM. Ok(JSValue::js_number(promised_id as f64)) } @@ -7733,7 +7844,7 @@ impl H2FrameParser { // `uncork()` would `_write()` the corked bytes, which re-enters JS on a non-native // socket; the process is exiting, so drop them and just release the slot's ref. if Self::corked() == Some(self.as_ctx_ptr()) { - CORK_OFFSET.with(|c| c.set(0)); + reset_cork_offset(); Self::set_corked(None); } // Removes the deferred task (its ctx is `self`) and releases the ref it holds. diff --git a/test/js/node/http2/h2-conformance.test.ts b/test/js/node/http2/h2-conformance.test.ts index 414bf6c3b68b..9602b26ca692 100644 --- a/test/js/node/http2/h2-conformance.test.ts +++ b/test/js/node/http2/h2-conformance.test.ts @@ -800,6 +800,178 @@ describe("push stream states (checklist §5.1, RFC 9113 §6.4/§8.4)", () => { }); }); +describe("PUSH_PROMISE ordering (RFC 9113 §6.6)", () => { + // A PUSH_PROMISE is only valid on a stream that is open or half-closed (remote) for the + // sender, so it has to reach the wire before the parent's END_STREAM. nghttp2 serializes + // HEADERS and PUSH_PROMISE ahead of pending DATA, which lets node call end() and then + // pushStream() in the same tick. A node client cancels a push that arrives after the + // parent's END_STREAM. + function push(stream: http2.ServerHttp2Stream) { + stream.pushStream({ ":path": "/pushed" }, (err, pushed) => { + if (err) { + stream.emit("error", err); + return; + } + pushed.on("error", () => {}); + pushed.respond({ ":status": 200 }); + pushed.end("pushed"); + }); + } + + test.each([ + [ + "end() then pushStream()", + 1, + (stream: http2.ServerHttp2Stream) => { + stream.end(); + push(stream); + }, + ], + [ + "pushStream() then end()", + 1, + (stream: http2.ServerHttp2Stream) => { + push(stream); + stream.end(); + }, + ], + [ + 'end("body") then pushStream()', + 1, + (stream: http2.ServerHttp2Stream) => { + stream.end("body"); + push(stream); + }, + ], + [ + "end() then pushStream() twice", + 2, + (stream: http2.ServerHttp2Stream) => { + stream.end(); + push(stream); + push(stream); + }, + ], + // Bodies over the 16 KB cork take the multi-frame path: only the last DATA frame is corked. + [ + "end(16400 bytes) then pushStream()", + 1, + (stream: http2.ServerHttp2Stream) => { + stream.end(Buffer.alloc(16400, "x")); + push(stream); + }, + ], + [ + "end(40000 bytes) then pushStream()", + 1, + (stream: http2.ServerHttp2Stream) => { + stream.end(Buffer.alloc(40000, "x")); + push(stream); + }, + ], + ])("the PUSH_PROMISE precedes the parent's END_STREAM when the handler runs %s", async (_, pushes, handle) => { + const serverError = Promise.withResolvers(); + const pushServer = http2.createServer(); + pushServer.on("stream", (stream: http2.ServerHttp2Stream) => { + stream.on("error", err => serverError.reject(err)); + stream.respond({ ":status": 200 }); + handle(stream); + }); + pushServer.listen(0); + await once(pushServer, "listening"); + const c = await RawH2.connect((pushServer.address() as net.AddressInfo).port); + try { + c.sendPreface(); + c.sendEmptySettings(); + c.sendFrame(FrameType.HEADERS, 0x5 /* END_HEADERS | END_STREAM */, 1, requestHeaderBlock("GET")); + // The last pushed stream's body: by then every frame of the parent is on the wire too. + await Promise.race([ + c.waitFor(f => f.type === FrameType.DATA && f.streamId === 2 * pushes && (f.flags & 0x1) !== 0), + serverError.promise, + ]); + const order = c.frames + .filter(f => f.streamId !== 0) + .map(f => { + const name = Object.keys(FrameType).find(k => FrameType[k as keyof typeof FrameType] === f.type) ?? f.type; + const endStream = (f.type === FrameType.DATA || f.type === FrameType.HEADERS) && (f.flags & 0x1) !== 0; + return `${name}(${f.streamId}${endStream ? ",END_STREAM" : ""})`; + }); + const lastPushPromise = order.lastIndexOf("PUSH_PROMISE(1)"); + const parentEnd = order.indexOf("DATA(1,END_STREAM)"); + const expected = ["HEADERS(1)", "DATA(1,END_STREAM)"]; + for (let i = 1; i <= pushes; i++) + expected.push("PUSH_PROMISE(1)", `HEADERS(${2 * i})`, `DATA(${2 * i},END_STREAM)`); + expect({ + // A multi-frame parent body adds DATA(1) frames without END_STREAM; their count is not the point. + order: order.filter(f => f !== "DATA(1)").toSorted(), + pushPromisesBeforeParentEnd: lastPushPromise !== -1 && lastPushPromise < parentEnd, + }).toEqual({ order: expected.toSorted(), pushPromisesBeforeParentEnd: true }); + } finally { + serverError.promise.catch(() => {}); + c.destroy(); + pushServer.close(); + } + }); + + test("a push on the first of two streams ended together does not shift the second stream's frame", async () => { + // Two open (POST) streams are responded, ended, and pushed in one synchronous batch, so both + // END_STREAM DATA frames sit in the cork together. The first push cuts the cork ahead of its + // END_STREAM, which shifts the second stream's END_STREAM. The second push must still land + // ahead of that shifted frame (the cut re-marks it), and neither frame may be corrupted. + const serverError = Promise.withResolvers(); + const streams: http2.ServerHttp2Stream[] = []; + const pushServer = http2.createServer(); + pushServer.on("stream", (stream: http2.ServerHttp2Stream) => { + stream.on("error", err => serverError.reject(err)); + streams.push(stream); + if (streams.length < 2) return; + // Both streams are still open (the client sent no body), so end() half-closes them and + // leaves them pushable. Respond on both, then end both, then push on both. + for (const s of streams) s.respond({ ":status": 200 }); + streams[0].end("aa"); + streams[1].end("bbb"); + streams.forEach((s, i) => { + s.pushStream({ ":path": "/p" + i }, (err, pushed) => { + if (err) { + s.emit("error", err); + return; + } + pushed.on("error", () => {}); + pushed.respond({ ":status": 200 }); + pushed.end("pushed" + i); + }); + }); + }); + pushServer.listen(0); + await once(pushServer, "listening"); + const c = await RawH2.connect((pushServer.address() as net.AddressInfo).port); + try { + c.sendPreface(); + c.sendEmptySettings(); + // POST with END_HEADERS but no END_STREAM: both request streams stay open. + c.sendFrame(FrameType.HEADERS, 0x4, 1, requestHeaderBlock("POST")); + c.sendFrame(FrameType.HEADERS, 0x4, 3, requestHeaderBlock("POST")); + // The second push's body (stream 4) is the last frame the server writes. + await Promise.race([ + c.waitFor(f => f.type === FrameType.DATA && f.streamId === 4 && (f.flags & 0x1) !== 0), + serverError.promise, + ]); + expect(c.frames.find(f => f.type === FrameType.GOAWAY)).toBeUndefined(); + const pos = (type: number, id: number) => c.frames.findIndex(f => f.type === type && f.streamId === id); + expect({ + promise1BeforeEnd1: pos(FrameType.PUSH_PROMISE, 1) < pos(FrameType.DATA, 1), + promise3BeforeEnd3: pos(FrameType.PUSH_PROMISE, 3) < pos(FrameType.DATA, 3), + body1: c.frames.find(f => f.type === FrameType.DATA && f.streamId === 1)!.payload.toString(), + body3: c.frames.find(f => f.type === FrameType.DATA && f.streamId === 3)!.payload.toString(), + }).toEqual({ promise1BeforeEnd1: true, promise3BeforeEnd3: true, body1: "aa", body3: "bbb" }); + } finally { + serverError.promise.catch(() => {}); + c.destroy(); + pushServer.close(); + } + }); +}); + describe("inbound flow control after local end-stream (RFC 9113 §6.9)", () => { // Regression coverage for the test-http2-pipe failure mode: the server responds and ends its // side before the request body arrives, the request body is piped into a backpressured diff --git a/test/js/node/http2/node-http2-syscall-fault.test.ts b/test/js/node/http2/node-http2-syscall-fault.test.ts index 3b5ad908cc97..0cf0668a1623 100644 --- a/test/js/node/http2/node-http2-syscall-fault.test.ts +++ b/test/js/node/http2/node-http2-syscall-fault.test.ts @@ -3,6 +3,8 @@ import { afterEach, describe, expect, test } from "bun:test"; import { bunEnv, bunExe, tls as certs, isASAN, isWindows } from "harness"; import { once } from "node:events"; import http2 from "node:http2"; +import net from "node:net"; +import { constants as osConstants } from "node:os"; import path from "node:path"; const skip = !fault.available() || isWindows; @@ -20,7 +22,7 @@ async function makeServer(handler: (stream: http2.ServerHttp2Stream, headers: ht server.on("sessionError", () => {}); server.listen(0, "127.0.0.1"); await once(server, "listening"); - const port = (server.address() as import("node:net").AddressInfo).port; + const port = (server.address() as net.AddressInfo).port; return { port, url: `http://127.0.0.1:${port}`, @@ -186,6 +188,49 @@ describe.skipIf(skip)("node:http2 under injected syscall faults", () => { } }); + // A send the kernel rejects latches the parser's fatal-write flag, and a deferred tick decides + // what to do about it. When the next send goes through, the transport has recovered and the + // tick only clears the latch. It used to return right there, although it is the same tick the + // corked HEADERS frame was registered for, so the frame stayed in the cork. macOS hit this on + // every connect while the preface was still written to a connecting socket (ENOTCONN there, + // EAGAIN on Linux). The peer below stays silent until it has the HEADERS, so nothing else + // pushes them out. + test.each([ + ["a request that ends with its HEADERS", {}, undefined], + ["a request that stays open", { ":method": "POST" }, { endStream: false }], + ])("send → one ENOTCONN, then recovery: HEADERS of %s still reach the peer", async (_, headers, options) => { + const gotHeaders = Promise.withResolvers(); + let received = Buffer.alloc(0); + const server = net.createServer(socket => { + socket.on("error", () => {}); + socket.on("data", chunk => { + received = Buffer.concat([received, chunk]); + // Skip the 24-byte preface, then walk the frames looking for HEADERS (type 1) on stream 1. + for (let at = 24; at + 9 <= received.length; at += 9 + received.readUIntBE(at, 3)) { + if (received[at + 3] === 1 && (received.readUInt32BE(at + 5) & 0x7fffffff) === 1) gotHeaders.resolve(); + } + }); + }); + server.listen(0, "127.0.0.1"); + await once(server, "listening"); + const port = (server.address() as net.AddressInfo).port; + // The first send is the preface, flushed when the socket connects. The session flushes again + // right after, and that send drains what the first one left buffered. + fault.set({ syscall: "send", action: "errno", errno: osConstants.errno.ENOTCONN, repeat: 1 }); + const client = http2.connect(`http://127.0.0.1:${port}`); + client.on("error", gotHeaders.reject); + client.on("close", () => gotHeaders.reject(new Error("session closed before its HEADERS left"))); + try { + const req = client.request({ ":path": "/", ...headers }, options); + req.on("error", () => {}); + await gotHeaders.promise; + } finally { + fault.clear(); + client.destroy(); + server.close(); + } + }); + test("https/2: recv → short reads (3 bytes) over TLS deliver complete response", async () => { const body = Buffer.alloc(256, "s"); const server = http2.createSecureServer({ key: certs.key, cert: certs.cert }); diff --git a/test/js/node/http2/node-http2.test.js b/test/js/node/http2/node-http2.test.js index 7cf5c3e7ab3a..0cac75c53f1c 100644 --- a/test/js/node/http2/node-http2.test.js +++ b/test/js/node/http2/node-http2.test.js @@ -6329,3 +6329,171 @@ describe("frames issued from inside a user-supplied Duplex transport's _write", }, ); }); + +describe.concurrent("session.request() over a transport that delivers synchronously", () => { + // An in-memory transport pair that hands every write to the other side from inside + // socket.write(), the way a mock socket in a unit test does. + function synchronousDuplexPair() { + class Side extends Duplex { + _read() {} + _write(chunk, _encoding, callback) { + this.peer.push(chunk); + callback(); + } + } + const a = new Side(); + const b = new Side(); + a.peer = b; + b.peer = a; + return [a, b]; + } + + function frame(type, flags, streamId, payload = Buffer.alloc(0)) { + const header = Buffer.alloc(9); + header.writeUIntBE(payload.length, 0, 3); + header[3] = type; + header[4] = flags; + header.writeUInt32BE(streamId, 5); + return Buffer.concat([header, Buffer.from(payload)]); + } + + // A peer that answers every HEADERS frame from inside its own 'data' handler: 200 with the + // body "ok". 0x88 is the static-table HPACK index for :status 200. + function replyToEveryRequestSynchronously(socket, onRequestFrame) { + const FRAME_HEADERS = 1; + const FRAME_DATA = 0; + const FRAME_SETTINGS = 4; + const PREFACE_LENGTH = 24; + let buffered = Buffer.alloc(0); + let prefaceSeen = false; + socket.write(frame(FRAME_SETTINGS, 0, 0)); + socket.on("data", chunk => { + buffered = Buffer.concat([buffered, chunk]); + if (!prefaceSeen) { + if (buffered.length < PREFACE_LENGTH) return; + buffered = buffered.subarray(PREFACE_LENGTH); + prefaceSeen = true; + } + while (buffered.length >= 9) { + const length = buffered.readUIntBE(0, 3); + if (buffered.length < 9 + length) break; + const type = buffered[3]; + const flags = buffered[4]; + const streamId = buffered.readUInt32BE(5); + buffered = buffered.subarray(9 + length); + if (type === FRAME_SETTINGS && !(flags & 1)) socket.write(frame(FRAME_SETTINGS, 1, 0)); + if (type === FRAME_HEADERS) { + onRequestFrame(streamId); + socket.write(frame(FRAME_HEADERS, 4, streamId, [0x88])); + socket.write(frame(FRAME_DATA, 1, streamId, "ok")); + } + } + }); + } + + // Node submits a session's frames from a scheduled write, so request() never reaches the + // transport itself. Bun flushed the HEADERS frame from inside request() (through the + // stream's _final), so over a transport that delivers synchronously the response came back + // before the caller could attach a 'response' listener: the status was lost while the body + // and 'close' still arrived. + it("submits nothing while request() is on the stack, so 'response' is never missed", async () => { + const [clientSide, peerSide] = synchronousDuplexPair(); + let insideRequestCall = false; + const peerSawRequestInsideCall = []; + replyToEveryRequestSynchronously(peerSide, () => peerSawRequestInsideCall.push(insideRequestCall)); + + const client = http2.connect("http://localhost", { createConnection: () => clientSide }); + try { + const results = []; + // The first request connects the session (its submit is deferred to 'connect' anyway); + // every later one runs on a session that is already connected. + for (const path of ["/1", "/2", "/3"]) { + insideRequestCall = true; + const req = client.request({ ":path": path }); + insideRequestCall = false; + + let status = 0; + let body = ""; + req.setEncoding("utf8"); + req.on("response", headers => (status = headers[":status"])); + req.on("data", chunk => (body += chunk)); + await new Promise((resolve, reject) => { + req.on("close", resolve); + req.on("error", reject); + }); + results.push({ status, body }); + } + + expect(results).toEqual([ + { status: 200, body: "ok" }, + { status: 200, body: "ok" }, + { status: 200, body: "ok" }, + ]); + expect(peerSawRequestInsideCall).toEqual([false, false, false]); + } finally { + client.destroy(); + } + }); + + // A session writes the frames it owes its transport at the end of the microtask checkpoint. + // Over a JS transport that write runs user code: the peer end parses the bytes, and its handlers + // queue nextTicks and cork frames of their own. Nothing else holds the event loop open for an + // in-process pair, so all of that has to be seen through before the loop can call the process + // done. It used to exit with the response written but never reported to the stream. + it.each([ + [ + "from inside the 'stream' handler", + `stream.respond({ ":status": 200 }); stream.write("he"); stream.end("llo");`, + ["response:200", "end:hello", "close"], + ], + [ + "from a nextTick queued by the 'stream' handler", + `process.nextTick(() => process.nextTick(() => stream.respond({ ":status": 204 }, { endStream: true })));`, + ["response:204", "end:", "close"], + ], + [ + "from a nextTick, with trailers sent from another", + `process.nextTick(() => { + stream.respond({ ":status": 200 }, { waitForTrailers: true }); + stream.on("wantTrailers", () => process.nextTick(() => stream.sendTrailers({ "x-trailer": "1" }))); + stream.end("hello"); + });`, + ["response:200", "end:hello", "close"], + ], + ])("the whole response arrives before the process exits when the server answers %s", async (_, respond, expected) => { + await using proc = Bun.spawn({ + cmd: [ + bunExe(), + "-e", + ` + const http2 = require("node:http2"); + const { duplexPair } = require("node:stream"); + const events = []; + process.on("exit", () => console.log(JSON.stringify(events))); + + const server = http2.createServer(); + server.on("stream", stream => { + ${respond} + }); + const [clientSide, serverSide] = duplexPair(); + server.emit("connection", serverSide); + + const client = http2.connect("http://localhost", { createConnection: () => clientSide }); + const req = client.request({ ":path": "/" }); + req.setEncoding("utf8"); + let body = ""; + req.on("response", headers => events.push("response:" + headers[":status"])); + req.on("data", chunk => (body += chunk)); + req.on("end", () => events.push("end:" + body)); + req.on("close", () => events.push("close")); + `, + ], + env: bunEnv, + stderr: "pipe", + }); + const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + expect(stderr).toBe(""); + expect(JSON.parse(stdout)).toEqual(expected); + expect(exitCode).toBe(0); + }); +});