node:http2: record empty trailers as sent when the stream ends without a 'wantTrailers' listener

src/js/node/http2.ts

@@ -393,6 +393,7 @@ const kRequestHeaders = Symbol("requestHeaders");
 // set so a second sendTrailers() in the same tick reports ERR_HTTP2_TRAILERS_ALREADY_SENT, not
 // ERR_HTTP2_INVALID_STREAM.
 const kSendingTrailers = Symbol("sendingTrailers");
+const kWantTrailers = Symbol("wantTrailers");
 const kSettingsAckGraceTimer = Symbol("settingsAckGraceTimer");
 // node: a socket can be bound to at most one Http2Session (ERR_HTTP2_SOCKET_BOUND).
 const kBoundSession = Symbol("boundSession");
@@ -2631,6 +2632,19 @@ class Http2Stream extends Duplex {
     }
   }
 
+  // https://github.com/nodejs/node/blob/v26.3.0/lib/internal/http2/core.js#L484-L493 (onStreamTrailers):
+  // a closed stream is not asked (close()'s RST_STREAM follows on its own); with no listener the
+  // stream sends empty trailers itself, which is what makes a later sendTrailers() fail.
+  [kWantTrailers](native) {
+    if ((this[bunHTTP2StreamStatus] & StreamState.Closed) !== 0) return;
+    if (this.listenerCount("wantTrailers") === 0) {
+      this.#sentTrailers = {};
+      native.noTrailers(this.#id);
+    } else {
+      this.emit("wantTrailers");
+    }
+  }
+
   setTimeout(timeout, callback) {
     const session = this[bunHTTP2Session];
     if (!session) return;
@@ -2910,22 +2924,7 @@ class Http2Stream extends Duplex {
             // markWritableDone before control returns here: stash the callback there so the
             // writable finishes ('finish' before 'close') instead of being torn down mid-final.
             this[bunHTTP2StreamFinal] = callback;
-            if ((this[bunHTTP2StreamStatus] & StreamState.Closed) !== 0 || this.listenerCount("wantTrailers") === 0) {
-              // No 'wantTrailers' listener — or the stream was close()d while the last write
-              // was still in flight, in which case node never asks for trailers (the compat
-              // onStreamTrailersReady would throw ERR_HTTP2_INVALID_STREAM on a closed
-              // stream): end the writable with the empty END_STREAM DATA frame directly.
-              native.noTrailers(this.#id);
-              // Mark trailers as "sent" so a later stream.sendTrailers()
-              // call hits the ERR_HTTP2_TRAILERS_ALREADY_SENT guard instead
-              // of invoking native noTrailers() a second time on an
-              // already-half-closed stream. The emit("wantTrailers") path
-              // below reaches the same result via sendTrailers({}) which
-              // assigns #sentTrailers itself.
-              this.#sentTrailers = {};
-            } else {
-              this.emit("wantTrailers");
-            }
+            this[kWantTrailers](native);
             // 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
@@ -4452,14 +4451,9 @@ class ServerHttp2Session extends Http2Session {
       if (!self || typeof stream !== "object") return;
       const status = stream[bunHTTP2StreamStatus];
       if ((status & StreamState.WantTrailer) !== 0) return;
-
       stream[bunHTTP2StreamStatus] = status | StreamState.WantTrailer;
-
-      if (stream.listenerCount("wantTrailers") === 0) {
-        self[bunHTTP2Native]?.noTrailers(stream.id);
-      } else {
-        stream.emit("wantTrailers");
-      }
+      const native = self[bunHTTP2Native];
+      if (native) stream[kWantTrailers](native);
     },
     goaway(self: ServerHttp2Session, errorCode: number, lastStreamId: number, opaqueData: Buffer) {
       if (!self) return;
@@ -5451,11 +5445,8 @@ class ClientHttp2Session extends Http2Session {
       const status = stream[bunHTTP2StreamStatus];
       if ((status & StreamState.WantTrailer) !== 0) return;
       stream[bunHTTP2StreamStatus] = status | StreamState.WantTrailer;
-      if (stream.listenerCount("wantTrailers") === 0) {
-        self[bunHTTP2Native]?.noTrailers(stream.id);
-      } else {
-        stream.emit("wantTrailers");
-      }
+      const native = self[bunHTTP2Native];
+      if (native) stream[kWantTrailers](native);
     }),
     goaway(self: ClientHttp2Session, errorCode: number, lastStreamId: number, opaqueData: Buffer) {
       if (!self) return;

test/js/node/http2/h2-conformance.test.ts

@@ -900,6 +900,213 @@ function requestHeaderBlock(method: "GET" | "POST", extra: Buffer = Buffer.alloc
 
 const CONTENT_LENGTH_5 = Buffer.concat([Buffer.from([0x0f, 0x0d]), hpackLiteral("5")]);
 
+function frameSummary(f: Frame) {
+  return { type: f.type, flags: f.flags, length: f.length };
+}
+
+// node's onStreamTrailers (lib/internal/http2/core.js): once the body of a waitForTrailers stream is
+// out, a stream without a 'wantTrailers' listener is ended by node itself via sendTrailers({}), so an
+// empty END_STREAM DATA frame goes out and the trailers count as sent: a later sendTrailers() must
+// fail with ERR_HTTP2_TRAILERS_ALREADY_SENT rather than put a trailer HEADERS frame on the
+// half-closed stream (§8.1: the trailer section is the last frame of the message). A stream that is
+// closed by then, or that a listener close()s instead of sending trailers, is not completed at all:
+// no END_STREAM frame, nothing recorded as sent, and close()'s RST_STREAM(NO_ERROR) ends it.
+describe("trailer block requests without a listener or on a closing stream (RFC 9113 §8.1)", () => {
+  test("client: a late sendTrailers() throws ERR_HTTP2_TRAILERS_ALREADY_SENT and sends nothing", async () => {
+    const raw = await RawH2Server.listen();
+    const client = http2.connect(`http://127.0.0.1:${raw.port}`);
+    client.on("error", () => {});
+    try {
+      const req = client.request({ ":method": "POST", ":path": "/" }, { waitForTrailers: true });
+      req.on("error", () => {});
+      await raw.waitFor(f => f.type === FrameType.SETTINGS && (f.flags & 0x1) === 0);
+      raw.sendFrame(FrameType.SETTINGS, 0, 0);
+      raw.sendFrame(FrameType.SETTINGS, 0x1, 0);
+      req.end("hello");
+      await raw.waitFor(f => f.streamId === 1 && f.type === FrameType.DATA && (f.flags & 0x1) !== 0);
+
+      expect(req.sentTrailers).toEqual({});
+      expect(() => req.sendTrailers({ "x-late": "1" })).toThrow(
+        expect.objectContaining({ code: "ERR_HTTP2_TRAILERS_ALREADY_SENT" }),
+      );
+      expect(req.sentTrailers).toEqual({});
+      // Anything the rejected call had written would reach the socket ahead of this PING's ACK.
+      raw.sendFrame(FrameType.PING, 0, 0, Buffer.alloc(8));
+      await raw.waitFor(f => f.type === FrameType.PING && (f.flags & 0x1) !== 0);
+      expect(raw.frames.filter(f => f.streamId === 1).map(frameSummary)).toEqual([
+        { type: FrameType.HEADERS, flags: 0x4 /* END_HEADERS */, length: expect.any(Number) },
+        { type: FrameType.DATA, flags: 0, length: 5 },
+        { type: FrameType.DATA, flags: 0x1 /* END_STREAM */, length: 0 },
+      ]);
+    } finally {
+      client.destroy();
+      raw.close();
+    }
+  });
+
+  test("server: sendTrailers() from 'prefinish' or later throws ERR_HTTP2_TRAILERS_ALREADY_SENT and sends nothing", async () => {
+    const outcome = Promise.withResolvers<{ prefinish: string; finish: string; sentTrailers: unknown }>();
+    const srv = http2.createServer();
+    srv.on("stream", (stream: any) => {
+      stream.on("error", (err: Error) => outcome.reject(err));
+      const attempt = () => {
+        try {
+          stream.sendTrailers({ "x-late": "1" });
+          return "sent";
+        } catch (e: any) {
+          return e.code;
+        }
+      };
+      // 'prefinish' fires from inside the call that writes the END_STREAM frame; 'finish' a tick
+      // later. Both are after the stream ended itself, so both attempts are too late.
+      let prefinish = "not emitted";
+      stream.on("prefinish", () => (prefinish = attempt()));
+      stream.on("finish", () => outcome.resolve({ prefinish, finish: attempt(), sentTrailers: stream.sentTrailers }));
+      stream.respond({ ":status": 200 }, { waitForTrailers: true });
+      stream.end("ok");
+    });
+    srv.listen(0, "127.0.0.1");
+    await once(srv, "listening");
+    const c = await RawH2.connect((srv.address() as net.AddressInfo).port);
+    try {
+      c.sendPreface();
+      c.sendEmptySettings();
+      // POST without END_STREAM: the request side stays open, so the response's END_STREAM only
+      // half-closes the stream (the stream object stays usable for the late sendTrailers calls).
+      c.sendFrame(FrameType.HEADERS, 0x4 /* END_HEADERS */, 1, requestHeaderBlock("POST"));
+      expect(await outcome.promise).toEqual({
+        prefinish: "ERR_HTTP2_TRAILERS_ALREADY_SENT",
+        finish: "ERR_HTTP2_TRAILERS_ALREADY_SENT",
+        sentTrailers: {},
+      });
+      c.sendFrame(FrameType.PING, 0, 0, Buffer.alloc(8));
+      await c.waitFor(f => f.type === FrameType.PING && (f.flags & 0x1) !== 0);
+      expect(c.frames.filter(f => f.streamId === 1).map(frameSummary)).toEqual([
+        { type: FrameType.HEADERS, flags: 0x4 /* END_HEADERS */, length: expect.any(Number) },
+        { type: FrameType.DATA, flags: 0, length: 2 },
+        { type: FrameType.DATA, flags: 0x1 /* END_STREAM */, length: 0 },
+      ]);
+    } finally {
+      c.destroy();
+      srv.close();
+    }
+  });
+
+  // onStreamTrailers returns without emitting once the stream is closed (a listener could only
+  // throw ERR_HTTP2_INVALID_STREAM); close()'s RST_STREAM goes out once the writable finishes.
+  test("client: 'wantTrailers' is not emitted on a stream close()d while its last chunk was still buffered", async () => {
+    const raw = await RawH2Server.listen();
+    const client = http2.connect(`http://127.0.0.1:${raw.port}`);
+    client.on("error", () => {});
+    const connected = once(client, "connect");
+    try {
+      await raw.waitFor(f => f.type === FrameType.SETTINGS && (f.flags & 0x1) === 0);
+      raw.sendFrame(FrameType.SETTINGS, 0, 0);
+      raw.sendFrame(FrameType.SETTINGS, 0x1, 0);
+      await connected;
+
+      const req = client.request({ ":method": "POST", ":path": "/" }, { waitForTrailers: true });
+      req.on("error", () => {});
+      let wantTrailers = 0;
+      req.on("wantTrailers", () => wantTrailers++);
+      // 'ready' follows the tick that uncorks a new request stream, so the first write below goes
+      // out at once. Its write callback only completes on a later turn, which leaves the chunk
+      // passed to end() buffered behind it: that final chunk (the one that asks for the trailer
+      // block) is written after close() has already marked the stream closed.
+      await once(req, "ready");
+      req.write("a");
+      req.end("b");
+      req.close();
+      expect(req.closed).toBe(true);
+
+      const rst = await raw.waitFor(f => f.streamId === 1 && f.type === FrameType.RST_STREAM);
+      expect(rst.payload.readUInt32BE(0)).toBe(ErrorCode.NO_ERROR);
+      expect(wantTrailers).toBe(0);
+      expect(req.sentTrailers).toBeUndefined();
+      expect(raw.frames.filter(f => f.streamId === 1).map(frameSummary)).toEqual([
+        { type: FrameType.HEADERS, flags: 0x4 /* END_HEADERS */, length: expect.any(Number) },
+        { type: FrameType.DATA, flags: 0, length: 1 },
+        { type: FrameType.DATA, flags: 0, length: 1 },
+        { type: FrameType.RST_STREAM, flags: 0, length: 4 },
+      ]);
+    } finally {
+      client.destroy();
+      raw.close();
+    }
+  });
+
+  test("client: a 'wantTrailers' listener that close()s the stream ends it with RST_STREAM(NO_ERROR) alone", async () => {
+    const raw = await RawH2Server.listen();
+    const client = http2.connect(`http://127.0.0.1:${raw.port}`);
+    client.on("error", () => {});
+    try {
+      const req = client.request({ ":method": "POST", ":path": "/" }, { waitForTrailers: true });
+      req.on("error", () => {});
+      const events: string[] = [];
+      req.on("wantTrailers", () => {
+        events.push("wantTrailers");
+        req.close();
+      });
+      req.on("finish", () => events.push("finish"));
+      const closed = once(req, "close");
+      await raw.waitFor(f => f.type === FrameType.SETTINGS && (f.flags & 0x1) === 0);
+      raw.sendFrame(FrameType.SETTINGS, 0, 0);
+      raw.sendFrame(FrameType.SETTINGS, 0x1, 0);
+      req.end("hello");
+
+      const rst = await raw.waitFor(f => f.streamId === 1 && f.type === FrameType.RST_STREAM);
+      await closed;
+      expect(rst.payload.readUInt32BE(0)).toBe(ErrorCode.NO_ERROR);
+      expect(events).toEqual(["wantTrailers", "finish"]);
+      expect(req.sentTrailers).toBeUndefined();
+      expect(raw.frames.filter(f => f.streamId === 1).map(frameSummary)).toEqual([
+        { type: FrameType.HEADERS, flags: 0x4 /* END_HEADERS */, length: expect.any(Number) },
+        { type: FrameType.DATA, flags: 0, length: 5 },
+        { type: FrameType.RST_STREAM, flags: 0, length: 4 },
+      ]);
+    } finally {
+      client.destroy();
+      raw.close();
+    }
+  });
+
+  test("server: a 'wantTrailers' listener that close()s the stream ends it with RST_STREAM(NO_ERROR) alone", async () => {
+    const outcome = Promise.withResolvers<{ events: string[]; sentTrailers: unknown }>();
+    const srv = http2.createServer();
+    srv.on("stream", (stream: any) => {
+      const events: string[] = [];
+      stream.on("error", (err: Error) => outcome.reject(err));
+      stream.on("wantTrailers", () => {
+        events.push("wantTrailers");
+        stream.close();
+      });
+      stream.on("finish", () => events.push("finish"));
+      stream.on("close", () => outcome.resolve({ events, sentTrailers: stream.sentTrailers }));
+      stream.respond({ ":status": 200 }, { waitForTrailers: true });
+      stream.end("ok");
+    });
+    srv.listen(0, "127.0.0.1");
+    await once(srv, "listening");
+    const c = await RawH2.connect((srv.address() as net.AddressInfo).port);
+    try {
+      c.sendPreface();
+      c.sendEmptySettings();
+      c.sendFrame(FrameType.HEADERS, 0x4 /* END_HEADERS */, 1, requestHeaderBlock("POST"));
+      const rst = await c.waitFor(f => f.streamId === 1 && f.type === FrameType.RST_STREAM);
+      expect(rst.payload.readUInt32BE(0)).toBe(ErrorCode.NO_ERROR);
+      expect(await outcome.promise).toEqual({ events: ["wantTrailers", "finish"], sentTrailers: undefined });
+      expect(c.frames.filter(f => f.streamId === 1).map(frameSummary)).toEqual([
+        { type: FrameType.HEADERS, flags: 0x4 /* END_HEADERS */, length: expect.any(Number) },
+        { type: FrameType.DATA, flags: 0, length: 2 },
+        { type: FrameType.RST_STREAM, flags: 0, length: 4 },
+      ]);
+    } finally {
+      c.destroy();
+      srv.close();
+    }
+  });
+});
+
 describe("request header and body framing (RFC 9113 §8.1)", () => {
   let deferredServer: http2.Http2Server;
   let deferredPort: number;