refactor timeout stuff to use options, not env directly

Author: josephjclark

packages/ws-worker/src/util/convert-lightning-plan.ts

@@ -48,6 +48,8 @@ export type WorkerRunOptions = ExecuteOptions & {
   outputDataclips?: boolean;
   payloadLimitMb?: number;
   jobLogLevel?: LogLevel;
+  timeoutRetryCount?: number;
+  timeoutRetryDelay?: number;
 };
 
 type ConversionOptions = {

packages/ws-worker/src/util/send-event.ts

@@ -2,24 +2,28 @@ import * as Sentry from '@sentry/node';
 import type { Context } from '../api/execute';
 import { LightningSocketError, LightningTimeoutError } from '../errors';
 
+// // When a message receives a timeout, how many times should we retry?
+// const TIMEOUT_RETRY_COUNT = process.env.WORKER_TIMEOUT_RETRY_COUNT
+//   ? parseInt(process.env.WORKER_TIMEOUT_RETRY_COUNT)
+//   : 10;
+
+// // When a message receives a timeout, how long should we wait before retrying?
+// const TIMEOUT_RETRY_DELAY =
+//   process.env.WORKER_TIMEOUT_RETRY_DELAY ??
+//   process.env.WORKER_MESSAGE_TIMEOUT_SECONDS ??
+//   30 * 1000;
+
 export const sendEvent = <T>(
-  context: Pick<Context, 'logger' | 'channel' | 'id'>,
+  context: Pick<Context, 'logger' | 'channel' | 'id' | 'options'>,
   event: string,
   payload?: any,
   attempts?: number
 ) => {
-  const thisAttempt = attempts ?? 1;
+  // Low defaults here are better for unit tests
+  const { timeoutRetryCount = 1, timeoutRetryDelay = 1 } =
+    context.options ?? {};
 
-  // When a message receives a timeout, how many times should we retry?
-  const TIMEOUT_RETRY_COUNT = process.env.WORKER_TIMEOUT_RETRY_COUNT
-    ? parseInt(process.env.WORKER_TIMEOUT_RETRY_COUNT)
-    : 10;
-
-  // When a message receives a timeout, how long should we wait before retrying?
-  const TIMEOUT_RETRY_DELAY =
-    process.env.WORKER_TIMEOUT_RETRY_DELAY ??
-    process.env.WORKER_MESSAGE_TIMEOUT_SECONDS ??
-    30 * 1000;
+  const thisAttempt = attempts ?? 1;
 
   const { channel, logger, id: runId = '<unknown run>' } = context;
 
@@ -55,25 +59,20 @@ export const sendEvent = <T>(
         report(new LightningSocketError(event, message));
       })
       .receive('timeout', () => {
-        if (thisAttempt >= TIMEOUT_RETRY_COUNT) {
+        if (thisAttempt >= timeoutRetryCount) {
           report(new LightningTimeoutError(event));
         } else {
           logger.warn(
             `${runId} event ${event} timed out, will retry (attempt ${
               thisAttempt + 1
-            } of ${TIMEOUT_RETRY_COUNT})`
+            } of ${timeoutRetryCount})`
           );
 
-          const delay =
-            typeof TIMEOUT_RETRY_DELAY === 'string'
-              ? parseInt(typeof TIMEOUT_RETRY_DELAY, 10)
-              : TIMEOUT_RETRY_DELAY;
-
           setTimeout(() => {
             sendEvent<T>(context, event, payload, thisAttempt + 1)
               .then(resolve)
               .catch(reject);
-          }, delay);
+          }, timeoutRetryDelay);
         }
       })
       .receive('ok', resolve);

packages/ws-worker/test/reasons.test.ts

@@ -247,6 +247,9 @@ test('exception: failed to load credential', async (t) => {
 });
 
 test('exception: credential timeout', async (t) => {
+  process.env.WORKER_TIMEOUT_RETRY_DELAY = '1';
+  process.env.WORKER_TIMEOUT_RETRY_COUNT = '5';
+
   const plan = createPlan({
     id: 'aa',
     expression: 'export default [(s) => s]',
@@ -264,6 +267,9 @@ test('exception: credential timeout', async (t) => {
 });
 
 test('kill: timeout', async (t) => {
+  process.env.WORKER_TIMEOUT_RETRY_DELAY = '1';
+  process.env.WORKER_TIMEOUT_RETRY_COUNT = '5';
+
   const plan = createPlan({
     id: 'x',
     expression: 'export default [(s) => { while(true) { } }]',

packages/ws-worker/test/util/send-event.test.ts

@@ -84,9 +84,6 @@ test.serial('should throw if the event is rejected', async (t) => {
 });
 
 test.serial('should throw if the event timesout and retry is 1', async (t) => {
-  process.env.WORKER_TIMEOUT_RETRY_DELAY = '1';
-  process.env.WORKER_TIMEOUT_RETRY_COUNT = '1';
-
   const EVENT_NAME = 'test';
   const channel = mockChannel({
     // No handler so no reply
@@ -104,14 +101,18 @@ test.serial('should throw if the event timesout and retry is 1', async (t) => {
   await t.throwsAsync(() => sendEvent(context, EVENT_NAME, {}), {
     instanceOf: LightningTimeoutError,
   });
+
+  // Check it did not retry at all
+  const events = logger._history.filter(
+    ({ level, message }: any) =>
+      level === 'warn' && /event test timed out/.test(message)
+  );
+  t.is(events.length, 0);
 });
 
 test.serial(
   'should throw after 5 attempts if the event timesout and retry is 5',
   async (t) => {
-    process.env.WORKER_TIMEOUT_RETRY_DELAY = '1';
-    process.env.WORKER_TIMEOUT_RETRY_COUNT = '5';
-
     const EVENT_NAME = 'test';
     const channel = mockChannel({
       // No handler so no reply
@@ -124,6 +125,10 @@ test.serial(
         id: 'x',
       } as any),
       logger,
+      options: {
+        timeoutRetryCount: 5,
+        timeoutRetryDelay: 1,
+      },
     };
 
     await t.throwsAsync(() => sendEvent(context, EVENT_NAME, {}), {
@@ -141,8 +146,6 @@ test.serial(
 test.serial(
   'should pass after 5 attempts if the event timesout and retry is 5',
   async (t) => {
-    process.env.WORKER_TIMEOUT_RETRY_DELAY = '1';
-    process.env.WORKER_TIMEOUT_RETRY_COUNT = '5';
     let count = 0;
 
     const EVENT_NAME = 'test';
@@ -165,6 +168,10 @@ test.serial(
         id: 'x',
       } as any),
       logger,
+      options: {
+        timeoutRetryCount: 5,
+        timeoutRetryDelay: 1,
+      },
     };
 
     const reply = await sendEvent(context, EVENT_NAME, {});