node:http2: make the stream lifecycle getters total, like node

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);
       }

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<JSValue> {
-        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());

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,

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();
+    }
+  });
+});