fix(node, fastify): answer HTTP/2 HEAD requests

packages/fastify/src/response.test.ts

@@ -1,4 +1,5 @@
 import { Buffer } from 'node:buffer'
+import http2 from 'node:http2'
 import { Readable, Writable } from 'node:stream'
 import FastifyCookie from '@fastify/cookie'
 import * as StandardServerNode from '@standard-server/node'
@@ -590,4 +591,148 @@ describe('sendStandardResponse', () => {
       expect(reply.send).not.toHaveBeenCalled()
     })
   })
+
+  describe('head request with a stream body', () => {
+    it('releases the body on an auto-exposed HEAD route', async ({ onTestFinished }) => {
+      const { body, isReleased } = createEndlessBody()
+      let sending: Promise<void> | undefined
+
+      const fastify = Fastify()
+      onTestFinished(() => fastify.close())
+
+      fastify.get('/', async (req, reply) => {
+        sending = sendStandardResponse(reply, { status: 201, headers: { 'x-custom-header': 'custom-value' }, body })
+        await sending
+      })
+
+      await fastify.ready()
+      const res = await request(fastify.server).head('/')
+
+      await expect(sending).resolves.toBeUndefined()
+
+      expect(res.status).toBe(201)
+      expect(res.headers).toMatchObject({ 'x-custom-header': 'custom-value' })
+
+      await vi.waitFor(() => {
+        expect(isReleased()).toBe(true)
+      })
+    })
+
+    it('releases the body on an auto-exposed http2 HEAD route', async ({ onTestFinished }) => {
+      const { body, isReleased } = createEndlessBody()
+      let sending: Promise<void> | undefined
+
+      const fastify = Fastify({ http2: true })
+      onTestFinished(() => fastify.close())
+
+      fastify.get('/', async (req, reply) => {
+        sending = sendStandardResponse(reply, { status: 201, headers: { 'x-custom-header': 'custom-value' }, body })
+        await sending
+      })
+
+      const response = await requestHttp2Head(await fastify.listen({ port: 0, host: '127.0.0.1' }))
+
+      await expect(sending).resolves.toBeUndefined()
+
+      expect(response.headers).toMatchObject({ ':status': 201, 'x-custom-header': 'custom-value' })
+      expect(response.body).toBe('')
+
+      await vi.waitFor(() => {
+        expect(isReleased()).toBe(true)
+      })
+    })
+
+    it('rejects and releases the body on an explicit http2 HEAD route', async ({ onTestFinished }) => {
+      const { body, isReleased } = createEndlessBody()
+      let sending: Promise<void> | undefined
+
+      const fastify = Fastify({ http2: true })
+      onTestFinished(() => fastify.close())
+
+      fastify.head('/', async (req, reply) => {
+        sending = sendStandardResponse(reply, { status: 201, headers: { 'x-custom-header': 'custom-value' }, body })
+        await sending.catch(() => {})
+      })
+
+      const response = await requestHttp2Head(await fastify.listen({ port: 0, host: '127.0.0.1' }))
+
+      await expect(sending).rejects.toMatchObject({ code: 'ERR_STREAM_WRITE_AFTER_END' })
+
+      expect(response.headers).toMatchObject({ ':status': 201, 'x-custom-header': 'custom-value' })
+      expect(response.body).toBe('')
+
+      await vi.waitFor(() => {
+        expect(isReleased()).toBe(true)
+      })
+    })
+  })
+
+  it('rejects with the stream error when the client resets an http2 stream with an error code', async ({ onTestFinished }) => {
+    const { body, isReleased } = createEndlessBody()
+    let sending: Promise<void> | undefined
+
+    const fastify = Fastify({ http2: true })
+    onTestFinished(() => fastify.close())
+
+    fastify.get('/', async (req, reply) => {
+      sending = sendStandardResponse(reply, { status: 200, headers: {}, body })
+      await sending.catch(() => {})
+    })
+
+    const client = http2.connect(await fastify.listen({ port: 0, host: '127.0.0.1' }))
+    onTestFinished(() => client.close())
+
+    const reqStream = client.request({ ':path': '/' })
+    reqStream.once('error', () => {})
+
+    await new Promise(r => reqStream.once('data', r))
+    reqStream.close(http2.constants.NGHTTP2_INTERNAL_ERROR)
+
+    await expect(sending).rejects.toMatchObject({ code: 'ERR_HTTP2_STREAM_ERROR' })
+
+    await vi.waitFor(() => {
+      expect(isReleased()).toBe(true)
+    })
+  })
 })
