fix(h2): honour maxResponseSize

lib/dispatcher/client-h2.js

@@ -10,7 +10,8 @@ const {
   InformationalError,
   InvalidArgumentError,
   HeadersTimeoutError,
-  BodyTimeoutError
+  BodyTimeoutError,
+  ResponseExceededMaxSizeError
 } = require('../core/errors.js')
 const {
   kUrl,
@@ -39,7 +40,8 @@ const {
   kRemoteSettings,
   kHTTP2Stream,
   kHTTP2SessionState,
-  kHTTP2Options
+  kHTTP2Options,
+  kMaxResponseSize
 } = require('../core/symbols.js')
 const { channels } = require('../core/diagnostics.js')
 
@@ -1022,9 +1024,11 @@ function writeH2 (client, request) {
   const state = {
     abort: null,
     body: request.body,
+    bytesRead: 0,
     client,
     contentLength: null,
     expectsPayload: false,
+    maxResponseSize: client[kMaxResponseSize],
     request,
     headersTimeout,
     bodyTimeout,
@@ -1322,12 +1326,22 @@ function onData (chunk) {
     return
   }
 
-  const { request } = state
+  const { request, maxResponseSize } = state
 
   if (request.aborted || request.completed) {
     return
   }
 
+  if (maxResponseSize > -1 && state.bytesRead + chunk.length > maxResponseSize) {
+    // Unlike HTTP/1.1, which destroys the socket because it cannot abandon one
+    // response without losing framing, resetting the offending stream leaves
+    // the session usable for its siblings.
+    state.abort(new ResponseExceededMaxSizeError())
+    return
+  }
+
+  state.bytesRead += chunk.length
+
   if (request.onResponseData(chunk) === false) {
     stream.pause()
   }

test/max-response-size.js

@@ -2,8 +2,10 @@
 
 const { tspl } = require('@matteo.collina/tspl')
 const { test, after, describe } = require('node:test')
-const { Client, errors } = require('..')
+const { once } = require('node:events')
+const { Client, H2CClient, errors } = require('..')
 const { createServer } = require('node:http')
+const { createServer: createH2CServer } = require('node:http2')
 
 describe('max response size', async (t) => {
   test('default max default size should allow all responses', async (t) => {
@@ -116,6 +118,85 @@ describe('max response size', async (t) => {
     await t.completed
   })
 
+  test('should throw an error if the response is too big over h2c', async (t) => {
+    t = tspl(t, { plan: 1 })
+
+    const server = createH2CServer()
+    after(() => {
+      server.close()
+    })
+
+    server.on('stream', (stream) => {
+      // the client resets the stream once the limit is passed
+      stream.on('error', () => {})
+      stream.respond({ ':status': 200 })
+      stream.end('hello')
+    })
+
+    server.listen(0)
+    await once(server, 'listening')
+
+    const client = new H2CClient(`http://localhost:${server.address().port}`, {
+      maxResponseSize: 1
+    })
+    after(() => client.destroy())
+
+    let error = null
+    try {
+      const { body } = await client.request({ path: '/', method: 'GET' })
+      await body.text()
+    } catch (err) {
+      error = err
+    }
+
+    t.ok(error instanceof errors.ResponseExceededMaxSizeError)
+
+    await t.completed
+  })
+
+  test('should keep the h2c session usable after a stream exceeds the limit', async (t) => {
+    t = tspl(t, { plan: 3 })
+
+    let connections = 0
+    const server = createH2CServer()
+    after(() => {
+      server.close()
+    })
+
+    server.on('connection', () => {
+      connections++
+    })
+    server.on('stream', (stream, headers) => {
+      stream.on('error', () => {})
+      stream.respond({ ':status': 200 })
+      stream.end(headers[':path'] === '/big' ? 'hello world' : 'ok')
+    })
+
+    server.listen(0)
+    await once(server, 'listening')
+
+    const client = new H2CClient(`http://localhost:${server.address().port}`, {
+      maxResponseSize: 2
+    })
+    after(() => client.destroy())
+
+    let error = null
+    try {
+      const { body } = await client.request({ path: '/big', method: 'GET' })
+      await body.text()
+    } catch (err) {
+      error = err
+    }
+
+    t.ok(error instanceof errors.ResponseExceededMaxSizeError)
+
+    const { body } = await client.request({ path: '/small', method: 'GET' })
+    t.strictEqual(await body.text(), 'ok')
+    t.strictEqual(connections, 1)
+
+    await t.completed
+  })
+
   test('invalid max response size should throw an error', async (t) => {
     t = tspl(t, { plan: 2 })