Introduce exponential backoff for stream handler errors (#260)

Author: austinborn

.github/workflows/main.yaml

@@ -37,7 +37,7 @@ jobs:
           set +e
             files=$(find ${{ github.workspace }}/test -name "*.test.ts" | xargs grep -l 'test.only')
           set -e
-          
+
           if [[ -n $files ]]; then
             echo "Error: Found /test.only/ in following test files:"
             echo "$files"

src/background-executor.ts

@@ -95,7 +95,7 @@ export async function callBackgroundExecutes(adapter: Adapter, apiShutdownPromis
         `Finished background execute for endpoint "${endpoint.name}", calling it again in 10ms...`,
       )
       metricsTimer()
-      timeoutsMap[endpoint.name] = setTimeout(handler, 10)
+      timeoutsMap[endpoint.name] = setTimeout(handler, 10) // 10ms
     }
 
     // Start recursive async calls

src/config/index.ts

@@ -2,11 +2,6 @@ import CensorList, { CensorKeyValue } from '../util/censor/censor-list'
 import { Validator, validator } from '../validation/utils'
 
 export const BaseSettingsDefinition = {
-  // V2 compat
-  // ADAPTER_URL: {
-  //   description: 'The URL of another adapter from which data needs to be retrieved',
-  //   type: 'string',
-  // },
   API_TIMEOUT: {
     description:
       'The number of milliseconds a request can be pending before returning a timeout error for data provider request',
@@ -63,11 +58,6 @@ export const BaseSettingsDefinition = {
     type: 'string',
     default: '127.0.0.1',
   },
-  // CACHE_REDIS_MAX_QUEUED_ITEMS: {
-  //   description: 'Maximum length of the client internal command queue',
-  //   type: 'number',
-  //   default: 500,
-  // },
   CACHE_REDIS_MAX_RECONNECT_COOLDOWN: {
     description: 'Max cooldown (in ms) before attempting redis reconnection',
     type: 'number',
@@ -112,6 +102,26 @@ export const BaseSettingsDefinition = {
     type: 'string',
     default: '',
   },
+  STREAM_HANDLER_RETRY_MAX_MS: {
+    type: 'number',
+    description:
+      'The maximum time (ms) to wait before running the stream handler (takes precedent over STREAM_HANDLER_RETRY_MIN_MS',
+    default: 20 * 60 * 1000,
+    validate: validator.integer({ min: 3_000, max: 24 * 60 * 60 * 1000 }),
+  },
+  STREAM_HANDLER_RETRY_MIN_MS: {
+    type: 'number',
+    description: 'The minimum/base time (ms) to wait before trying to run the stream handler',
+    default: 100,
+    validate: validator.integer({ min: 100, max: 10_000 }),
+  },
+  STREAM_HANDLER_RETRY_EXP_FACTOR: {
+    type: 'number',
+    description:
+      'The factor for exponential back-off to wait before running the stream handler (1 = no change from STREAM_HANDLER_RETRY_MIN_MS)',
+    default: 3,
+    validate: validator.integer({ min: 1, max: 10 }),
+  },
   SUBSCRIPTION_SET_MAX_ITEMS: {
     type: 'number',
     description: 'The maximum number of subscriptions set',
@@ -123,30 +133,18 @@ export const BaseSettingsDefinition = {
     type: 'boolean',
     default: true,
   },
-  // DATA_PROVIDER_URL: {
-  //   description: 'Legacy variable that has the same functionality as ADAPTER_URL',
-  //   type: 'string',
-  // },
   DEBUG: {
     description: 'Toggles debug mode',
     type: 'boolean',
     default: false,
   },
-  // DEFAULT_WS_HEARTBEAT_INTERVAL: {
-  //   description: 'Interval between WS heartbeat pings (ms)',
-  //   type: 'number',
-  //   default: 30000,
-  // },
   EA_PORT: {
     description:
       'Port through which the EA will listen for REST requests (if mode is set to "reader" or "reader-writer")',
     type: 'number',
     default: 8080,
     validate: validator.port(),
   },
-  // ERROR_CAPACITY: {
-  //   type: 'number',
-  // },
   EXPERIMENTAL_METRICS_ENABLED: {
     description:
       'Flag to specify whether or not to collect metrics. Used as fallback for METRICS_ENABLED',
@@ -250,12 +248,6 @@ export const BaseSettingsDefinition = {
     default: 10_000,
     validate: validator.integer({ min: 500, max: 30_000 }),
   },
-  // WS_TIME_UNTIL_HANDLE_NEXT_MESSAGE_OVERRIDE: {
-  //   description: 'Time to wait until adapter should handle next WS message',
-  //   type: 'number',
-  // },
-
-  // V3
   CACHE_POLLING_MAX_RETRIES: {
     description:
       'Max amount of times to attempt to find EA response in the cache after the Transport has been set up',

src/metrics/index.ts

@@ -294,6 +294,11 @@ export const metrics = new Metrics(() => ({
     help: 'The number of redis commands sent',
     labelNames: ['status', 'function_name'],
   }),
+  streamHandlerErrors: new client.Counter({
+    name: 'stream_handler_errors',
+    help: 'The number of stream handler errors per endpoint x transport',
+    labelNames: ['adapter_endpoint', 'transport'] as const,
+  }),
   cacheWarmerCount: new client.Gauge({
     name: 'cache_warmer_get_count',
     help: 'The number of cache warmers running',

src/transports/abstract/streaming.ts

@@ -3,6 +3,7 @@ import { EndpointContext } from '../../adapter'
 import { censorLogs, makeLogger } from '../../util'
 import { TypeFromDefinition } from '../../validation/input-params'
 import { SubscriptionTransport } from './subscription'
+import { metrics } from '../../metrics'
 
 const logger = makeLogger('StreamingTransport')
 
@@ -37,10 +38,17 @@ export abstract class StreamingTransport<
   // This only tracks when the connection was established, not when the subscription was requested
   providerDataStreamEstablished = 0
 
+  retryCount = 0
+  retryTime = 0
+
   override async backgroundHandler(
     context: EndpointContext<T>,
     desiredSubs: TypeFromDefinition<T['Parameters']>[],
   ): Promise<void> {
+    if (this.retryTime > Date.now()) {
+      return
+    }
+
     logger.debug('Generating delta (subscribes & unsubscribes)')
 
     const desiredSubsSet = new Set(desiredSubs.map((s) => JSON.stringify(s)))
@@ -62,7 +70,25 @@ export abstract class StreamingTransport<
       censorLogs(() => logger.trace(`Will unsubscribe to: ${JSON.stringify(subscriptions.stale)}`))
     }
 
-    await this.streamHandler(context, subscriptions)
+    try {
+      await this.streamHandler(context, subscriptions)
+      this.retryCount = 0
+    } catch (error) {
+      censorLogs(() => logger.error(error, (error as Error).stack))
+      metrics
+        .get('streamHandlerErrors')
+        .labels({ adapter_endpoint: context.endpointName, transport: this.name })
+        .inc()
+      const timeout = Math.min(
+        context.adapterSettings.STREAM_HANDLER_RETRY_MIN_MS *
+          context.adapterSettings.STREAM_HANDLER_RETRY_EXP_FACTOR ** this.retryCount,
+        context.adapterSettings.STREAM_HANDLER_RETRY_MAX_MS,
+      )
+      this.retryTime = Date.now() + timeout
+      this.retryCount += 1
+      logger.info(`Waiting ${timeout}ms before backgroundHandler retry #${this.retryCount}`)
+      return
+    }
 
     logger.debug('Setting local state to subscription set value')
     this.localSubscriptions = desiredSubs

src/transports/abstract/subscription.ts

@@ -14,9 +14,7 @@ const logger = makeLogger('SubscriptionTransport')
  *
  * @typeParam T - all types related to the [[Transport]]
  */
-export abstract class SubscriptionTransport<const T extends TransportGenerics>
-  implements Transport<T>
-{
+export abstract class SubscriptionTransport<T extends TransportGenerics> implements Transport<T> {
   responseCache!: ResponseCache<T>
   subscriptionSet!: SubscriptionSet<TypeFromDefinition<T['Parameters']>>
   subscriptionTtl!: number
@@ -54,6 +52,7 @@ export abstract class SubscriptionTransport<const T extends TransportGenerics>
 
   async backgroundExecute(context: EndpointContext<T>): Promise<void> {
     logger.debug('Starting background execute')
+
     const entries = await this.subscriptionSet.getAll()
 
     // Keep track of active subscriptions for background execute

test/transports/websocket.test.ts

@@ -520,7 +520,7 @@ test.serial(
 
     const metrics = await testAdapter.getMetrics()
     metrics.assert(t, {
-      name: 'bg_execute_errors',
+      name: 'stream_handler_errors',
       labels: {
         adapter_endpoint: 'test',
         transport: 'default_single_transport',
@@ -776,7 +776,7 @@ test.serial('does not hang the background execution if the open handler hangs',
 
   const metrics = await testAdapter.getMetrics()
   metrics.assert(t, {
-    name: 'bg_execute_errors',
+    name: 'stream_handler_errors',
     labels: {
       adapter_endpoint: 'test',
       transport: 'default_single_transport',