+
+function createEndlessBody() {
+  let released = false
+
+  const body = (async function* () {
+    try {
+      while (true) {
+        yield 'foo'
+        await new Promise(r => setTimeout(r, 10))
+      }
+    }
+    finally {
+      released = true
+    }
+  })()
+
+  return { body, isReleased: () => released }
+}
+
+async function requestHttp2Head(origin: string): Promise<{ headers: http2.IncomingHttpHeaders, body: string }> {
+  const client = http2.connect(origin)
+
+  try {
+    const reqStream = client.request({ ':path': '/', ':method': 'HEAD' })
+
+    const chunks: Buffer[] = []
+    reqStream.on('data', (chunk: Buffer) => chunks.push(chunk))
+
+    const headers = await new Promise<http2.IncomingHttpHeaders>((resolve, reject) => {
+      reqStream.once('response', resolve)
+      reqStream.once('error', reject)
+    })
+
+    await new Promise(r => reqStream.once('close', r))
+
+    return { headers, body: Buffer.concat(chunks).toString() }
+  }
+  finally {
+    client.close()
+  }
+}

packages/fastify/src/response.ts

@@ -32,8 +32,16 @@ export async function sendStandardResponse(
       return
     }
 
-    reply.raw.once('error', reject)
-    reply.raw.once('close', resolve)
+    const connection = 'stream' in reply.raw ? reply.raw.stream : reply.raw
+
+    connection.once('error', reject)
+    connection.once('close', () => {
+      if (typeof resBody === 'object' && !resBody.closed) {
+        resBody.destroy()
+      }
+
+      resolve()
+    })
 
     try {
       reply.status(standardResponse.status)
@@ -46,7 +54,7 @@ export async function sendStandardResponse(
         }
       }
 
-      // fastify pipes and cleans up the stream body itself, no manual piping needed
+      // fastify pipes the stream body itself, no manual piping needed
       reply.send(resBody)
     }
     catch (error) {

packages/node/src/response.test.ts

@@ -1,6 +1,9 @@
 import type { StandardResponse } from '@standard-server/core'
 import type { IncomingMessage, ServerResponse } from 'node:http'
+import type { Http2ServerRequest, Http2ServerResponse } from 'node:http2'
+import type { AddressInfo } from 'node:net'
 import { Buffer } from 'node:buffer'
+import http2 from 'node:http2'
 import Stream from 'node:stream'
 import request from 'supertest'
 import * as Body from './body'
@@ -498,4 +501,141 @@ describe('sendStandardResponse', () => {
       expect((responseStream as any).setHeader).not.toHaveBeenCalled()
     })
   })
