Skip to content
Open
Show file tree
Hide file tree
Changes from 3 commits
Commits
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
71 changes: 60 additions & 11 deletions src/runtime/api/bun/h2/connection.rs
Original file line number Diff line number Diff line change
Expand Up @@ -150,6 +150,46 @@ pub struct Feed {
/// behind a non-reading peer before the session is treated as flooded (NGHTTP2_ERR_FLOODED).
const MAX_OUTBOUND_ACK_QUEUE: u32 = 1000;

/// nghttp2's NGHTTP2_DEFAULT_GLITCH_BURST and NGHTTP2_DEFAULT_GLITCH_RATE.
const GLITCH_BURST: u32 = 10_000;
const GLITCH_RATE: u32 = 330;

/// nghttp2_ratelim: `rate` tokens per whole elapsed second, capped at `burst`.
struct RateLimit {
burst: u32,
rate: u32,
tokens: u32,
last_refill: std::time::Instant,
}

impl RateLimit {
fn new(burst: u32, rate: u32) -> Self {
RateLimit {
burst,
rate,
tokens: burst,
last_refill: std::time::Instant::now(),
}
}

/// Returns false once the bucket is empty.
fn drain(&mut self) -> bool {
let elapsed = self.last_refill.elapsed().as_secs();
if elapsed > 0 {
let gain = u32::try_from(elapsed)
.unwrap_or(u32::MAX)
.saturating_mul(self.rate);
self.tokens = self.tokens.saturating_add(gain).min(self.burst);
self.last_refill += std::time::Duration::from_secs(elapsed);
}
if self.tokens == 0 {
return false;
}
self.tokens -= 1;
true
}
}

/// What the connection engine calls back into the embedder (the JSC binding) for. Methods take
/// `&self`: the JSC binding (H2FrameParser) is fully interior-mutable (Cell/JsCell) and its host
/// functions receive `&Self`, so it can own the `Connection` and pass itself as the sink without an
Expand Down Expand Up @@ -301,6 +341,8 @@ pub struct Connection {
/// obq_flood_counter_). Reset only via note_outbound_drained() when the
/// embedder confirms its outbound buffer emptied — never per receive().
obq_ack_pending: u32,
/// nghttp2's glitch_ratelim.
glitch_limit: RateLimit,

/// Scratch buffer for the outbound HPACK-encoded header block.
enc_buf: Vec<u8>,
Expand Down Expand Up @@ -339,6 +381,7 @@ impl Connection {
data_in_flight: None,
terminated: false,
obq_ack_pending: 0,
glitch_limit: RateLimit::new(GLITCH_BURST, GLITCH_RATE),
enc_buf: Vec::new(),
replenish_buf: Vec::new(),
evict_buf: Vec::new(),
Expand Down Expand Up @@ -657,10 +700,8 @@ impl Connection {
Some(FrameType::PushPromise) => self.handle_push_promise(sink, hdr, payload),
Some(FrameType::AltSvc) => self.handle_altsvc(sink, hdr, payload),
Some(FrameType::Origin) => self.handle_origin(sink, hdr, payload),
// PRIORITY has no scheduling effect here; structurally validated above, otherwise ignored.
Some(FrameType::Priority) => false,
// §4.1: unknown frame types are silently discarded.
_ => false,
// Both are ignored (§4.1 for unknown types). nghttp2 charges each one as a glitch.
Some(FrameType::Priority) | None => self.note_glitch(sink),
}
}

Expand Down Expand Up @@ -797,6 +838,15 @@ impl Connection {
true
}

/// nghttp2's session_update_glitch_ratelim. Returns true when the session was torn down.
fn note_glitch(&mut self, sink: &impl Sink) -> bool {
if self.glitch_limit.drain() {
return false;
}
self.send_go_away(sink, ErrorCode::EnhanceYourCalm, b"too many ignored frames");
true
}

fn handle_ping(&mut self, sink: &impl Sink, hdr: &FrameHeader, payload: &[u8]) -> bool {
if wire::flags::has(hdr.flags, wire::flags::ACK) {
sink.on_ping(payload, true);
Expand Down Expand Up @@ -1760,6 +1810,9 @@ impl Connection {

/// RFC 7838 §4 ALTSVC: optional 2-byte origin-length + origin, then the Alt-Svc field value.
fn handle_altsvc(&mut self, sink: &impl Sink, hdr: &FrameHeader, payload: &[u8]) -> bool {
if self.is_server {
return self.note_glitch(sink);
}
if payload.len() < 2 {
return false; // malformed ALTSVC is ignored (§4)
}
Expand All @@ -1769,12 +1822,8 @@ impl Connection {
}
let origin = &payload[2..2 + origin_len];
let value = &payload[2 + origin_len..];
// RFC 7838 4 MUST-ignore rules: a server never accepts ALTSVC; on stream 0 the origin
// must be present; on a request stream it must be empty (the stream's own origin applies).
if self.is_server
|| (hdr.stream_id == 0 && origin.is_empty())
|| (hdr.stream_id != 0 && !origin.is_empty())
{
// RFC 7838 §4: the origin must be present on stream 0 and empty on a request stream.
if (hdr.stream_id == 0 && origin.is_empty()) || (hdr.stream_id != 0 && !origin.is_empty()) {
return false;
}
sink.on_altsvc(hdr.stream_id, origin, value);
Expand All @@ -1787,7 +1836,7 @@ impl Connection {
// only - a server receiving it must ignore it. The whole payload is delivered once; the
// embedder iterates the (2-byte length, origin) entries and surfaces a single event.
if hdr.stream_id != 0 || self.is_server {
return false;
return self.note_glitch(sink);
}
sink.on_origin(payload);
false
Expand Down
95 changes: 95 additions & 0 deletions test/js/node/http2/h2-conformance.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -1928,3 +1928,98 @@ describe("stream release after a queued END_STREAM", () => {
}
});
});

// nghttp2 keeps one token bucket per session for frames a peer has no reason to send often: 10,000
// tokens, 330 back per second. A PRIORITY frame, a frame of an unknown type, and an ALTSVC or ORIGIN
// frame sent to a server each take one. With none left the session ends with
// GOAWAY(ENHANCE_YOUR_CALM), which node reports as ERR_HTTP2_ERROR "Protocol error".
describe("floods of ignored frames (nghttp2's glitch rate limit)", () => {
const BURST = 10_000;
const PRIORITY = encodeFrame(FrameType.PRIORITY, 0, 1001, Buffer.from([0, 0, 0, 0, 16]));
const UNKNOWN = encodeFrame(0xfa, 0, 0, Buffer.from("12345"));
const ALTSVC = encodeFrame(0x0a, 0, 0, Buffer.concat([Buffer.from([0, 1]), Buffer.from('xh2=":1"')]));
const ORIGIN = encodeFrame(0x0c, 0, 0, Buffer.concat([Buffer.from([0, 1]), Buffer.from("x")]));
const PROTOCOL_ERROR_SESSION = { code: "ERR_HTTP2_ERROR", message: "Protocol error" };
const pingFrame = (id: number) => encodeFrame(FrameType.PING, 0, 0, Buffer.alloc(8, id));
const isPingAck = (id: number) => (f: Frame) =>
f.type === FrameType.PING && (f.flags & 0x1) !== 0 && f.payload[0] === id;
const isFatal = (f: Frame) => f.type === FrameType.RST_STREAM || f.type === FrameType.GOAWAY;

/** `count` frames that cycle through `kinds`, so each kind makes up an equal share. */
const flood = (kinds: Buffer[], count: number) =>
Buffer.concat(Array.from({ length: count }, (_, i) => kinds[i % kinds.length]));

async function listenRaw(server: http2.Http2Server): Promise<RawH2> {
server.listen(0);
await once(server, "listening");
const c = await RawH2.connect((server.address() as net.AddressInfo).port);
c.sendPreface();
c.sendEmptySettings();
return c;
}

// BURST frames are always tolerated: the bucket starts with that many tokens. BURST more empty it
// unless they take 30 seconds to get here. With two kinds, both have to be charged for that.
test.each([
["PRIORITY", [PRIORITY]],
["unknown frame types", [UNKNOWN]],
["ALTSVC and ORIGIN", [ALTSVC, ORIGIN]],
])("server: %s", async (_, kinds) => {
let sessionError: any;
const server = http2.createServer();
server.on("sessionError", e => (sessionError = e));
const c = await listenRaw(server);
try {
c.send(Buffer.concat([flood(kinds, BURST), pingFrame(1)]));
await c.waitFor(isPingAck(1), 4_000);
expect(c.frames.filter(isFatal)).toEqual([]);
c.send(flood(kinds, BURST));
expect(goawayErrorCode(await c.waitForGoaway(4_000))).toBe(ErrorCode.ENHANCE_YOUR_CALM);
await c.waitClosed(4_000);
expect({ code: sessionError?.code, message: sessionError?.message }).toEqual(PROTOCOL_ERROR_SESSION);
} finally {
c.destroy();
server.close();
}
});

test("client: PRIORITY and unknown frame types", async () => {
const kinds = [PRIORITY, UNKNOWN];
const raw = await RawH2Server.listen();
const client = http2.connect(`http://127.0.0.1:${raw.port}`);
const sessionError = once(client, "error");
try {
await raw.waitFor(f => f.type === FrameType.SETTINGS);
raw.sendFrame(FrameType.SETTINGS, 0, 0);
raw.sendFrame(FrameType.SETTINGS, 0x1, 0);
raw.socket!.write(Buffer.concat([flood(kinds, BURST), pingFrame(1)]));
await raw.waitFor(isPingAck(1), 4_000);
expect(raw.frames.filter(isFatal)).toEqual([]);
raw.socket!.write(flood(kinds, BURST));
expect(goawayErrorCode(await raw.waitFor(f => f.type === FrameType.GOAWAY, 4_000))).toBe(
ErrorCode.ENHANCE_YOUR_CALM,
);
const [err] = await sessionError;
expect({ code: err.code, message: err.message }).toEqual(PROTOCOL_ERROR_SESSION);
} finally {
client.destroy();
raw.close();
}
});

test("an empty bucket gets 330 tokens back per second", async () => {
const server = http2.createServer();
const c = await listenRaw(server);
try {
c.send(Buffer.concat([flood([PRIORITY], BURST), pingFrame(1)]));
await c.waitFor(isPingAck(1), 4_000);
await Bun.sleep(1100); // the refill under test is per whole second of wall clock
c.send(Buffer.concat([flood([PRIORITY], 300), pingFrame(2)]));
await c.waitFor(isPingAck(2), 4_000);
expect(c.frames.filter(isFatal)).toEqual([]);
} finally {
c.destroy();
server.close();
}
});
});
Loading