fix: correctly peek only first maxBuffer bytes in createPeekStream

Refactor createPeekStream into createLazyTransform:
- Use Transform instead of Duplex for a simpler, idiomatic interface
- Synchronous `choose(data)` replaces callback-based `swap(err, stream)`
- Buffer chunks in an array, concat only once when threshold is reached
- Default maxBuffer to 10 (the only value used in practice)
- Rename to createLazyTransform to better reflect its purpose

Author: hardik-kaji

index.js

@@ -2,15 +2,15 @@
 
 const zlib = require('node:zlib')
 const { inherits, format } = require('node:util')
-const { pipeline, compose, Duplex } = require('node:stream')
+const { pipeline, compose } = require('node:stream')
 
 const fp = require('fastify-plugin')
 const encodingNegotiator = require('@fastify/accept-negotiator')
 const mimedb = require('mime-db')
 const { Minipass } = require('minipass')
 const { Readable } = require('readable-stream')
 
-const { isStream, isGzip, isDeflate, intoAsyncIterator, isWebReadableStream, isFetchResponse, webStreamToNodeReadable } = require('./lib/utils')
+const { isStream, isGzip, isDeflate, intoAsyncIterator, isWebReadableStream, isFetchResponse, webStreamToNodeReadable, createLazyTransform } = require('./lib/utils')
 
 const InvalidRequestEncodingError = createError('FST_CP_ERR_INVALID_CONTENT_ENCODING', 'Unsupported Content-Encoding: %s', 415)
 const InvalidRequestCompressedPayloadError = createError('FST_CP_ERR_INVALID_CONTENT', 'Could not decompress the request payload using the provided encoding', 400)
@@ -555,75 +555,27 @@ function maybeUnzip (payload, serialize) {
   return Readable.from(intoAsyncIterator(result))
 }
 
