fix(readable): keep body bytes that arrive after setEncoding() (#5620)

* fix(readable): keep body bytes that arrive after setEncoding()

A response body was silently truncated when setEncoding() was used and
the body arrived in more than one chunk: .text(), .json(), .bytes(),
.arrayBuffer() and .blob() all returned short data and raised nothing.

setEncoding() installs a StringDecoder, so state.buffer holds decoded
strings from then on and the trailing bytes of a multi-byte sequence
split across a chunk boundary sit inside the decoder. #5003 worked
around that by snapshotting the raw chunks buffered at the time of the
call into kPreservedBuffer, but chunks arriving afterwards were never
added, and consumeStart() treated the two sources as mutually exclusive,
so everything but the snapshot was dropped.

Drop kPreservedBuffer and read the single source of truth instead:
state.buffer, re-encoded back to bytes by consumePush(), plus the bytes
the decoder is still holding. Nothing is retained beside the stream's
own buffer, and push() stays untouched on the hot path.

Fixes: #5611

* test: drive the setEncoding() regression test off events, not timers

The chunk that has to arrive after setEncoding() is now released by a
handshake with the server rather than a 50ms timer, and the wait for the
body to arrive is the client's 'drain' event rather than a 150ms sleep.

Author: mcollina

lib/api/readable.js

@@ -15,7 +15,6 @@ const kContentType = Symbol('kContentType')
 const kContentLength = Symbol('kContentLength')
 const kUsed = Symbol('kUsed')
 const kBytesRead = Symbol('kBytesRead')
-const kPreservedBuffer = Symbol('kPreservedBuffer')
 
 const noop = () => {}
 
@@ -326,36 +325,14 @@ class BodyReadable extends Readable {
    */
   setEncoding (encoding) {
     if (Buffer.isEncoding(encoding)) {
-      // Preserve raw Buffer chunks for the consume path (body.text(),
-      // body.json(), etc.) before super.setEncoding() replaces them
-      // with decoded strings. Without this, the consume path would
-      // lose access to the original bytes — some of which may be held
-      // by the decoder for incomplete multi-byte sequences, and the
-      // rest converted to strings that can't be safely concatenated
-      // byte-wise.
-      const state = this._readableState
-      const buffer = state.buffer
-      if (buffer && state.length > 0) {
-        const bufferIndex = state.bufferIndex ?? 0
-        const preserved = []
-        const source = typeof buffer.slice === 'function'
-          ? buffer.slice(bufferIndex)
-          : buffer
-        for (const data of source) {
-          if (Buffer.isBuffer(data)) {
-            preserved.push(data)
-          }
-        }
-        if (preserved.length > 0) {
-          this[kPreservedBuffer] = (this[kPreservedBuffer] || []).concat(preserved)
-        }
-      }
-
       // Delegate to Node.js Readable.setEncoding() which initializes a
       // StringDecoder and re-encodes already-buffered chunks. This properly
       // handles multi-byte sequences split at chunk boundaries for the
       // for-await / on('data') paths. Without this, Node.js uses
       // buf.toString(encoding) on each chunk, producing U+FFFD for split chars.
+      //
+      // The consume path (body.text(), body.json(), ...) copes with the
+      // decoded strings this leaves in state.buffer, see consumeStart().
       super.setEncoding(encoding)
     }
     return this
@@ -464,17 +441,7 @@ function consumeStart (consume) {
 
   const { _readableState: state } = consume.stream
 
-  // If setEncoding() was called, state.buffer may contain decoded strings
-  // (which would break Buffer.concat in chunksDecode). Use the preserved
-  // raw Buffers (saved before super.setEncoding() in setEncoding()) for
-  // byte-level accurate consumption. Otherwise read from state.buffer.
-  const preserved = consume.stream[kPreservedBuffer]
-  if (preserved && preserved.length > 0) {
-    for (const chunk of preserved) {
-      consumePush(consume, chunk)
-    }
-    consume.stream[kPreservedBuffer] = null
-  } else if (state.bufferIndex) {
+  if (state.bufferIndex) {
     const start = state.bufferIndex
     const end = state.buffer.length
     for (let n = start; n < end; n++) {
@@ -486,6 +453,16 @@ function consumeStart (consume) {
     }
   }
 
+  // If setEncoding() was called, state.buffer holds decoded strings, which
+  // consumePush() turns back into bytes. The trailing bytes of a multi-byte
+  // sequence split across a chunk boundary are not part of any of those
+  // strings, they are held inside the decoder until the rest arrives, so
+  // take them from there.
+  const decoder = state.decoder
+  if (decoder != null && decoder.lastNeed > 0) {
+    consumePush(consume, Buffer.from(decoder.lastChar.subarray(0, decoder.lastTotal - decoder.lastNeed)))
+  }
+
   if (state.endEmitted) {
     // No `this` to read the consume off here: consumeStart is a free function, called from
     // the queueMicrotask above. The callback below does have one, because the emitter passes
@@ -588,14 +565,22 @@ function consumeEnd (consume, encoding) {
 
 /**
  * @param {Consume} consume
- * @param {Buffer} chunk
+ * @param {Buffer|string} chunk
  * @returns {void}
  */
 function consumePush (consume, chunk) {
   if (consume.body === null) {
     return
   }
 
+  if (typeof chunk === 'string') {
+    // Buffered before the consume started, while an encoding was set.
+    // consume.length has to stay a byte count and chunksDecode()/chunksConcat()
+    // only work on bytes, so re-encode. A string's own length is in UTF-16 code
+    // units and Uint8Array.prototype.set() ignores a string argument entirely.
+    chunk = Buffer.from(chunk, consume.stream._readableState.encoding)
+  }
+
   consume.length += chunk.length
   consume.body.push(chunk)
 }

test/client-request.js

@@ -1574,6 +1574,52 @@ test('setEncoding(\'utf8\') handles 3-byte UTF-8 characters split across chunks'
   await t.completed
 })
 
+test('#5611 - setEncoding() then .text() does not truncate a body that arrives in several chunks', async (t) => {
+  t = tspl(t, { plan: 2 })
+
+  // '傳' is 3 bytes in UTF-8, split across the two chunks below.
+  const text = 'abc傳def'
+  const buf = Buffer.from(text)
+
+  // The rest of the body is released only once the client has called
+  // setEncoding(), which is what puts the second chunk on the wrong side of it.
+  const sendRest = new EE()
+
+  const server = createServer({ joinDuplicateHeaders: true }, async (req, res) => {
+    res.writeHead(200, { 'content-type': 'text/plain; charset=utf-8' })
+    res.write(buf.subarray(0, 4))
+    await EE.once(sendRest, 'go')
+    res.end(buf.subarray(4))
+  })
+  after(() => {
+    server.closeAllConnections?.()
+    server.close()
+  })
+
+  server.listen(0, async () => {
+    const client = new Client(`http://localhost:${server.address().port}`)
+    after(client.destroy.bind(client))
+
+    const { body } = await client.request({ path: '/', method: 'GET' })
+    body.setEncoding('utf8')
+
+    // 'drain' means the request has run to completion, so the whole body has
+    // arrived and is buffered as decoded strings, with the split character
+    // held inside the decoder. Waiting on the body itself is not an option:
+    // reading it would disturb it and make the consume below unusable.
+    const completed = EE.once(client, 'drain')
+    sendRest.emit('go')
+    await completed
+
+    const result = await body.text()
+
+    t.strictEqual(result, text)
+    t.strictEqual(Buffer.byteLength(result), buf.length)
+  })
+
+  await t.completed
+})
+
 test('#3736 - Aborted Response (without consuming body)', async (t) => {
   const plan = tspl(t, { plan: 1 })
 

test/readable.js

@@ -226,4 +226,136 @@ describe('Readable', () => {
 
     t.strictEqual(text, 'hello world')
   })
+
+  test('setEncoding() then .text() keeps chunks pushed after setEncoding()', async function (t) {
+    t = tspl(t, { plan: 2 })
+
+    function resume () {
+    }
+    function abort () {
+    }
+    const r = new Readable({ resume, abort })
+
+    // '傳' is 3 bytes in UTF-8, so cutting every 2 bytes splits each of them
+    // across a chunk boundary.
+    const expected = 'a傳b傳c傳d'
+    const buf = Buffer.from(expected)
+    const chunks = []
+    for (let n = 0; n < buf.length; n += 2) {
+      chunks.push(buf.subarray(n, n + 2))
+    }
+
+    // Buffered when setEncoding() runs: these are replaced by a single decoded
+    // string, with the tail of the split '傳' held inside the decoder.
+    r.push(chunks[0])
+    r.push(chunks[1])
+
+    r.setEncoding('utf8')
+
+    // Pushed after setEncoding() but before .text() is called: these are
+    // buffered as decoded strings, not as bytes.
+    r.push(chunks[2])
+    r.push(chunks[3])
+
+    const promise = r.text()
+
+    // Pushed after .text() but before the consume actually starts.
+    r.push(chunks[4])
+
+    setImmediate(() => {
+      // Pushed once the consume is running.
+      r.push(chunks[5])
+      r.push(chunks[6])
+      r.push(null)
+    })
+
+    const text = await promise
+
+    t.strictEqual(text, expected)
+    t.strictEqual(Buffer.byteLength(text), buf.length)
+  })
+
+  test('setEncoding() with only a partial character buffered', async function (t) {
+    t = tspl(t, { plan: 1 })
+
+    function resume () {
+    }
+    function abort () {
+    }
+    const r = new Readable({ resume, abort })
+
+    const buf = Buffer.from('傳')
+
+    // Decodes to the empty string, so setEncoding() leaves nothing buffered
+    // and the byte stays inside the decoder.
+    r.push(buf.subarray(0, 1))
+
+    r.setEncoding('utf8')
+
+    r.push(buf.subarray(1))
+
+    process.nextTick(() => {
+      r.push(null)
+    })
+
+    t.strictEqual(await r.text(), '傳')
+  })
+
+  for (const encoding of ['utf8', 'hex', 'base64', 'latin1']) {
+    test(`setEncoding('${encoding}') before any chunk arrives`, async function (t) {
+      t = tspl(t, { plan: 5 })
+
+      function resume () {
+      }
+      function abort () {
+      }
+
+      const buf = Buffer.from('hello 傳 world')
+
+      // Nothing is buffered when setEncoding() runs, so every chunk reaches
+      // state.buffer as a decoded string.
+      function body () {
+        const r = new Readable({ resume, abort })
+        r.setEncoding(encoding)
+        r.push(buf.subarray(0, 4))
+        r.push(buf.subarray(4, 8))
+        process.nextTick(() => {
+          r.push(buf.subarray(8))
+          r.push(null)
+        })
+        return r
+      }
+
+      t.deepStrictEqual(await body().bytes(), new Uint8Array(buf))
+      t.deepStrictEqual(new Uint8Array(await body().arrayBuffer()), new Uint8Array(buf))
+
+      const blob = await body().blob()
+      t.strictEqual(blob.size, buf.length)
+      t.deepStrictEqual(Buffer.from(await blob.arrayBuffer()), buf)
+
+      t.strictEqual(await body().text(), buf.toString(encoding))
+    })
+  }
+
+  test('setEncoding() then .json()', async function (t) {
+    t = tspl(t, { plan: 1 })
+
+    function resume () {
+    }
+    function abort () {
+    }
+    const r = new Readable({ resume, abort })
+
+    const buf = Buffer.from(JSON.stringify({ hello: '傳' }))
+
+    r.setEncoding('utf8')
+    r.push(buf.subarray(0, 12))
+
+    process.nextTick(() => {
+      r.push(buf.subarray(12))
+      r.push(null)
+    })
+
+    t.deepStrictEqual(await r.json(), { hello: '傳' })
+  })
 })