http2-goaway.js

'use strict'

const { tspl } = require('@matteo.collina/tspl')
const { test, after } = require('node:test')
const { constants, createSecureServer } = require('node:http2')
const { once, EventEmitter } = require('node:events')

const pem = require('@metcoder95/https-pem')

const { Agent, Client, Pool, fetch } = require('..')

test('#3046 - GOAWAY Frame', async t => {
  t = tspl(t, { plan: 8 })

  const server = createSecureServer(await pem.generate({ opts: { keySize: 2048 } }))

  server.on('stream', (stream, headers) => {
    setTimeout(() => {
      if (stream.closed) return
      stream.end('Hello World')
    }, 100)

    stream.respond({
      'content-type': 'text/plain; charset=utf-8',
      'x-custom-h2': 'hello',
      ':status': 200
    })
  })

  server.on('session', session => {
    setTimeout(() => {
      session.goaway()
    }, 50)
  })

  after(() => server.close())
  await once(server.listen(0), 'listening')

  const client = new Client(`https://localhost:${server.address().port}`, {
    connect: {
      rejectUnauthorized: false
    },
    allowH2: true
  })
  after(() => client.close())

  client.on('connectionError', () => {
    t.fail('unexpected connectionError')
  })

  client.on('disconnect', (url, disconnectClient, err) => {
    t.ok(url instanceof URL)
    t.deepStrictEqual(disconnectClient, [client])
    t.strictEqual(err.message, 'HTTP/2: "GOAWAY" frame received with code 0')
    t.strictEqual(err.code, 'UND_ERR_INFO')
  })

  const response = await client.request({
    path: '/',
    method: 'GET',
    headers: {
      'x-my-header': 'foo'
    }
  })

  t.strictEqual(response.headers['content-type'], 'text/plain; charset=utf-8')
  t.strictEqual(response.headers['x-custom-h2'], 'hello')
  t.strictEqual(response.statusCode, 200)
  t.strictEqual(await response.body.text(), 'Hello World')

  await t.completed
})

test('#3753 - Handle GOAWAY Gracefully', async (t) => {
  t = tspl(t, { plan: 30 })

  const server = createSecureServer(await pem.generate({ opts: { keySize: 2048 } }))
  let counter = 0
  let session = null

  server.on('session', s => {
    session = s
  })

  server.on('stream', (stream) => {
    counter++

    // Due to the nature of the test, we need to ignore the error
    // that is thrown when the session is destroyed and stream
    // is in-flight
    stream.on('error', () => {})
    if (counter === 9 && session != null) {
      session.goaway()
    }

    stream.respond({
      'content-type': 'text/plain',
      ':status': 200
    })
    setTimeout(() => {
      if (stream.closed) return
      stream.end('hello world')
    }, 150)
  })

  after(() => server.close())
  await once(server.listen(0), 'listening')

  const client = new Client(`https://localhost:${server.address().port}`, {
    connect: {
      rejectUnauthorized: false
    },
    pipelining: 2,
    allowH2: true
  })
  after(() => client.close())

  for (let i = 0; i < 15; i++) {
    client.request({
      path: '/',
      method: 'GET',
      headers: {
        'x-my-header': 'foo'
      }
    }, (err, response) => {
      if (err) {
        t.strictEqual(err.message, 'HTTP/2: "GOAWAY" frame received with code 0')
        t.strictEqual(err.code, 'UND_ERR_INFO')
      } else {
        t.strictEqual(response.statusCode, 200)
        ;(async function () {
          let body
          try {
            body = await response.body.text()
          } catch (err) {
            t.fail(err)
            return
          }
          t.strictEqual(body, 'hello world')
        })()
      }
    })
  }

  await t.completed
})

test('#5468 - requeue multiplexed requests after mid-burst GOAWAY', async (t) => {
  t = tspl(t, { plan: 4 })

  const server = createSecureServer(await pem.generate({ opts: { keySize: 2048 } }))
  let firstStream = false
  let streams = 0

  server.on('stream', (stream) => {
    stream.on('error', () => {})
    streams++

    if (!firstStream) {
      firstStream = true
      stream.session?.goaway(constants.NGHTTP2_INTERNAL_ERROR, 0)
      return
    }

    stream.respond({ ':status': 200 })
    stream.end('ok')
  })

  after(() => server.close())
  await once(server.listen(0), 'listening')

  const pool = new Pool(`https://localhost:${server.address().port}`, {
    allowH2: true,
    connections: 10,
    connect: {
      rejectUnauthorized: false
    }
  })
  after(async () => pool.close())

  const requests = 40
  const codes = {}
  let ok = 0

  await Promise.all(Array.from({ length: requests }, async () => {
    try {
      const response = await pool.request({ path: '/', method: 'GET' })
      await response.body.dump()

      if (response.statusCode === 200) {
        ok++
      }
    } catch (err) {
      const code = err?.code || err?.cause?.code || err?.name || 'unknown'
      codes[code] = (codes[code] || 0) + 1
    }
  }))

  t.strictEqual(codes.UND_ERR_INFO, undefined)
  t.ok((codes.ERR_HTTP2_SESSION_ERROR || 0) <= 2)
  t.ok(ok >= requests - 2)
  t.ok(streams >= requests - 1)

  await t.completed
})

