Skip to content

h2 server: decode and discard HEADERS on a closed client stream instead of resetting or re-opening it #37985

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
alii wants to merge 1 commit into
base: main
Choose a base branch
from
Open

h2 server: decode and discard HEADERS on a closed client stream instead of resetting or re-opening it #37985

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
72 changes: 45 additions & 27 deletions src/runtime/api/bun/h2/connection.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -134,7 +134,9 @@ struct HeaderBlockMeta {
#[derive(Clone, Copy, PartialEq, Eq)]
enum BlockDisposition {
Deliver,
/// HEADERS arrived on a closed stream: answered with RST_STREAM(STREAM_CLOSED).
/// HEADERS arrived on a closed stream (a `Closed` entry, or a peer id at or below the
/// high-water mark with no entry): decoded for HPACK and silently discarded (§5.1) — no
/// frame out, no sink callbacks, no state change.
StreamClosed,
/// The embedder refused the stream (can_open_stream = false, node's maxSessionMemory):
/// answered with RST_STREAM(ENHANCE_YOUR_CALM).
Expand Down Expand Up @@ -304,7 +306,15 @@ pub struct Connection {
evict_buf: Vec<u32>,

preface_received: usize,
/// Highest stream id seen in either direction (inbound HEADERS/PUSH_PROMISE, locally sent
/// header blocks and pushes). Used for GOAWAY and the RST_STREAM idle check.
pub last_stream_id: u32,
/// Highest peer-initiated stream id (§5.1.1): on a server, the highest odd id an inbound
/// HEADERS has opened, refused streams included; on a client, the highest promised id. A
/// peer id at or below it with no `streams` entry is closed, never idle (nghttp2's
/// last_recv_stream_id). Unlike `last_stream_id` it is never raised by local streams, so a
/// server that pushed even ids still recognises a lower odd id as new.
pub last_peer_stream_id: u32,
pub going_away: bool,
}

Expand Down Expand Up @@ -338,6 +348,7 @@ impl Connection {
evict_buf: Vec::new(),
preface_received: 0,
last_stream_id: 0,
last_peer_stream_id: 0,
going_away: false,
}
}
Expand Down Expand Up @@ -586,9 +597,9 @@ impl Connection {
self.replenish_buf = buf;
// Evict closed streams so the map (and this scan) stay bounded on long-lived connections.
// A late DATA/RST/WINDOW_UPDATE for an evicted id takes the unknown-stream path, which
// answers RST_STREAM(STREAM_CLOSED) - the 5.1 closed-state behavior. A late HEADERS for an
// evicted id re-opens a fresh entry (the parity check still applies); that matches how
// trailers-after-close are treated as a new block by the legacy parser as well.
// answers RST_STREAM(STREAM_CLOSED). On a server a late HEADERS for an evicted client id
// is still recognised as closed (id <= `last_peer_stream_id`) and is decoded then
// discarded per §5.1; on a client it re-opens a fresh entry.
let mut evict = std::mem::take(&mut self.evict_buf);
evict.clear();
for (id, s) in self.streams.iter() {
Expand Down Expand Up @@ -929,9 +940,9 @@ impl Connection {
let recv_init = self.local_settings.initial_window_size;
let is_new = !self.streams.contains_key(&hdr.stream_id);
// RFC 9113 5.1.1: client-initiated streams use odd ids - a server receiving HEADERS that
// would open an even-id stream is a connection PROTOCOL_ERROR. (Monotonicity is not
// checked here: a client legitimately receives HEADERS on even promised ids that are
// numerically below its own latest odd id.)
// would open an even-id stream is a connection PROTOCOL_ERROR. (Monotonicity is only
// applied server-side, below: a client legitimately receives HEADERS on even promised ids
// that are numerically below its own latest odd id.)
if is_new && self.is_server && hdr.stream_id.is_multiple_of(2) {
self.send_go_away(
sink,
Expand All @@ -940,13 +951,20 @@ impl Connection {
);
return true;
}
// §5.1.1: an unknown client id at or below the highest one already opened is a closed
// stream (evicted, or skipped and implicitly closed), not a new one: §5.1 minimal
// processing — decode the block for HPACK sync, discard it, open/refuse/count nothing.
let closed = is_new && self.is_server && hdr.stream_id <= self.last_peer_stream_id;
let is_new = is_new && !closed;
let refused = is_new && self.is_server && !sink.can_open_stream();
let mut disposition = if refused {
let mut disposition = if closed {
BlockDisposition::StreamClosed
} else if refused {
BlockDisposition::Refused
} else {
BlockDisposition::Deliver
};
if !refused {
if disposition == BlockDisposition::Deliver {
let cur_state = self
.streams
.entry(hdr.stream_id)
Expand Down Expand Up @@ -975,11 +993,10 @@ impl Connection {
// nghttp2 (session_on_*_headers_received): HEADERS for a stream whose remote
// half already ended (half-closed (remote)) is escalated to a CONNECTION error
// of type STREAM_CLOSED — node surfaces it as NghttpError "Stream was already
// closed or invalid" and tears the session down. A stream that closed for any
// other reason (e.g. we reset it and the peer's trailers were already in
// flight) keeps the conservative stream-level handling: the block is still
// decoded for HPACK sync (§4.3), then refused with RST_STREAM(STREAM_CLOSED)
// by finish_header_block.
// closed or invalid" and tears the session down. A fully closed stream (e.g.
// reset while the peer's trailers were in flight) gets §5.1's minimal
// processing instead: finish_header_block decodes the block for HPACK sync
// and discards it without a frame or callback.
if cur_state == State::HalfClosedRemote {
self.local_connection_error(
sink,
Expand All @@ -1000,6 +1017,11 @@ impl Connection {
if hdr.stream_id > self.last_stream_id {
self.last_stream_id = hdr.stream_id;
}
// Only a server's inbound HEADERS opens a peer stream (odd, checked above); a client's
// "new" HEADERS is a response on its own stream. Client peer ids come via PUSH_PROMISE.
if self.is_server && hdr.stream_id > self.last_peer_stream_id {
self.last_peer_stream_id = hdr.stream_id;
}
if !refused {
sink.on_stream_open(hdr.stream_id);
}
Expand Down Expand Up @@ -1244,16 +1266,8 @@ impl Connection {
sink.on_stream_rejected(target);
return false;
}
BlockDisposition::StreamClosed => {
// §5.1: HEADERS on a closed/half-closed-remote stream is a stream error of type
// STREAM_CLOSED. The block was decoded above purely for HPACK-table sync.
self.send_rst_stream(sink, target, ErrorCode::StreamClosed);
if let Some(s) = self.streams.get_mut(&target) {
s.state = State::Closed;
}
sink.on_stream_reset(target, ErrorCode::StreamClosed.as_u32());
return false;
}
// §5.1 "closed": minimally processed (HPACK sync above) and discarded, as nghttp2 does.
BlockDisposition::StreamClosed => return false,
BlockDisposition::Deliver => {}
}
// RFC 9113 §8.3.1 (nghttp2_http_on_request_headers): a request block needs exactly one
Expand Down Expand Up @@ -1735,6 +1749,9 @@ impl Connection {
if promised > self.last_stream_id {
self.last_stream_id = promised;
}
if promised > self.last_peer_stream_id {
self.last_peer_stream_id = promised;
}

self.header_block.clear();
self.header_block.extend_from_slice(&payload[off..end]);
Expand Down Expand Up @@ -1923,9 +1940,10 @@ impl Connection {
/// The outbound half does not run through this engine yet, so without this hook a
/// completed request's entry would linger as HalfClosedRemote forever — the map (and
/// the per-batch replenish/evict scans) would grow by one entry per request. Removal
/// has the same observable behavior as scan-eviction of a Closed stream: late frames
/// for the id take the unknown-stream path (RST STREAM_CLOSED, the §5.1 closed-state
/// answer) and a late HEADERS re-opens a fresh entry.
/// has the same observable behavior as scan-eviction of a Closed stream: late
/// DATA/RST/WINDOW_UPDATE take the unknown-stream path (RST STREAM_CLOSED); a late HEADERS
/// is decoded then discarded as a closed-stream block on a server (id <=
/// `last_peer_stream_id`, never re-opened) and re-opens a fresh entry on a client.
pub fn close_stream(&mut self, stream_id: u32) {
self.streams.remove(&stream_id);
}
Expand Down
178 changes: 178 additions & 0 deletions test/js/node/http2/h2-conformance.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1299,6 +1299,184 @@ describe("inbound stream lifecycle", () => {
}
});

/** A server answering every stream 200 "ok" that records each stream reaching JS. */
async function recordingSession() {
const seen: { id: number; sync?: string }[] = [];
const server = http2.createServer();
server.on("stream", (stream: any, headers: any) => {
seen.push({ id: stream.id, sync: headers["x-bun-sync"] });
stream.on("error", () => {});
try {
stream.respond({ ":status": 200 });
stream.end("ok");
} catch {}
});
server.listen(0);
await once(server, "listening");
const c = await RawH2.connect((server.address() as net.AddressInfo).port);
c.sendPreface();
c.sendEmptySettings();
return { server, c, seen };
}
Comment thread
coderabbitai[bot] marked this conversation as resolved.

/** PING as a barrier: once its ACK arrives the server has processed every frame written
* before it, so the next write lands in a later read. A GOAWAY fails fast instead. */
async function pingBarrier(c: RawH2, tag: number) {
const opaque = Buffer.alloc(8, tag);
c.sendFrame(FrameType.PING, 0, 0, opaque);
const frame = await c.waitFor(
f =>
f.type === FrameType.GOAWAY || (f.type === FrameType.PING && (f.flags & 0x1) === 1 && f.payload.equals(opaque)),
);
expect(frame.type).toBe(FrameType.PING);
}

/** A request block split across HEADERS(END_STREAM) + CONTINUATION whose CONTINUATION half
* inserts `x-bun-sync: 1` into the HPACK dynamic table (index 62). */
function splitRequestBlock(id: number): Buffer {
const insert = Buffer.concat([Buffer.from([0x40]), hpackLiteral("x-bun-sync"), hpackLiteral("1")]);
return Buffer.concat([
encodeFrame(FrameType.HEADERS, 0x1 /* END_STREAM, no END_HEADERS */, id, requestHeaderBlock("GET")),
encodeFrame(FrameType.CONTINUATION, 0x4 /* END_HEADERS */, id, insert),
]);
}

/** After splitRequestBlock() was sent on a closed stream: it must have been decoded (a request
* on `probeId` referencing index 62 is answered, no GOAWAY) yet never surfaced as a request. */
async function expectDecodedButDiscarded(c: RawH2, seen: { id: number; sync?: string }[], probeId: number) {
c.sendFrame(FrameType.HEADERS, 0x5, probeId, Buffer.concat([requestHeaderBlock("GET"), Buffer.from([0xbe])]));
const resp = await c.waitFor(
f => (f.type === FrameType.HEADERS && f.streamId === probeId) || f.type === FrameType.GOAWAY,
);
expect(resp.type).toBe(FrameType.HEADERS);
await pingBarrier(c, 0xff);
expect(c.frames.find(f => f.type === FrameType.GOAWAY)).toBeUndefined();
expect(seen.filter(s => s.sync !== undefined)).toEqual([{ id: probeId, sync: "1" }]);
}

function headersOrRstOn(c: RawH2, id: number, since = 0) {
return c.frames
.slice(since)
.filter(f => f.streamId === id && (f.type === FrameType.HEADERS || f.type === FrameType.RST_STREAM));
}

// RFC 9113 §5.1 "closed": HEADERS on a closed stream is minimally processed (HPACK state
// updated) and discarded — nothing is sent on that stream, nothing reaches JS, no GOAWAY.
test("discards HEADERS following the peer's RST_STREAM in the same write", async () => {
const { server, c, seen } = await recordingSession();
try {
const cancel = Buffer.alloc(4);
cancel.writeUInt32BE(ErrorCode.CANCEL, 0);
c.send(
Buffer.concat([
encodeFrame(FrameType.HEADERS, 0x4 /* END_HEADERS */, 1, requestHeaderBlock("GET")),
encodeFrame(FrameType.RST_STREAM, 0, 1, cancel),
splitRequestBlock(1),
]),
);
await expectDecodedButDiscarded(c, seen, 3);
expect(c.frames.filter(f => f.type === FrameType.RST_STREAM && f.streamId === 1)).toEqual([]);
} finally {
c.destroy();
server.close();
}
});

test("discards HEADERS arriving in a later read for a stream the peer reset", async () => {
const { server, c, seen } = await recordingSession();
try {
const cancel = Buffer.alloc(4);
cancel.writeUInt32BE(ErrorCode.CANCEL, 0);
c.send(
Buffer.concat([
encodeFrame(FrameType.HEADERS, 0x4 /* END_HEADERS */, 1, requestHeaderBlock("GET")),
encodeFrame(FrameType.RST_STREAM, 0, 1, cancel),
]),
);
await pingBarrier(c, 1);
const since = c.frames.length;
c.send(splitRequestBlock(1));
await expectDecodedButDiscarded(c, seen, 3);
expect(headersOrRstOn(c, 1, since)).toEqual([]);
} finally {
c.destroy();
server.close();
}
});

test("discards HEADERS reusing the id of a fully answered stream", async () => {
const { server, c, seen } = await recordingSession();
try {
c.sendFrame(FrameType.HEADERS, 0x5, 1, requestHeaderBlock("GET"));
await c.waitFor(f => f.type === FrameType.DATA && f.streamId === 1 && (f.flags & 0x1) === 1);
await pingBarrier(c, 1);
const since = c.frames.length;
c.send(splitRequestBlock(1));
await expectDecodedButDiscarded(c, seen, 3);
expect(headersOrRstOn(c, 1, since)).toEqual([]);
} finally {
c.destroy();
server.close();
}
});

// §5.1.1: opening stream 5 implicitly closed the never-used stream 3.
test("discards HEADERS on a lower stream id than one already opened", async () => {
const { server, c, seen } = await recordingSession();
try {
c.sendFrame(FrameType.HEADERS, 0x5, 5, requestHeaderBlock("GET"));
await c.waitFor(f => f.type === FrameType.HEADERS && f.streamId === 5);
await pingBarrier(c, 1);
c.send(splitRequestBlock(3));
await expectDecodedButDiscarded(c, seen, 7);
expect(c.frames.filter(f => f.streamId === 3)).toEqual([]);
} finally {
c.destroy();
server.close();
}
});

// The closed-vs-new line is the highest CLIENT stream id: even ids the server promised above it
// (pushes 2, 4, 6 off stream 1) must not make a first HEADERS on stream 3 look closed.
test("a client stream id below the server's promised ids is still new", async () => {
const seen: number[] = [];
const server = http2.createServer();
server.on("stream", (stream: any) => {
seen.push(stream.id);
stream.on("error", () => {});
const pushes = stream.id === 1 ? 3 : 0;
for (let i = 0; i < pushes; i++) {
stream.pushStream({ ":path": `/push/${i}` }, (err: any, push: any) => {
if (err) return;
push.on("error", () => {});
push.respond({ ":status": 200 });
push.end("p");
});
}
stream.respond({ ":status": 200 });
stream.end("ok");
});
server.listen(0);
await once(server, "listening");
const c = await RawH2.connect((server.address() as net.AddressInfo).port);
try {
c.sendPreface();
c.sendEmptySettings(); // ENABLE_PUSH defaults to 1
c.sendFrame(FrameType.HEADERS, 0x5, 1, requestHeaderBlock("GET"));
// Three PUSH_PROMISEs on stream 1 reserve 2, 4 and 6.
await c.waitFor(f => f.type === FrameType.PUSH_PROMISE && f.streamId === 1 && f.payload.readUInt32BE(0) === 6);
c.sendFrame(FrameType.HEADERS, 0x5, 3, requestHeaderBlock("GET"));
const resp = await c.waitFor(
f => (f.type === FrameType.HEADERS && f.streamId === 3) || f.type === FrameType.GOAWAY,
);
expect(resp.type).toBe(FrameType.HEADERS);
expect(seen).toEqual([1, 3]);
} finally {
c.destroy();
server.close();
}
});

// A header-value `toString` runs user JS while sendTrailers holds the native `&mut Stream`;
// feeding the stream's own RST_STREAM (then another read) back into the parser from that
// callback must not free the Stream out from under the caller (use-after-free under ASAN).
Expand Down
Loading