Repository: nodejs/undici

Problem statement:
Retry Handler Fails When Request Has a Body
## Bug Description

When a body is provided to a request, such as in a `PUT` request, the retry handler fails with `UND_ERR_REQ_CONTENT_LENGTH_MISMATCH`.

## Reproducible By

repro.ts:

```
import http from 'http';
import { Agent, RetryAgent } from 'undici';

const myDispatcher = new RetryAgent(new Agent());

async function failListener(req, res) {
  res.setHeader('content-type', 'application/json');
  res.writeHead(500);
  res.end(`{"message": "failure"}`);
}

async function makeRetryRequest() {
  const result = await fetch('http://localhost:3333/fail', {
    method: 'PUT',
    body: JSON.stringify({ foo: 'bar', baz: 'bat' }),
    headers: {
      'content-type': 'application/json',
    },
    dispatcher: myDispatcher,
  });
  console.log(result);
}

async function main() {
  http.createServer(failListener).listen(3333, makeRetryRequest);
}

main();
```

```
npx tsx repro.ts
```

## Expected Behavior

The request should be retried and should not fail because `undici` fails to handle re-use of the `body` correctly.

## Logs & Screenshots

```
> npx tsx repro.ts
node:internal/deps/undici/undici:11372
    Error.captureStackTrace(err, this);
          ^

TypeError: fetch failed
    at Object.fetch (node:internal/deps/undici/undici:11372:11)
    at Server.makeRetryRequest REDACTED/repro.ts:13:18) {
  cause: RequestContentLengthMismatchError: Request body length does not match content-length header
      at AsyncWriter.end REDACTED/node_modules/undici/lib/dispatcher/client-h1.js:1319:15)
      at writeIterable (REDACTED/node_modules/undici/lib/dispatcher/client-h1.js:1201:12) {
    code: 'UND_ERR_REQ_CONTENT_LENGTH_MISMATCH'
  }
}

Node.js v20.8.1
```

## Environment

* macOS 14.4.1
* nodejs v20.8.1
* undici 6.18.0

### Additional context

None



Hints:
This is indeed tricky. On one hand, state altering verbs should not be automatically retried, on the other hand, it's a pretty common case.


It might be better to just scope this to the safe methods only `GET`, `HEAD`, and `OPTIONS` instead of the idempotent ones.

Tho the issue seems to root the `AsyncIterable` that is made out of `fetch`.
As the `AsyncIterable` has been already consumed at its first request, when the `RetryHandler` attempts a new request, the `content-length` mismatch surfaces as `undici` fails to iterate over the body because it was already consumed.

Not sure what will be the course of action here, I'd say that maybe the handler can make a best attempt to detect specific bodies (Stream, AsyncIterable) and clone them and attempt to free them after successful request. Just will add the overhead of duplicated bodies
I think the fix here is to avoid the use of an AsyncIterable in fetch if the full body was specified. We can save quite a bit of overhead.
FWIW I tested directly with `Streams` and are handled properly by the `RetryHandler`, but facing similar issue to as per fetch with `AsyncIterator` (once consumed it does not loop). So this is definitely something that the handler might need to be aware of

 

