add retry for http retryable requests (#1390)

Author: prakashsvmx

src/internal/client.ts

@@ -66,7 +66,7 @@ import {
 } from './helper.ts'
 import { joinHostPort } from './join-host-port.ts'
 import { PostPolicy } from './post-policy.ts'
-import { request } from './request.ts'
+import { requestWithRetry } from './request.ts'
 import { drainResponse, readAsBuffer, readAsString } from './response.ts'
 import type { Region } from './s3-endpoints.ts'
 import { getS3Endpoint } from './s3-endpoints.ts'
@@ -723,7 +723,7 @@ export class TypedClient {
       reqOptions.headers.authorization = signV4(reqOptions, this.accessKey, this.secretKey, region, date, sha256sum)
     }
 
-    const response = await request(this.transport, reqOptions, body)
+    const response = await requestWithRetry(this.transport, reqOptions, body)
     if (!response.statusCode) {
       throw new Error("BUG: response doesn't have a statusCode")
     }

src/internal/request.ts

@@ -2,34 +2,89 @@ import type * as http from 'node:http'
 import type * as https from 'node:https'
 import type * as stream from 'node:stream'
 import { pipeline } from 'node:stream'
+import { promisify } from 'node:util'
 
 import type { Transport } from './type.ts'
 
+const pipelineAsync = promisify(pipeline)
+
 export async function request(
   transport: Transport,
   opt: https.RequestOptions,
   body: Buffer | string | stream.Readable | null = null,
 ): Promise<http.IncomingMessage> {
   return new Promise<http.IncomingMessage>((resolve, reject) => {
-    const requestObj = transport.request(opt, (resp) => {
-      resolve(resp)
+    const requestObj = transport.request(opt, (response) => {
+      resolve(response)
     })
 
-    if (!body || Buffer.isBuffer(body) || typeof body === 'string') {
-      requestObj
-        .on('error', (e: unknown) => {
-          reject(e)
-        })
-        .end(body)
+    requestObj.on('error', reject)
 
-      return
+    if (!body || Buffer.isBuffer(body) || typeof body === 'string') {
+      requestObj.end(body)
+    } else {
+      pipelineAsync(body, requestObj).catch(reject)
     }
+  })
+}
+
+const MAX_RETRIES = 10
+const EXP_BACK_OFF_BASE_DELAY = 1000 // Base delay for exponential backoff
+const ADDITIONAL_DELAY_FACTOR = 1.0 // to avoid synchronized retries
+
+// Retryable error codes for HTTP ( ref: minio-go)
+export const retryHttpCodes: Record<string, boolean> = {
+  408: true,
+  429: true,
+  499: true,
+  500: true,
+  502: true,
+  503: true,
+  504: true,
+  520: true,
+}
+
+const isHttpRetryable = (httpResCode: number) => {
+  return retryHttpCodes[httpResCode] !== undefined
+}
+
+const sleep = (ms: number) => {
+  return new Promise((resolve) => setTimeout(resolve, ms))
+}
 
-    // pump readable stream
-    pipeline(body, requestObj, (err) => {
-      if (err) {
-        reject(err)
+const getExpBackOffDelay = (retryCount: number) => {
+  const backOffBy = EXP_BACK_OFF_BASE_DELAY * 2 ** retryCount
+  const additionalDelay = Math.random() * backOffBy * ADDITIONAL_DELAY_FACTOR
+  return backOffBy + additionalDelay
+}
+
+export async function requestWithRetry(
+  transport: Transport,
+  opt: https.RequestOptions,
+  body: Buffer | string | stream.Readable | null = null,
+  maxRetries: number = MAX_RETRIES,
+): Promise<http.IncomingMessage> {
+  let attempt = 0
+  while (attempt <= maxRetries) {
+    try {
+      const response = await request(transport, opt, body)
+      // Check if the HTTP status code is retryable
+      if (isHttpRetryable(response.statusCode as number)) {
+        throw new Error(`Retryable HTTP status: ${response.statusCode}`) // trigger retry attempt with calculated delay
       }
-    })
-  })
+      return response // Success, return the raw response
+    } catch (err) {
+      attempt++
+
+      if (attempt > maxRetries) {
+        throw new Error(`Request failed after ${maxRetries} retries: ${err}`)
+      }
+      const delay = getExpBackOffDelay(attempt)
+      // eslint-disable-next-line no-console
+      // console.warn( `${new Date().toLocaleString()} Retrying request (attempt ${attempt}/${maxRetries}) after ${delay}ms due to: ${err}`,)
+      await sleep(delay)
+    }
+  }
+
+  throw new Error(`${MAX_RETRIES} Retries exhausted, request failed.`)
 }

src/internal/xml-parser.ts

@@ -56,9 +56,10 @@ export function parseError(xml: string, headerInfo: Record<string, unknown>) {
 }
 
 // Generates an Error object depending on http statusCode and XML body
-export async function parseResponseError(response: http.IncomingMessage) {
+export async function parseResponseError(response: http.IncomingMessage): Promise<Record<string, string>> {
   const statusCode = response.statusCode
-  let code: string, message: string
+  let code = '',
+    message = ''
   if (statusCode === 301) {
     code = 'MovedPermanently'
     message = 'Moved Permanently'
@@ -77,9 +78,17 @@ export async function parseResponseError(response: http.IncomingMessage) {
   } else if (statusCode === 501) {
     code = 'MethodNotAllowed'
     message = 'Method Not Allowed'
+  } else if (statusCode === 503) {
+    code = 'SlowDown'
+    message = 'Please reduce your request rate.'
   } else {
-    code = 'UnknownError'
-    message = `${statusCode}`
+    const hErrCode = response.headers['x-minio-error-code'] as string
+    const hErrDesc = response.headers['x-minio-error-desc'] as string
+
+    if (hErrCode && hErrDesc) {
+      code = hErrCode
+      message = hErrDesc
+    }
   }
   const headerInfo: Record<string, string | undefined | null> = {}
   // A value created by S3 compatible server that uniquely identifies the request.

tests/unit/test.js

@@ -751,12 +751,6 @@ describe('Client', function () {
           () => done(),
         )
       })
-      it('should fail on incompatible argument type (null) for statOpts object', (done) => {
-        client.statObject('hello', 'testStatOpts', null).then(
-          () => done(new Error('expecting error')),
-          () => done(),
-        )
-      })
       it('should fail on incompatible argument type (sting) for statOpts object', (done) => {
         client.statObject('hello', 'testStatOpts', '  ').then(
           () => done(new Error('expecting error')),