feat: improving retries; consider restart of app

node/clients/index.ts

@@ -6,7 +6,7 @@ import type {
   ImportExecution,
 } from 'ssesandbox04.catalog-importer'
 
-import { CURRENT_MD_SCHEMA, MD_ENTITIES } from '../helpers'
+import { CURRENT_MD_SCHEMA, MAX_RETRIES, MD_ENTITIES } from '../helpers'
 import Cosmos from './Cosmos'
 import HttpClient from './HttpClient'
 import PrivateClient from './PrivateClient'
@@ -59,7 +59,7 @@ export default {
       exponentialTimeoutCoefficient: 2,
       exponentialBackoffCoefficient: 2,
       initialBackoffDelay: 100,
-      retries: 10,
+      retries: MAX_RETRIES,
       timeout: 300000,
       concurrency: 1,
     },

node/helpers/constants.ts

@@ -123,6 +123,7 @@ export const IMPORT_STATUS: { [keyof in ImportStatus]: ImportStatus } = {
 
 export const STEP_DELAY = 1000
 export const DEFAULT_BATCH_CONCURRENCY = 500
+export const MAX_RETRIES = 10
 export const ONE_RESULT = { page: 1, pageSize: 1 }
 export const COMMON_WHERE = `(status<>${IMPORT_STATUS.TO_BE_DELETED})AND(status<>${IMPORT_STATUS.DELETING})`
 export const GET_DETAILS_CONCURRENCY = 25

node/helpers/setupVerifyImports.ts

@@ -3,24 +3,39 @@ import {
   getFirstImportPending,
   getFirstImportRunning,
   getFirstImportToBeDeleted,
+  IMPORT_STATUS,
+  updateImportStatus,
 } from '.'
 import runImport from '../events'
 
 let cachedContext: Context | undefined
 const TIMEOUT = 10000
+let hasImportRun = false
 
-export const getCachedContext = () => {
+export function getCachedContext() {
   return cachedContext
 }
 
 export function setCachedContext(context: Context) {
   cachedContext = context
 }
 
+export function setHasImportRun() {
+  hasImportRun = true
+}
+
+export function getHasImportRun() {
+  return hasImportRun
+}
+
 const verifyImports = async () => {
   const context = getCachedContext()
 
-  if (!context || (await getFirstImportRunning(context))) return
+  if (!context) return
+
+  const importRunning = await getFirstImportRunning(context)
+
+  if (importRunning && getHasImportRun()) return
 
   const nextImportToBeDeleted = await getFirstImportToBeDeleted(context)
 
@@ -30,12 +45,24 @@ const verifyImports = async () => {
     return
   }
 
-  const nextPendingImport = await getFirstImportPending(context)
+  const nextPendingImport =
+    importRunning && !getHasImportRun()
+      ? importRunning
+      : await getFirstImportPending(context)
 
   if (!nextPendingImport) return
 
+  if (nextPendingImport.status === IMPORT_STATUS.RUNNING) {
+    await updateImportStatus(
+      context,
+      nextPendingImport.id,
+      IMPORT_STATUS.PENDING
+    )
+  }
+
   context.state.body = nextPendingImport
   runImport(context)
+  setHasImportRun()
 }
 
 export const setupVerifyImports = () => {

node/helpers/utils.ts

@@ -5,6 +5,7 @@ import {
   DEFAULT_BATCH_CONCURRENCY,
   DEFAULT_VBASE_BUCKET,
   IMPORT_STATUS,
+  MAX_RETRIES,
   STEP_DELAY,
   STEPS,
 } from '.'
@@ -16,9 +17,34 @@ export const getCurrentSettings = async ({ clients: { apps } }: Context) =>
 export const delay = (ms: number) =>
   new Promise((resolve) => setTimeout(resolve, ms))
 
+const promiseWithConditionalRetry = async <T, R = void>(
+  fn: (element: T) => Promise<Maybe<R>> | R,
+  arg: T,
+  retries = 0
+): Promise<Maybe<R>> => {
+  return Promise.resolve(fn?.(arg))?.catch(async (e) => {
+    const message = e.message.toLowerCase()
+    const messageToRetry =
+      message.includes('500') ||
+      message.includes('429') ||
+      message.includes('network error') ||
+      message.includes('networkerror') ||
+      message.includes('genericerror') ||
+      message.includes('unhealthy')
+
+    if (messageToRetry && retries < MAX_RETRIES) {
+      await delay(STEP_DELAY * (retries + 1))
+
+      return promiseWithConditionalRetry(fn, arg, retries + 1)
+    }
+
+    throw e
+  })
+}
+
 export const batch = async <T, R = void>(
   data: T[],
-  elementCallback: (element: T) => Maybe<Promise<R>> | R,
+  elementCallback: (element: T) => Promise<Maybe<R>> | R,
   concurrency = DEFAULT_BATCH_CONCURRENCY
 ) => {
   const cloneData = [...data]
@@ -28,7 +54,9 @@ export const batch = async <T, R = void>(
     if (!cloneData.length) return
 
     const result = ((await Promise.all(
-      cloneData.splice(0, concurrency).map(elementCallback)
+      cloneData.splice(0, concurrency).map(async (element) => {
+        return promiseWithConditionalRetry(elementCallback, element)
+      })
     )) as unknown) as R
 
     if (result) {
@@ -45,7 +73,7 @@ export const batch = async <T, R = void>(
 
 export const sequentialBatch = async <T, R = void>(
   data: T[],
-  elementCallback: (element: T) => Maybe<Promise<R>> | R
+  elementCallback: (element: T) => Promise<Maybe<R>> | R
 ) => {
   return batch(data, elementCallback, 1)
 }

react/components/graphql/index.ts

@@ -31,7 +31,7 @@ type CustomQueryHookOptions<T, V> = QueryHookOptions<T, V> &
 type CustomMutationHookOptions<T, V> = MutationHookOptions<T, V> &
   CustomGraphQLOptions
 
-const MAX_RETRIES = 5
+const MAX_RETRIES = 10
 const RETRY_DELAY = 500
 
 export const getGraphQLMessageDescriptor = (error: GraphQLError) => ({
@@ -57,7 +57,7 @@ const useErrorRetry = <T = Query, V = undefined>(
       message.includes('unhealthy')
 
     if (messageToRetry && retries.current < MAX_RETRIES) {
-      setTimeout(() => refetch(), RETRY_DELAY)
+      setTimeout(() => refetch(), RETRY_DELAY * (retries.current + 1))
       retries.current++
     } else {
       toastError &&