+
+  describe('http2 HEAD request', () => {
+    it('sends the status and headers', async () => {
+      let sending: Promise<void> | undefined
+
+      const response = await requestHttp2Head((_, res) => {
+        sending = sendStandardResponse(res, {
+          status: 201,
+          headers: { 'x-custom-header': 'custom-value' },
+          body: undefined,
+        })
+      })
+
+      await expect(sending).resolves.toBeUndefined()
+
+      expect(response.headers).toMatchObject({
+        ':status': 201,
+        'x-custom-header': 'custom-value',
+      })
+      expect(response.body).toBe('')
+    })
+
+    it('rejects and releases a stream body it cannot send', async () => {
+      let clean = false
+      const body = (async function* () {
+        try {
+          while (true) {
+            yield 'foo'
+            await new Promise(r => setTimeout(r, 10))
+          }
+        }
+        finally {
+          clean = true
+        }
+      })()
+
+      let sending: Promise<void> | undefined
+
+      const response = await requestHttp2Head((_, res) => {
+        sending = sendStandardResponse(res, { status: 201, headers: {}, body })
+        sending.catch(() => {})
+      })
+
+      await expect(sending).rejects.toMatchObject({ code: 'ERR_STREAM_WRITE_AFTER_END' })
+
+      expect(response.headers).toMatchObject({
+        ':status': 201,
+        'content-type': 'text/event-stream',
+      })
+      expect(response.body).toBe('')
+
+      await vi.waitFor(() => {
+        expect(clean).toBe(true)
+      })
+    })
+  })
+
+  describe('http2 stream reset by the client', () => {
+    it.each([
+      ['resolves on a cancel', http2.constants.NGHTTP2_CANCEL, undefined],
+      ['rejects with the stream error on an error code', http2.constants.NGHTTP2_INTERNAL_ERROR, 'ERR_HTTP2_STREAM_ERROR'],
+    ])('%s', async (_, code, errorCode) => {
+      let clean = false
+      const body = (async function* () {
+        try {
+          while (true) {
+            yield 'foo'
+            await new Promise(r => setTimeout(r, 10))
+          }
+        }
+        finally {
+          clean = true
+        }
+      })()
+
+      let sending: Promise<void> | undefined
+
+      const server = http2.createServer((req, res) => {
+        sending = sendStandardResponse(res, { status: 200, headers: {}, body })
+      })
+      await new Promise<void>(r => server.listen(0, r))
+
+      const client = http2.connect(`http://localhost:${(server.address() as AddressInfo).port}`)
+
+      try {
+        const reqStream = client.request({ ':path': '/' })
+        reqStream.once('error', () => {})
+
+        await new Promise(r => reqStream.once('data', r))
+        reqStream.close(code)
+
+        if (errorCode) {
+          await expect(sending).rejects.toMatchObject({ code: errorCode })
+        }
+        else {
+          await expect(sending).resolves.toBeUndefined()
+        }
+
+        await vi.waitFor(() => {
+          expect(clean).toBe(true)
+        })
+      }
+      finally {
+        client.close()
+        await new Promise(r => server.close(r))
+      }
+    })
+  })
 })
+
+async function requestHttp2Head(
+  listener: (req: Http2ServerRequest, res: Http2ServerResponse) => void,
+): Promise<{ headers: http2.IncomingHttpHeaders, body: string }> {
+  const server = http2.createServer(listener)
+  await new Promise<void>(r => server.listen(0, r))
+
+  const client = http2.connect(`http://localhost:${(server.address() as AddressInfo).port}`)
+
+  try {
+    const reqStream = client.request({ ':path': '/', ':method': 'HEAD' })
+
+    const chunks: Buffer[] = []
+    reqStream.on('data', (chunk: Buffer) => chunks.push(chunk))
+
+    const headers = await new Promise<http2.IncomingHttpHeaders>((resolve, reject) => {
+      reqStream.once('response', resolve)
+      reqStream.once('error', reject)
+    })
+
+    await new Promise(r => reqStream.once('close', r))
+
+    return { headers, body: Buffer.concat(chunks).toString() }
+  }
+  finally {
+    client.close()
+    await new Promise(r => server.close(r))
+  }
+}

packages/node/src/response.ts

@@ -33,8 +33,10 @@ export async function sendStandardResponse(
       return
     }
 
-    res.once('error', reject)
-    res.once('close', resolve)
+    const connection = 'stream' in res ? res.stream : res
+
+    connection.once('error', reject)
+    connection.once('close', resolve)
 
     try {
       // DON'T use `res.writeHead` because it send response immediately in chunked mode
@@ -55,7 +57,7 @@ export async function sendStandardResponse(
         res.end(resBody)
       }
       else {
-        res.once('close', () => {
+        connection.once('close', () => {
           if (!resBody.closed) {
             resBody.destroy(getNodeResponseError(res) ?? undefined)
           }