Merge pull request #1088 from OpenFn/fix-worker-validation-timeout-1016

Ameliorate crash-loop during worker startup

Author: josephjclark

packages/engine-multi/CHANGELOG.md

@@ -1,5 +1,11 @@
 # engine-multi
 
+## 1.7.0
+
+### Minor Changes
+
+- cf8d3c2: Add retry with backoff to worker validation
+
 ## 1.6.14
 
 ### Patch Changes

packages/engine-multi/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@openfn/engine-multi",
-  "version": "1.6.14",
+  "version": "1.7.0",
   "description": "Multi-process runtime engine",
   "main": "dist/index.js",
   "type": "module",

packages/engine-multi/src/api/validate-worker.ts

@@ -1,16 +1,45 @@
-// TODO let me deal with the fallout first
-
+import { Logger } from '@openfn/logger';
 import { EngineAPI } from '../types';
 
-// Simple vaidation function to ensure that a worker is loaded
+class WorkerValidationError extends Error {
+  name = 'WorkerValidationError';
+  constructor(message: string) {
+    super(message);
+  }
+}
+
+// Simple validation function to ensure that a worker is loaded
 // Call a handshake task in a worker thread
-// This really jsut validates that the worker path exists
+// Validates that the worker
+// Timeout and retry logic to help on containerized environments
+// (e.g. Kubernetes) with filesystem boot delays
+export default async (
+  api: EngineAPI,
+  logger: Logger,
+  options: { timeout?: number; retries?: number } = {}
+): Promise<void> => {
+  const { timeout = Infinity, retries = 1 } = options;
+  let start = Date.now();
+  for (let i = 0; i < retries; i++) {
+    try {
+      // TODO argument drive this
+      await api.callWorker('handshake', [], {}, { timeout });
+      const duration = Date.now() - start;
+      logger.debug(`Engine worker validated in ${duration}ms`);
+      return;
+    } catch (e) {
+      if (i >= retries - 1) {
+        const duration = Date.now() - start;
+        logger.error(`Engine worker validation failed in in ${duration}ms`);
+        throw new WorkerValidationError(`Failed to validate worker thread. This likely happened because:
+1. An invalid worker file was passed to the engine
+2. The filesystem was temporarily unavailable`);
+      }
 
-export default async (api: EngineAPI, timeout = 5000) => {
-  try {
-    // TODO argument drive this
-    await api.callWorker('handshake', [], {}, { timeout });
-  } catch (e) {
-    throw new Error('Invalid worker path');
+      // exponential backoff: 1s, 2s, 4s, etc.
+      const backoffMs = 1000 * Math.pow(2, i);
+      logger.warn(`Worker validation failed: will retry in ${backoffMs}ms`);
+      await new Promise((resolve) => setTimeout(resolve, backoffMs));
+    }
   }
 };

packages/engine-multi/src/engine.ts

@@ -82,6 +82,9 @@ export type EngineOptions = {
   statePropsToRemove?: string[];
   whitelist?: RegExp[];
   proxyStdout?: boolean;
+
+  workerValidationTimeout?: number;
+  workerValidationRetries?: number;
 };
 
 export type InternalEngine = RuntimeEngine & {
@@ -136,13 +139,11 @@ const createEngine = async (
   );
   engine.callWorker = callWorker;
 
-  await validateWorker(engine);
+  await validateWorker(engine, options.logger, {
+    timeout: options.workerValidationTimeout,
+    retries: options.workerValidationRetries,
+  });
 
-  // TODO I think this needs to be like:
-  // take a plan
-  // create, register and return  a state object
-  // should it also load the initial data clip?
-  // when does that happen? No, that's inside execute
   const registerWorkflow = (plan: ExecutionPlan, input: State) => {
     // TODO throw if already registered?
     const state = createState(plan, input);

packages/engine-multi/test/api/validate-worker.test.ts

@@ -7,24 +7,47 @@ import validateWorker from '../../src/api/validate-worker';
 
 const logger = createMockLogger();
 
+test.afterEach(() => {
+  logger._reset();
+});
+
 test('validate should not throw if the worker path is valid', async (t) => {
   const workerPath = path.resolve('dist/test/worker-functions.js');
   const api = initWorkers(workerPath, {}, logger);
-  await t.notThrowsAsync(() => validateWorker(api as any, 500));
+
+  await t.notThrowsAsync(() => validateWorker(api as any, logger));
 });
 
 test('validate should throw if the worker path is invalid', async (t) => {
   const workerPath = 'a/b/c.js';
   const api = initWorkers(workerPath, {}, logger);
-  await t.throwsAsync(() => validateWorker(api as any, 500), {
-    message: 'Invalid worker path',
+
+  await t.throwsAsync(() => validateWorker(api as any, logger), {
+    name: 'WorkerValidationError',
   });
 });
 
 test('validate should throw if the worker does not respond to a handshake', async (t) => {
   const workerPath = path.resolve('src/test/bad-worker.js');
   const api = initWorkers(workerPath, {}, logger);
-  await t.throwsAsync(() => validateWorker(api as any, 500), {
-    message: 'Invalid worker path',
+  const opts = { timeout: 100 };
+
+  await t.throwsAsync(() => validateWorker(api as any, logger, opts), {
+    name: 'WorkerValidationError',
+  });
+});
+
+test('validate should retry with a backoff', async (t) => {
+  const workerPath = path.resolve('src/test/bad-worker.js');
+  const api = initWorkers(workerPath, {}, logger);
+  const opts = { timeout: 10, retries: 3 };
+
+  await t.throwsAsync(() => validateWorker(api as any, logger, opts), {
+    name: 'WorkerValidationError',
+  });
+
+  const warnings = logger._history.filter(([level, _icon, message]) => {
+    return level === 'warn' && message.match(/will retry/);
   });
+  t.is(warnings.length, 2);
 });

packages/lightning-mock/CHANGELOG.md

@@ -1,5 +1,12 @@
 # @openfn/lightning-mock
 
+## 2.3.1
+
+### Patch Changes
+
+- Updated dependencies [cf8d3c2]
+  - @openfn/engine-multi@1.7.0
+
 ## 2.3.0
 
 ### Minor Changes

packages/lightning-mock/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@openfn/lightning-mock",
-  "version": "2.3.0",
+  "version": "2.3.1",
   "private": true,
   "description": "A mock Lightning server",
   "main": "dist/index.js",

packages/ws-worker/CHANGELOG.md

@@ -1,5 +1,16 @@
 # ws-worker
 
+## 1.18.0
+
+### Minor Changes
+
+- cf8d3c2: Add retry with backoff to worker validation
+
+### Patch Changes
+
+- Updated dependencies [cf8d3c2]
+  - @openfn/engine-multi@1.7.0
+
 ## 1.17.0
 
 ### Minor Changes

packages/ws-worker/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@openfn/ws-worker",
-  "version": "1.17.0",
+  "version": "1.18.0",
   "description": "A Websocket Worker to connect Lightning to a Runtime Engine",
   "main": "dist/index.js",
   "type": "module",

packages/ws-worker/src/start.ts

@@ -91,6 +91,8 @@ if (args.mock) {
     maxWorkers: args.capacity,
     statePropsToRemove: args.statePropsToRemove,
     runTimeoutMs: args.maxRunDurationSeconds * 1000,
+    workerValidationTimeout: args.engineValidationTimeoutMs,
+    workerValidationRetries: args.engineValidationRetries,
   };
   logger.debug('Creating runtime engine...');
   logger.debug('Engine options:', engineOptions);