Skip to content
17 changes: 14 additions & 3 deletions src/runtime/api/bun/h2/connection.rs
Original file line number Diff line number Diff line change
Expand Up @@ -355,6 +355,8 @@ pub(crate) struct Connection {

preface_received: usize,
pub last_stream_id: u32,
/// Highest peer-initiated stream id (§5.1.1); never raised by local streams.
pub last_peer_stream_id: u32,
pub going_away: bool,
}

Expand Down Expand Up @@ -394,6 +396,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 @@ -1833,9 +1836,14 @@ impl Connection {
payload[off + 3],
]) & 0x7fff_ffff;
off += 4;
// §5.1.1 / §8.4: server-initiated streams use even ids, never 0, and
// cannot be reused.
if promised == 0 || promised & 1 == 1 || self.streams.contains_key(&promised) {
// §5.1.1 / §8.4: server-initiated streams use even ids, never 0, and each new one is
// numbered above every earlier promise, including promises whose streams have since been
// closed and evicted (nghttp2 fails the session for these too).
Comment thread
robobun marked this conversation as resolved.
if promised == 0
|| promised & 1 == 1
|| promised <= self.last_peer_stream_id
|| self.streams.contains_key(&promised)
{
self.send_go_away(
sink,
ErrorCode::ProtocolError,
Comment thread
robobun marked this conversation as resolved.
Expand All @@ -1855,6 +1863,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;
}
Comment thread
robobun marked this conversation as resolved.

self.header_block.clear();
self.header_block.extend_from_slice(&payload[off..end]);
Expand Down
155 changes: 155 additions & 0 deletions test/js/node/http2/h2-conformance.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -800,6 +800,161 @@ describe("push stream states (checklist §5.1, RFC 9113 §6.4/§8.4)", () => {
});
});

