fix: retry on body support

lib/handler/retry-handler.js

@@ -3,14 +3,22 @@ const assert = require('node:assert')
 
 const { kRetryHandlerDefaultRetry } = require('../core/symbols')
 const { RequestRetryError } = require('../core/errors')
-const { isDisturbed, parseHeaders, parseRangeHeader } = require('../core/util')
+const {
+  isStream,
+  isDisturbed,
+  isIterable,
+  isAsyncIterable,
+  parseHeaders,
+  parseRangeHeader
+} = require('../core/util')
 
 function calculateRetryAfterHeader (retryAfter) {
   const current = Date.now()
   return new Date(retryAfter).getTime() - current
 }
 
 class RetryHandler {
+  #isBodyIteratorLike = false
   constructor (opts, handlers) {
     const { retryOptions, ...dispatchOpts } = opts
     const {
@@ -63,6 +71,10 @@ class RetryHandler {
     this.end = null
     this.etag = null
     this.resume = null
+    this.#isBodyIteratorLike =
+      typeof this.opts.body === 'object' &&
+      !isStream(this.opts.body) &&
+      (isIterable(this.opts.body) || isAsyncIterable(this.opts.body))
 
     // Handle possible onConnect duplication
     this.handler.onConnect(reason => {
@@ -174,7 +186,9 @@ class RetryHandler {
         this.abort(
           new RequestRetryError('Request failed', statusCode, {
             headers,
-            count: this.retryCount
+            data: {
+              count: this.retryCount
+            }
           })
         )
         return false
@@ -278,7 +292,7 @@ class RetryHandler {
 
     const err = new RequestRetryError('Request failed', statusCode, {
       headers,
-      count: this.retryCount
+      data: { count: this.retryCount }
     })
 
     this.abort(err)
@@ -298,7 +312,16 @@ class RetryHandler {
   }
 
   onError (err) {
-    if (this.aborted || isDisturbed(this.opts.body)) {
+    console.log(
+      this.aborted,
+      isDisturbed(this.opts.body),
+      this.#isBodyIteratorLike
+    )
+    if (
+      this.aborted ||
+      isDisturbed(this.opts.body) ||
+      this.#isBodyIteratorLike
+    ) {
       return this.handler.onError(err)
     }
 

test/retry-handler.js

@@ -1,9 +1,10 @@
 '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 { Readable } = require('node:stream')
 
 const { RetryHandler, Client } = require('..')
 const { RequestHandler } = require('../lib/api/api-request')
@@ -204,6 +205,75 @@
   await t.completed
 })
 
+test('Issue #3288', { only: true }, 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) {
+          console.log('chunk', 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 })
 
@@ -655,7 +725,7 @@
   await t.completed
 })
 
-test('retrying a request with a body', async t => {
+test('retrying a request with a body', { only: true }, async t => {
   let counter = 0
   const server = createServer()
   const dispatchOptions = {
@@ -685,6 +755,7 @@
   t = tspl(t, { plan: 1 })
 
   server.on('request', (req, res) => {
+    console.log({ counter })
     switch (counter) {
       case 0:
         req.destroy()
@@ -734,6 +805,146 @@
   await t.completed
 })
 
+test('retrying a request with a body (stream)', { only: true }, 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) => {
+    console.log({ counter })
+    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)', { only: true }, 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 })