test('Agent keeps a Pool with requests requeued after GOAWAY', async (t) => {
  const server = createSecureServer(await pem.generate({ opts: { keySize: 2048 } }))
  let firstSession = null
  const sessionCounts = new Map()

  server.on('session', (session) => {
    sessionCounts.set(session, 0)
    firstSession ??= session
  })

  server.on('stream', (stream) => {
    stream.on('error', () => {})

    const count = sessionCounts.get(stream.session) + 1
    sessionCounts.set(stream.session, count)

    // The first request warms the session. Accept two requests from the burst,
    // then refuse the remaining streams so the Client has to requeue them.
    if (stream.session === firstSession && count === 3) {
      firstSession.goaway(constants.NGHTTP2_NO_ERROR, stream.id)
    }

    stream.respond({ ':status': 200 })
    setTimeout(() => {
      if (!stream.destroyed) {
        stream.end('ok')
      }
    }, 150)
  })

  await once(server.listen(0), 'listening')

  let pool = null
  const agent = new Agent({
    factory (origin) {
      pool = new Pool(origin, {
        connections: 1,
        connect: {
          rejectUnauthorized: false
        }
      })
      return pool
    }
  })
  const disconnectErrors = []
  agent.on('disconnect', (_origin, _targets, err) => {
    disconnectErrors.push(err)
  })

  t.after(async () => {
    agent.removeAllListeners('disconnect')
    await agent.close()
    server.close()
    await once(server, 'close')
  })

  const url = `https://localhost:${server.address().port}`
  const warmup = await fetch(url, { dispatcher: agent })
  await warmup.text()

  const disconnected = once(agent, 'disconnect')
  const requests = Array.from({ length: 6 }, async () => {
    const response = await fetch(url, {
      dispatcher: agent,
      signal: AbortSignal.timeout(3000)
    })

    return `${response.status}:${await response.text()}`
  })

  await disconnected
  const poolWasClosed = pool.closed
  const results = await Promise.allSettled(requests)
  await new Promise(resolve => setTimeout(resolve, 100))

  // GOAWAY detaches the session before the refused requests are replayed. The
  // Pool must remain open while those requests are pending; otherwise its
  // replacement Client is destroyed as soon as the replay finishes.
  t.assert.deepStrictEqual(results, Array(6).fill(null).map(() => ({
    status: 'fulfilled',
    value: '200:ok'
  })))
  t.assert.deepStrictEqual(disconnectErrors.map(err => err.code), ['UND_ERR_INFO'])
  t.assert.strictEqual(poolWasClosed, false)
})

test('#5089 - Handle GOAWAY Gracefully', async (t) => {
  t = tspl(t, { plan: 7 })

  const server = createSecureServer({
    ...await pem.generate({
      opts: { keySize: 2048 }
    }),
    settings: {
      maxConcurrentStreams: 2
    }
  })
  let firstSession = null
  const sessionCounts = new Map()

  server.on('session', (session) => {
    sessionCounts.set(session, 0)

    if (firstSession == null) {
      firstSession = session
    }
  })

  server.on('stream', (stream) => {
    stream.on('error', () => {})

    const count = sessionCounts.get(stream.session) + 1
    sessionCounts.set(stream.session, count)

    if (stream.session === firstSession && count === 2) {
      setTimeout(() => {
        firstSession.goaway()
      }, 20)
    }

    stream.respond({
      'content-type': 'text/plain',
      ':status': 200
    })
    setTimeout(() => {
      stream.end('hello world')
    }, 150)
  })

  after(() => server.close())
  await once(server.listen(0), 'listening')

  const client = new Client(`https://localhost:${server.address().port}`, {
    connect: {
      rejectUnauthorized: false
    },
    maxConcurrentStreams: 2,
    pipelining: 10,
    allowH2: true
  })
  after(() => client.close())

  client.on('connectionError', () => {
    t.fail('unexpected connectionError')
  })

  client.once('disconnect', (url, disconnectClient, err) => {
    t.ok(url instanceof URL)
    t.deepStrictEqual(disconnectClient, [client])
    t.strictEqual(err.message, 'HTTP/2: "GOAWAY" frame received with code 0')
    t.strictEqual(err.code, 'UND_ERR_INFO')
  })

  const results = await Promise.all(Array.from({ length: 6 }, () => client.request({
    path: '/',
    method: 'GET',
    headers: {
      'x-my-header': 'foo'
    }
  }).then(async (response) => {
    return {
      statusCode: response.statusCode,
      body: await response.body.text()
    }
  })))

  t.deepStrictEqual([...sessionCounts.values()], [2, 4])
  t.deepStrictEqual(results.map((result) => result.statusCode), Array(6).fill(200))
  t.deepStrictEqual(results.map((result) => result.body), Array(6).fill('hello world'))

  await t.completed
})