describe.concurrent("promised stream ids (RFC 9113 §5.1.1)", () => {
Comment thread
robobun marked this conversation as resolved.
/** A raw server with a connected `http2.connect()` client that has `requests` requests in
* flight (streams 1, 3, ...) and the SETTINGS exchange done. Pushed streams delivered to the
* client are collected in `pushedIds`. */
async function clientWithRawServer(requests = 1) {
const raw = await RawH2Server.listen();
const client = http2.connect(`http://127.0.0.1:${raw.port}`);
const errors: Error[] = [];
client.on("error", err => errors.push(err));
const sessionClosed = new Promise<void>(resolve => client.once("close", resolve));
const pushedIds: number[] = [];
client.on("stream", (pushed: http2.ClientHttp2Stream) => {
pushedIds.push(pushed.id!);
pushed.on("error", () => {});
});
const reqs: http2.ClientHttp2Stream[] = [];
for (let i = 0; i < requests; i++) {
const req = client.request({ ":path": `/${i}` });
req.on("error", () => {});
reqs.push(req);
}
await raw.waitFor(f => f.type === FrameType.HEADERS && f.streamId === 2 * requests - 1);
Comment thread
robobun marked this conversation as resolved.
raw.sendFrame(FrameType.SETTINGS, 0, 0);
raw.sendFrame(FrameType.SETTINGS, 0x1, 0);
/** PUSH_PROMISE on stream 1 reserving `promisedId`. */
function sendPushPromise(promisedId: number) {
const promised = Buffer.alloc(4);
promised.writeUInt32BE(promisedId, 0);
raw.sendFrame(
FrameType.PUSH_PROMISE,
0x4 /* END_HEADERS */,
1,
Buffer.concat([promised, requestHeaderBlock("GET")]),
);
}
/** A legal PUSH_PROMISE: resolves with the pushed stream once the client has delivered it. */
async function pushPromise(promisedId: number): Promise<http2.ClientHttp2Stream> {
const delivered = once(client, "stream");
sendPushPromise(promisedId);
const [pushed] = await delivered;
return pushed;
}
/** RST_STREAM(CANCEL) for a pushed stream; resolves once the client has closed it. */
async function resetPushed(pushed: http2.ClientHttp2Stream) {
const closed = new Promise<void>(resolve => pushed.once("close", resolve));
const cancel = Buffer.alloc(4);
cancel.writeUInt32BE(ErrorCode.CANCEL, 0);
raw.sendFrame(FrameType.RST_STREAM, 0, pushed.id!, cancel);
await closed;
}
/** The client failed the session: a GOAWAY went out and the session was destroyed with
* nghttp2's NGHTTP2_ERR_PROTO. The GOAWAY's error code is not pinned: bun writes
* PROTOCOL_ERROR, node writes INTERNAL_ERROR because its `session.destroy(err)` runs first. */
async function expectSessionProtocolError() {
await raw.waitFor(f => f.type === FrameType.GOAWAY);
await sessionClosed;
expect(
errors.map(err => ({
code: (err as NodeJS.ErrnoException).code,
errno: (err as NodeJS.ErrnoException).errno,
message: err.message,
})),
).toEqual([{ code: "ERR_HTTP2_ERROR", errno: -505, message: "Protocol error" }]);
expect(client.destroyed).toBe(true);
}
let pings = 0;
/** PING round trip: the client has finished processing everything sent before it, and the
* connection is still alive (no GOAWAY, no session error). */
async function expectAlive() {
const opaque = Buffer.alloc(8);
opaque.writeUInt32BE(++pings, 4);
raw.sendFrame(FrameType.PING, 0, 0, opaque);
await raw.waitFor(f => f.type === FrameType.PING && (f.flags & 0x1) !== 0 && f.payload.equals(opaque));
expect(raw.frames.find(f => f.type === FrameType.GOAWAY)).toBeUndefined();
expect(errors).toEqual([]);
}
function close() {
client.destroy();
raw.close();
}
return {
raw,
reqs,
pushedIds,
sendPushPromise,
pushPromise,
resetPushed,
expectSessionProtocolError,
expectAlive,
close,
};
}

test("a PUSH_PROMISE whose promised id does not exceed the previous one fails the session", async () => {
const peer = await clientWithRawServer();
try {
await peer.pushPromise(4);
peer.sendPushPromise(2);
await peer.expectSessionProtocolError();
expect(peer.pushedIds).toEqual([4]);
} finally {
peer.close();
}
});

test("a PUSH_PROMISE reusing the id of a pushed stream the server already reset fails the session", async () => {
const peer = await clientWithRawServer();
try {
await peer.resetPushed(await peer.pushPromise(2));
// Stream 2 is closed on the client and its state released, but the id stays used up.
await peer.expectAlive();
peer.sendPushPromise(2);
await peer.expectSessionProtocolError();
expect(peer.pushedIds).toEqual([2]);
} finally {
peer.close();
}
});

test("a higher promised id is accepted after an earlier pushed stream was reset", async () => {
const peer = await clientWithRawServer();
try {
await peer.resetPushed(await peer.pushPromise(2));
await peer.expectAlive();
await peer.pushPromise(4);
await peer.expectAlive();
expect(peer.pushedIds).toEqual([2, 4]);
} finally {
peer.close();
}
});

// The mark a promised id is checked against only counts the server's ids: the client's own
// requests are numbered above the push here (the response on stream 3 has already been
// received), and that must not make promised id 2 look reused.
test("a promised id below the client's own stream ids is accepted", async () => {
const peer = await clientWithRawServer(2);
try {
const response = once(peer.reqs[1], "response");
peer.raw.sendFrame(
FrameType.HEADERS,
0x5 /* END_STREAM | END_HEADERS */,
3,
Buffer.from([0x88]) /* :status 200 */,
);
await response;
await peer.pushPromise(2);
await peer.expectAlive();
expect(peer.pushedIds).toEqual([2]);
} finally {
peer.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
Expand Down
Loading