Patch:
diff --git a/lib/core/symbols.js b/lib/core/symbols.js
--- a/lib/core/symbols.js
+++ b/lib/core/symbols.js
@@ -20,6 +20,7 @@ module.exports = {
   kHost: Symbol('host'),
   kNoRef: Symbol('no ref'),
   kBodyUsed: Symbol('used'),
+  kBody: Symbol('abstracted request body'),
   kRunning: Symbol('running'),
   kBlocking: Symbol('blocking'),
   kPending: Symbol('pending'),
diff --git a/lib/core/util.js b/lib/core/util.js
--- a/lib/core/util.js
+++ b/lib/core/util.js
@@ -1,19 +1,72 @@
 'use strict'
 
 const assert = require('node:assert')
-const { kDestroyed, kBodyUsed, kListeners } = require('./symbols')
+const { kDestroyed, kBodyUsed, kListeners, kBody } = require('./symbols')
 const { IncomingMessage } = require('node:http')
 const stream = require('node:stream')
 const net = require('node:net')
-const { InvalidArgumentError } = require('./errors')
 const { Blob } = require('node:buffer')
 const nodeUtil = require('node:util')
 const { stringify } = require('node:querystring')
+const { EventEmitter: EE } = require('node:events')
+const { InvalidArgumentError } = require('./errors')
 const { headerNameLowerCasedRecord } = require('./constants')
 const { tree } = require('./tree')
 
 const [nodeMajor, nodeMinor] = process.versions.node.split('.').map(v => Number(v))
 
+class BodyAsyncIterable {
+  constructor (body) {
+    this[kBody] = body
+    this[kBodyUsed] = false
+  }
+
+  async * [Symbol.asyncIterator] () {
+    assert(!this[kBodyUsed], 'disturbed')
+    this[kBodyUsed] = true
+    yield * this[kBody]
+  }
+}
+
+function wrapRequestBody (body) {
+  if (isStream(body)) {
+    // TODO (fix): Provide some way for the user to cache the file to e.g. /tmp
+    // so that it can be dispatched again?
+    // TODO (fix): Do we need 100-expect support to provide a way to do this properly?
+    if (bodyLength(body) === 0) {
+      body
+        .on('data', function () {
+          assert(false)
+        })
+    }
+
+    if (typeof body.readableDidRead !== 'boolean') {
+      body[kBodyUsed] = false
+      EE.prototype.on.call(body, 'data', function () {
+        this[kBodyUsed] = true
+      })
+    }
+
+    return body
+  } else if (body && typeof body.pipeTo === 'function') {
+    // TODO (fix): We can't access ReadableStream internal state
+    // to determine whether or not it has been disturbed. This is just
+    // a workaround.
+    return new BodyAsyncIterable(body)
+  } else if (
+    body &&
+    typeof body !== 'string' &&
+    !ArrayBuffer.isView(body) &&
+    isIterable(body)
+  ) {
+    // TODO: Should we allow re-using iterable if !this.opts.idempotent
+    // or through some other flag?
+    return new BodyAsyncIterable(body)
+  } else {
+    return body
+  }
+}
+
 function nop () {}
 
 function isStream (obj) {
@@ -634,5 +687,6 @@ module.exports = {
   isHttpOrHttpsPrefixed,
   nodeMajor,
   nodeMinor,
-  safeHTTPMethods: ['GET', 'HEAD', 'OPTIONS', 'TRACE']
+  safeHTTPMethods: ['GET', 'HEAD', 'OPTIONS', 'TRACE'],
+  wrapRequestBody
 }
diff --git a/lib/handler/retry-handler.js b/lib/handler/retry-handler.js
--- a/lib/handler/retry-handler.js
+++ b/lib/handler/retry-handler.js
@@ -3,7 +3,12 @@ const assert = require('node:assert')
 
 const { kRetryHandlerDefaultRetry } = require('../core/symbols')
 const { RequestRetryError } = require('../core/errors')
-const { isDisturbed, parseHeaders, parseRangeHeader } = require('../core/util')
+const {
+  isDisturbed,
+  parseHeaders,
+  parseRangeHeader,
+  wrapRequestBody
+} = require('../core/util')
 
 function calculateRetryAfterHeader (retryAfter) {
   const current = Date.now()
@@ -29,7 +34,7 @@ class RetryHandler {
 
     this.dispatch = handlers.dispatch
     this.handler = handlers.handler
-    this.opts = dispatchOpts
+    this.opts = { ...dispatchOpts, body: wrapRequestBody(opts.body) }
     this.abort = null
     this.aborted = false
     this.retryOpts = {
@@ -174,7 +179,9 @@ class RetryHandler {
         this.abort(
           new RequestRetryError('Request failed', statusCode, {
             headers,
-            count: this.retryCount
+            data: {
+              count: this.retryCount
+            }
           })
         )
         return false
@@ -278,7 +285,7 @@ class RetryHandler {
 
     const err = new RequestRetryError('Request failed', statusCode, {
       headers,
-      count: this.retryCount
+      data: { count: this.retryCount }
     })
 
     this.abort(err)


Test patch:
diff --git a/test/retry-handler.js b/test/retry-handler.js
--- a/test/retry-handler.js
+++ b/test/retry-handler.js
@@ -4,6 +4,7 @@ const { tspl } = require('@matteo.collina/tspl')
 const { test, after } = require('node:test')
 const { createServer } = require('node:http')
 const { once } = require('node:events')
+const { Readable } = require('node:stream')
 
 const { RetryHandler, Client } = require('..')
 const { RequestHandler } = require('../lib/api/api-request')
@@ -204,6 +205,74 @@ test('Should account for network and response errors', async t => {
   await t.completed
 })
 
+test('Issue #3288 - request with body (asynciterable)', async t => {
+  t = tspl(t, { plan: 6 })
+  const server = createServer()
+  const dispatchOptions = {
+    method: 'POST',
+    path: '/',
+    headers: {
+      'content-type': 'application/json'
+    },
+    body: (function * () {
+      yield 'hello'
+      yield 'world'
+    })()
+  }
+
+  server.on('request', (req, res) => {
+    res.writeHead(500, {
+      'content-type': 'application/json'
+    })
+
+    res.end('{"message": "failed"}')
+  })
+
+  server.listen(0, () => {
+    const client = new Client(`http://localhost:${server.address().port}`)
+    const handler = new RetryHandler(dispatchOptions, {
+      dispatch: client.dispatch.bind(client),
+      handler: {
+        onConnect () {
+          t.ok(true, 'pass')
+        },
+        onBodySent () {
+          t.ok(true, 'pass')
+        },
+        onHeaders (status, _rawHeaders, resume, _statusMessage) {
+          t.strictEqual(status, 500)
+          return true
+        },
+        onData (chunk) {
+          return true
+        },
+        onComplete () {
+          t.fail()
+        },
+        onError (err) {
+          t.equal(err.message, 'Request failed')
+          t.equal(err.statusCode, 500)
+          t.equal(err.data.count, 1)
+        }
+      }
+    })
+
+    after(async () => {
+      await client.close()
+      server.close()
+
+      await once(server, 'close')
+    })
+
+    client.dispatch(
+      dispatchOptions,
+      handler
+    )
+  })
+
+  await t.completed
+})
+
 test('Should use retry-after header for retries', async t => {
   t = tspl(t, { plan: 4 })
 
@@ -734,6 +803,145 @@ test('retrying a request with a body', async t => {
   await t.completed
 })
 
+test('retrying a request with a body (stream)', async t => {
+  let counter = 0
+  const server = createServer()
+  const dispatchOptions = {
+    retryOptions: {
+      retry: (err, { state, opts }, done) => {
+        counter++
+
+        if (
+          err.statusCode === 500 ||
+          err.message.includes('other side closed')
+        ) {
+          setTimeout(done, 500)
+          return
+        }
+
+        return done(err)
+      }
+    },
+    method: 'POST',
+    path: '/',
+    headers: {
+      'content-type': 'application/json'
+    },
+    body: Readable.from(Buffer.from(JSON.stringify({ hello: 'world' })))
+  }
+
+  t = tspl(t, { plan: 3 })
+
+  server.on('request', (req, res) => {
+    switch (counter) {
+      case 0:
+        res.writeHead(500)
+        res.end('failed')
+        return
+      default:
+        t.fail()
+    }
+  })
+
+  server.listen(0, () => {
+    const client = new Client(`http://localhost:${server.address().port}`)
+    const handler = new RetryHandler(dispatchOptions, {
+      dispatch: client.dispatch.bind(client),
+      handler: new RequestHandler(dispatchOptions, (err, data) => {
+        t.equal(err.statusCode, 500)
+        t.equal(err.data.count, 1)
+        t.equal(err.code, 'UND_ERR_REQ_RETRY')
+      })
+    })
+
+    after(async () => {
+      await client.close()
+      server.close()
+
+      await once(server, 'close')
+    })
+
+    client.dispatch(
+      dispatchOptions,
+      handler
+    )
+  })
+
+  await t.completed
+})
+
+test('retrying a request with a body (buffer)', async t => {
+  let counter = 0
+  const server = createServer()
+  const dispatchOptions = {
+    retryOptions: {
+      retry: (err, { state, opts }, done) => {
+        counter++
+
+        if (
+          err.statusCode === 500 ||
+          err.message.includes('other side closed')
+        ) {
+          setTimeout(done, 500)
+          return
+        }
+
+        return done(err)
+      }
+    },
+    method: 'POST',
+    path: '/',
+    headers: {
+      'content-type': 'application/json'
+    },
+    body: Buffer.from(JSON.stringify({ hello: 'world' }))
+  }
+
+  t = tspl(t, { plan: 1 })
+
+  server.on('request', (req, res) => {
+    switch (counter) {
+      case 0:
+        req.destroy()
+        return
+      case 1:
+        res.writeHead(500)
+        res.end('failed')
+        return
+      case 2:
+        res.writeHead(200)
+        res.end('hello world!')
+        return
+      default:
+        t.fail()
+    }
+  })
+
+  server.listen(0, () => {
+    const client = new Client(`http://localhost:${server.address().port}`)
+    const handler = new RetryHandler(dispatchOptions, {
+      dispatch: client.dispatch.bind(client),
+      handler: new RequestHandler(dispatchOptions, (err, data) => {
+        t.ifError(err)
+      })
+    })
+
+    after(async () => {
+      await client.close()
+      server.close()
+
+      await once(server, 'close')
+    })
+
+    client.dispatch(
+      dispatchOptions,
+      handler
+    )
+  })
+
+  await t.completed
+})
+
 test('should not error if request is not meant to be retried', async t => {
   t = tspl(t, { plan: 3 })
 
@@ -777,8 +985,7 @@ test('should not error if request is not meant to be retried', async t => {
           t.strictEqual(Buffer.concat(chunks).toString('utf-8'), 'Bad request')
         },
         onError (err) {
-          console.log({ err })
-          t.fail()
+          t.fail(err)
         }
       }
     })