fix: throw on retry when payload is consumed

lib/handler/retry-handler.js

@@ -68,6 +68,7 @@ class RetryHandler {
     this.end = null
     this.etag = null
     this.resume = null
+    this.consumed = false
 
     // Handle possible onConnect duplication
     this.handler.onConnect(reason => {
@@ -193,6 +194,19 @@ class RetryHandler {
       this.resume = null
 
       if (statusCode !== 206) {
+        // Abort when status code is not partial content
+        // and payload is already consumed because downstream
+        // will concatenate response from two request wrongly.
+        if (this.consumed) {
+          this.abort(
+            new RequestRetryError('unsupported retry when payload is consumed', statusCode, {
+              headers,
+              data: { count: this.retryCount }
+            })
+          )
+          return false
+        }
+
         return true
       }
 
@@ -294,6 +308,7 @@ class RetryHandler {
   }
 
   onData (chunk) {
+    this.consumed = true
     this.start += chunk.length
 
     return this.handler.onData(chunk)

test/issue-3356.js

@@ -0,0 +1,57 @@
+'use strict'
+
+const { tspl } = require('@matteo.collina/tspl')
+const { test, after } = require('node:test')
+const { createServer } = require('node:http')
+const { once } = require('node:events')
+
+const { fetch, Agent, RetryAgent } = require('..')
+
+test('https://github.com/nodejs/undici/issues/3356', async (t) => {
+  t = tspl(t, { plan: 3 })
+
+  let shouldRetry = true
+  const server = createServer()
+  server.on('request', (req, res) => {
+    res.writeHead(200, { 'content-type': 'text/plain' })
+    if (shouldRetry) {
+      shouldRetry = false
+
+      res.flushHeaders()
+      res.write('h')
+      setTimeout(() => { res.end('ello world!') }, 100)
+    } else {
+      res.end('hello world!')
+    }
+  })
+
+  server.listen(0)
+
+  await once(server, 'listening')
+
+  after(async () => {
+    server.close()
+
+    await once(server, 'close')
+  })
+
+  const agent = new RetryAgent(new Agent({ bodyTimeout: 50 }), {
+    errorCodes: ['UND_ERR_BODY_TIMEOUT']
+  })
+
+  const response = await fetch(`http://localhost:${server.address().port}`, {
+    dispatcher: agent
+  })
+  setTimeout(async () => {
+    try {
+      t.equal(response.status, 200)
+      // consume response
+      await response.text()
+    } catch (err) {
+      t.equal(err.name, 'TypeError')
+      t.equal(err.cause.code, 'UND_ERR_REQ_RETRY')
+    }
+  }, 200)
+
+  await t.completed
+})