-function createPeekStream (maxBuffer, onpeek) {
-  let buf = Buffer.alloc(0)
-  let dest = null
-
-  return new Duplex({
-    write (chunk, encoding, cb) {
-      if (dest) {
-        return dest.write(chunk, encoding, cb)
-      }
-
-      buf = Buffer.concat([buf, chunk])
-
-      if (buf.length < maxBuffer) return cb()
-
-      onpeek(buf, (err, stream) => {
-        if (err) return cb(err)
-        dest = stream
-        dest.on('data', (d) => this.push(d))
-        dest.on('end', () => this.push(null))
-        dest.write(buf, cb)
-        buf = null
-      })
-    },
-
-    final (cb) {
-      if (dest) {
-        dest.end(cb)
-        return
-      }
-
-      onpeek(buf, (err, stream) => {
-        if (err) return cb(err)
-        dest = stream
-        dest.on('data', (d) => this.push(d))
-        dest.on('end', () => { this.push(null); cb() })
-        if (buf && buf.length > 0) {
-          dest.end(buf)
-        } else {
-          dest.end()
-        }
-        buf = null
-      })
-    },
-
-    read () {}
-  })
-}
-
 function zipStream (deflate, encoding) {
-  return createPeekStream(10, function (data, swap) {
+  return createLazyTransform(function (data) {
     switch (isCompressed(data)) {
-      case 1: return swap(null, new Minipass())
-      case 2: return swap(null, new Minipass())
+      case 1: return new Minipass()
+      case 2: return new Minipass()
     }
-    return swap(null, deflate[encoding]())
+    return deflate[encoding]()
   })
 }
 
 function unzipStream (inflate, maxRecursion) {
   if (!(maxRecursion >= 0)) maxRecursion = 3
-  return createPeekStream(10, function (data, swap) {
+  return createLazyTransform(function (data) {
     // This path is never taken, when `maxRecursion` < 0 it is automatically set back to 3
     /* c8 ignore next */
-    if (maxRecursion < 0) return swap(new Error('Maximum recursion reached'))
+    if (maxRecursion < 0) throw new Error('Maximum recursion reached')
     switch (isCompressed(data)) {
-      case 1: return swap(null, compose(inflate.gzip(), unzipStream(inflate, maxRecursion - 1)))
-      case 2: return swap(null, compose(inflate.deflate(), unzipStream(inflate, maxRecursion - 1)))
+      case 1: return compose(inflate.gzip(), unzipStream(inflate, maxRecursion - 1))
+      case 2: return compose(inflate.deflate(), unzipStream(inflate, maxRecursion - 1))
     }
-    return swap(null, new Minipass())
+    return new Minipass()
   })
 }
 

lib/utils.js

@@ -1,6 +1,6 @@
 'use strict'
 
-const { Readable: NodeReadable } = require('node:stream')
+const { Readable: NodeReadable, Transform } = require('node:stream')
 
 // https://datatracker.ietf.org/doc/html/rfc8878#section-3.1.1
 function isZstd (buffer) {
@@ -104,4 +104,61 @@ async function * intoAsyncIterator (payload) {
   yield payload
 }
 
-module.exports = { isZstd, isGzip, isDeflate, isStream, intoAsyncIterator, isWebReadableStream, isFetchResponse, webStreamToNodeReadable }
+/**
+ * Creates a Transform that lazily selects an inner transform by peeking
+ * at the first `maxBuffer` bytes.
+ *
+ * @param {(data: Buffer) => import('stream').Transform} choose
+ *   Called once with the peeked bytes. Must return the transform stream
+ *   that all data (including the peeked bytes) will be piped through.
+ * @param {number} [maxBuffer=10] - Number of bytes to peek before deciding.
+ * @returns {import('stream').Transform}
+ */
+function createLazyTransform (choose, maxBuffer = 10) {
+  let chunks = []
+  let chunksLength = 0
+  let dest = null
+
+  return new Transform({
+    transform (chunk, encoding, cb) {
+      if (dest) {
+        return dest.write(chunk, encoding, cb)
+      }
+
+      chunks.push(chunk)
+      chunksLength += chunk.length
+
+      if (chunksLength < maxBuffer) return cb()
+
+      const buf = Buffer.concat(chunks)
+      chunks = null
+
+      dest = choose(buf.subarray(0, maxBuffer))
+      dest.on('data', (d) => this.push(d))
+      dest.on('end', () => this.push(null))
+      dest.write(buf, cb)
+    },
+
+    flush (cb) {
+      if (!dest) {
+        const buf = Buffer.concat(chunks)
+        chunks = null
+
+        dest = choose(buf)
+        dest.on('data', (d) => this.push(d))
+        dest.on('end', () => { this.push(null); cb() })
+        if (buf.length > 0) {
+          dest.end(buf)
+        } else {
+          dest.end()
+        }
+        return
+      }
+
+      dest.on('end', cb)
+      dest.end()
+    }
+  })
+}
+
+module.exports = { isZstd, isGzip, isDeflate, isStream, intoAsyncIterator, isWebReadableStream, isFetchResponse, webStreamToNodeReadable, createLazyTransform }

test/utils.test.js

@@ -3,8 +3,9 @@
 const { createReadStream } = require('node:fs')
 const { Socket } = require('node:net')
 const { Duplex, PassThrough, Readable, Stream, Transform, Writable } = require('node:stream')
+const { pipeline } = require('node:stream/promises')
 const { test } = require('node:test')
-const { isStream, isZstd, isDeflate, isGzip, intoAsyncIterator } = require('../lib/utils')
+const { isStream, isZstd, isDeflate, isGzip, intoAsyncIterator, createLazyTransform } = require('../lib/utils')
 
 test('isStream() utility should be able to detect Streams', async (t) => {
   t.plan(12)
@@ -124,3 +125,123 @@ test('intoAsyncIterator() utility should handle different data', async (t) => {
     equal(chunk, obj)
   }
 })
+
+test('createLazyTransform() should peek at first N bytes and pipe through selected stream', async (t) => {
+  const input = Buffer.from('hello world')
+  const chunks = []
+
+  const peekStream = createLazyTransform((data) => {
+    t.assert.equal(data.length, 5)
+    t.assert.equal(data.toString(), 'hello')
+    return new PassThrough()
+  }, 5)
+
+  peekStream.on('data', (chunk) => chunks.push(chunk))
+
+  await new Promise((resolve, reject) => {
+    peekStream.on('end', resolve)
+    peekStream.on('error', reject)
+    peekStream.write(input)
+    peekStream.end()
+  })
+
+  t.assert.equal(Buffer.concat(chunks).toString(), 'hello world')
+})
+
+test('createLazyTransform() should work when data arrives in small chunks', async (t) => {
+  const chunks = []
+
+  const peekStream = createLazyTransform((data) => {
+    t.assert.equal(data.length, 4)
+    t.assert.equal(data.toString(), 'abcd')
+    return new PassThrough()
+  }, 4)
+
+  peekStream.on('data', (chunk) => chunks.push(chunk))
+
+  await new Promise((resolve, reject) => {
+    peekStream.on('end', resolve)
+    peekStream.on('error', reject)
+    peekStream.write(Buffer.from('ab'))
+    peekStream.write(Buffer.from('cd'))
+    peekStream.write(Buffer.from('ef'))
+    peekStream.end()
+  })
+
+  t.assert.equal(Buffer.concat(chunks).toString(), 'abcdef')
+})
+
+test('createLazyTransform() should handle stream ending before maxBuffer is reached', async (t) => {
+  const chunks = []
+
+  const peekStream = createLazyTransform((data) => {
+    t.assert.equal(data.toString(), 'hi')
+    return new PassThrough()
+  })
+
+  peekStream.on('data', (chunk) => chunks.push(chunk))
+
+  await new Promise((resolve, reject) => {
+    peekStream.on('end', resolve)
+    peekStream.on('error', reject)
+    peekStream.write(Buffer.from('hi'))
+    peekStream.end()
+  })
+
+  t.assert.equal(Buffer.concat(chunks).toString(), 'hi')
+})
+
+test('createLazyTransform() should propagate errors from choose function', async (t) => {
+  const peekStream = createLazyTransform(() => {
+    throw new Error('test error')
+  }, 2)
+
+  await t.assert.rejects(
+    pipeline(Readable.from(Buffer.from('hello')), peekStream, new PassThrough()),
+    { message: 'test error' }
+  )
+})
+
+test('createLazyTransform() should handle empty stream', async (t) => {
+  const chunks = []
+
+  const peekStream = createLazyTransform((data) => {
+    t.assert.equal(data.length, 0)
+    return new PassThrough()
+  })
+
+  peekStream.on('data', (chunk) => chunks.push(chunk))
+
+  await new Promise((resolve, reject) => {
+    peekStream.on('end', resolve)
+    peekStream.on('error', reject)
+    peekStream.end()
+  })
+
+  t.assert.equal(Buffer.concat(chunks).length, 0)
+})
+
+test('createLazyTransform() should work with a transform stream as destination', async (t) => {
+  const chunks = []
+
+  const upper = new Transform({
+    transform (chunk, enc, cb) {
+      cb(null, chunk.toString().toUpperCase())
+    }
+  })
+
+  const peekStream = createLazyTransform(() => {
+    return upper
+  }, 3)
+
+  peekStream.on('data', (chunk) => chunks.push(chunk))
+
+  await new Promise((resolve, reject) => {
+    peekStream.on('end', resolve)
+    peekStream.on('error', reject)
+    peekStream.write(Buffer.from('hello'))
+    peekStream.end()
+  })
+
+  t.assert.equal(Buffer.concat(chunks).toString(), 'HELLO')
+})