test('#5453 - request onto a GOAWAY-rejected session is requeued on a fresh session', async (t) => {
  t = tspl(t, { plan: 5 })

  const http2 = require('node:http2')
  const originalConnect = http2.connect

  // A session that has received GOAWAY rejects new streams synchronously.
  class FakeSession extends EventEmitter {
    constructor () {
      super()
      this.closed = false
      this.destroyed = false
    }

    request () {
      const err = new Error('New streams cannot be created after receiving a GOAWAY')
      err.code = 'ERR_HTTP2_GOAWAY_SESSION'
      throw err
    }

    destroy () {
      if (this.destroyed) return
      this.destroyed = true
      this.emit('close')
    }

    ref () {}
    unref () {}
  }

  const session = new FakeSession()
  let connectCalls = 0
  let streams = 0

  http2.connect = function connectStub (...args) {
    connectCalls++
    return connectCalls === 1 ? session : originalConnect.apply(this, args)
  }
  after(() => { http2.connect = originalConnect })

  const server = createSecureServer(await pem.generate({ opts: { keySize: 2048 } }))
  server.on('stream', (stream) => {
    streams++
    stream.respond({ ':status': 200 })
    stream.end('ok')
  })
  after(() => server.close())
  await once(server.listen(0), 'listening')

  const client = new Client(`https://localhost:${server.address().port}`, {
    allowH2: true,
    connect: { rejectUnauthorized: false }
  })
  after(() => client.close())

  const response = await client.request({ path: '/', method: 'GET' })
  const chunks = []
  response.body.on('data', chunk => chunks.push(chunk))
  await once(response.body, 'end')

  t.strictEqual(response.statusCode, 200)
  t.strictEqual(Buffer.concat(chunks).toString(), 'ok')
  t.strictEqual(streams, 1)
  t.strictEqual(connectCalls, 2)
  t.strictEqual(session.destroyed, true)

  await t.completed
})

test('a request whose response already started fails instead of being replayed after GOAWAY', { timeout: 10000 }, async (t) => {
  const p = tspl(t, { plan: 4 })

  const server = createSecureServer(await pem.generate({ opts: { keySize: 2048 } }))
  let secondStream = null
  let secondRequests = 0

  server.on('stream', (stream, headers) => {
    stream.on('error', () => {})

    if (headers[':path'] === '/second') {
      secondRequests++
      secondStream = stream
      // Respond without ending, so the GOAWAY below arrives mid-response.
      stream.respond({ ':status': 200 })
      return
    }

    stream.respond({ ':status': 200 })
    stream.end('ok')
  })

  t.after(() => server.close())
  await once(server.listen(0), 'listening')

  const client = new Client(`https://localhost:${server.address().port}`, {
    allowH2: true,
    connect: { rejectUnauthorized: false }
  })
  t.after(() => client.close())

  // Stream 1 completes, so the GOAWAY below can name a lower stream id.
  const first = await client.request({ path: '/first', method: 'GET' })
  p.strictEqual(await first.body.text(), 'ok')

  // Stream 3 has received its response headers when this resolves.
  const second = await client.request({ path: '/second', method: 'GET' })

  // A misbehaving peer names stream 1 as the last processed, even though it
  // already responded on stream 3. undici must fail the request with the
  // GOAWAY error rather than replay a response it has already started.
  secondStream.session.goaway(constants.NGHTTP2_NO_ERROR, 1)

  await p.rejects(second.body.text(), { code: 'UND_ERR_INFO' })
  p.strictEqual(secondRequests, 1)

  // After the GOAWAY, a new request still succeeds on a fresh connection.
  const third = await client.request({ path: '/third', method: 'GET' })
  p.strictEqual(await third.body.text(), 'ok')

  await p.completed
})