Merge pull request #490 from OpenFn/release/next

Release: worker 0.2.7

integration-tests/worker/CHANGELOG.md

@@ -1,5 +1,16 @@
 # @openfn/integration-tests-worker
 
+## 1.0.17
+
+### Patch Changes
+
+- Updated dependencies [d542aa9]
+- Updated dependencies [793d523]
+- Updated dependencies [f17fb4a]
+  - @openfn/ws-worker@0.2.7
+  - @openfn/engine-multi@0.1.11
+  - @openfn/lightning-mock@1.0.12
+
 ## 1.0.16
 
 ### Patch Changes

integration-tests/worker/package.json

@@ -1,7 +1,7 @@
 {
   "name": "@openfn/integration-tests-worker",
   "private": true,
-  "version": "1.0.16",
+  "version": "1.0.17",
   "description": "Lightning WOrker integration tests",
   "author": "Open Function Group <[email protected]>",
   "license": "ISC",

integration-tests/worker/test/exit-reasons.test.ts

@@ -53,3 +53,26 @@ test('crash: syntax error', async (t) => {
   t.is(error_type, 'CompileError');
   t.regex(error_message, /Unexpected token \(1:9\)$/);
 });
+
+test('exception: autoinstall error', async (t) => {
+  const attempt = {
+    id: crypto.randomUUID(),
+    jobs: [
+      {
+        adaptor: '@openfn/[email protected]',
+        body: 'fn((s) => s)',
+      },
+    ],
+  };
+
+  const result = await run(attempt);
+
+  const { reason, error_type, error_message } = result;
+
+  t.is(reason, 'exception');
+  t.is(error_type, 'AutoinstallError');
+  t.regex(
+    error_message,
+    /Error installing @openfn\/[email protected]/
+  );
+});

packages/cli/CHANGELOG.md

@@ -1,5 +1,12 @@
 # @openfn/cli
 
+## 0.4.8
+
+### Patch Changes
+
+- Updated dependencies [857c42b]
+  - @openfn/runtime@0.1.4
+
 ## 0.4.7
 
 ### Patch Changes

packages/cli/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@openfn/cli",
-  "version": "0.4.7",
+  "version": "0.4.8",
   "description": "CLI devtools for the openfn toolchain.",
   "engines": {
     "node": ">=18",

packages/engine-multi/CHANGELOG.md

@@ -1,5 +1,14 @@
 # engine-multi
 
+## 0.1.11
+
+### Patch Changes
+
+- 793d523: Updated purge strategy
+- f17fb4a: Better error handling in autoinstall
+- Updated dependencies [857c42b]
+  - @openfn/runtime@0.1.4
+
 ## 0.1.10
 
 ### Patch Changes

packages/engine-multi/package.json

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

packages/engine-multi/src/api/autoinstall.ts

@@ -12,6 +12,7 @@ import { install as runtimeInstall } from '@openfn/runtime';
 import type { Logger } from '@openfn/logger';
 import type { ExecutionContext } from '../types';
 import { AUTOINSTALL_COMPLETE, AUTOINSTALL_ERROR } from '../events';
+import { AutoinstallError } from '../errors';
 
 // none of these options should be on the plan actually
 export type AutoinstallOptions = {
@@ -54,6 +55,7 @@ const autoinstall = async (context: ExecutionContext): Promise<ModulePaths> => {
   // TODO would rather do all this in parallel but this is fine for now
   // TODO set iteration is weirdly difficult?
   const paths: ModulePaths = {};
+
   for (const a of adaptors) {
     // Ensure that this is not blacklisted
     // TODO what if it is? For now we'll log and skip it
@@ -71,8 +73,6 @@ const autoinstall = async (context: ExecutionContext): Promise<ModulePaths> => {
     const needsInstalling = !(await isInstalledFn(a, repoDir, logger));
     if (needsInstalling) {
       if (!pending[a]) {
-        // TODO because autoinstall can take a while, we should emit that we're starting
-        // add a promise to the pending array
         const startTime = Date.now();
         pending[a] = installFn(a, repoDir, logger)
           .then(() => {
@@ -86,15 +86,26 @@ const autoinstall = async (context: ExecutionContext): Promise<ModulePaths> => {
             });
             delete pending[a];
           })
-          .catch((e) => {
+          .catch((e: any) => {
+            delete pending[a];
+
+            logger.error(`ERROR autoinstalling ${a}: ${e.message}`);
+            logger.error(e);
             const duration = Date.now() - startTime;
             context.emit(AUTOINSTALL_ERROR, {
               module: name,
               version: version!,
               duration,
               message: e.message || e.toString(),
             });
+
+            // wrap and re-throw the error
+            throw new AutoinstallError(a, e);
           });
+      } else {
+        logger.info(
+          `autoinstall waiting for previous promise for ${a} to resolve...`
+        );
       }
       // Return the pending promise (safe to do this multiple times)
       // TODO if this is a chained promise, emit something like "using cache for ${name}"

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

@@ -23,15 +23,15 @@ type WorkerOptions = {
 
 // Adds a `callWorker` function to the API object, which will execute a task in a worker
 export default function initWorkers(
-  api: EngineAPI,
+  engine: EngineAPI,
   workerPath: string,
   options: WorkerOptions = {},
   logger?: Logger
 ) {
   // TODO can we verify the worker path and throw if it's invalid?
   // workerpool won't complain if we give it a nonsense path
   const workers = createWorkers(workerPath, options);
-  api.callWorker = (
+  engine.callWorker = (
     task: string,
     args: any[] = [],
     events: any = {},
@@ -48,22 +48,20 @@ export default function initWorkers(
       promise.timeout(timeout);
     }
 
-    if (options.purge) {
-      promise.then(() => {
-        const { pendingTasks } = workers.stats();
-        if (pendingTasks == 0) {
-          logger?.debug('Purging workers');
-          api.emit(PURGE);
-          workers.terminate();
-        }
-      });
-    }
-
     return promise;
   };
 
+  engine.purge = () => {
+    const { pendingTasks } = workers.stats();
+    if (pendingTasks == 0) {
+      logger?.debug('Purging workers');
+      engine.emit(PURGE);
+      workers.terminate();
+    }
+  };
+
   // This will force termination instantly
-  api.closeWorkers = () => {
+  engine.closeWorkers = () => {
     workers.terminate(true);
 
     // Defer the return to allow workerpool to close down

packages/engine-multi/src/api/execute.ts

@@ -99,11 +99,11 @@ const execute = async (context: ExecutionContext) => {
       error(context, { workflowId: state.plan.id, error: e });
       logger.error(e);
     });
-  } catch (e) {
-    // generic error wrapper
-    // this will catch anything you!
-    const wrappedError = new ExecutionError(e);
-    error(context, { workflowId: state.plan.id, error: wrappedError });
+  } catch (e: any) {
+    if (!e.severity) {
+      e = new ExecutionError(e);
+    }
+    error(context, { workflowId: state.plan.id, error: e });
   }
 };
 

packages/engine-multi/src/api/lifecycle.ts

@@ -135,12 +135,18 @@ export const log = (
   });
 };
 
-export const error = (context: ExecutionContext, event: any) => {
+export const error = (
+  context: ExecutionContext,
+  event: internalEvents.ErrorEvent
+) => {
   const { threadId = '-', error } = event;
 
   context.emit(externalEvents.WORKFLOW_ERROR, {
     threadId,
+    // @ts-ignore
     type: error.type || error.name || 'ERROR',
     message: error.message || error.toString(),
+    // default to exception because if we don't know, it's our fault
+    severity: error.severity || 'exception',
   });
 };

packages/engine-multi/src/engine.ts

@@ -98,11 +98,6 @@ const createEngine = async (options: EngineOptions, workerPath?: string) => {
   const contexts: Record<string, ExecutionContext> = {};
   const deferredListeners: Record<string, Record<string, EventHandler>[]> = {};
 
-  // TODO I think this is for later
-  //const activeWorkflows: string[] = [];
-
-  // TOOD I wonder if the engine should a) always accept a worker path
-  // and b) validate it before it runs
   let resolvedWorkerPath;
   if (workerPath) {
     // If a path to the worker has been passed in, just use it verbatim
@@ -183,14 +178,17 @@ const createEngine = async (options: EngineOptions, workerPath?: string) => {
       delete deferredListeners[workflowId];
     }
 
-    // execute(context);
-
     // Run the execute on a timeout so that consumers have a chance
     // to register listeners
     setTimeout(() => {
       // TODO typing between the class and interface isn't right
       // @ts-ignore
-      execute(context);
+      execute(context).finally(() => {
+        delete contexts[workflowId];
+        if (options.purge && Object.keys(contexts).length === 0) {
+          engine.purge?.();
+        }
+      });
     }, 1);
 
     // hmm. Am I happy to pass the internal workflow state OUT of the handler?
@@ -202,7 +200,6 @@ const createEngine = async (options: EngineOptions, workerPath?: string) => {
         context.once(evt, fn),
       off: (evt: string, fn: (...args: any[]) => void) => context.off(evt, fn),
     };
-    // return context;
   };
 
   const listen = (

packages/engine-multi/src/errors.ts

@@ -57,5 +57,17 @@ export class CompileError extends EngineError {
   }
 }
 
-// Autoinstall Error (exception)
+export class AutoinstallError extends EngineError {
+  severity = 'exception'; // Syntax errors are crashes, but what if we get a module resolution thing?
+  type = 'AutoinstallError';
+  name = 'AutoinstallError';
+  message;
+
+  constructor(specifier: string, error: any) {
+    super();
+
+    this.message = `Error installing ${specifier}: ${error.message}`;
+  }
+}
+
 // CredentialsError (exception)

packages/engine-multi/src/events.ts

@@ -60,6 +60,7 @@ export interface WorkflowCompletePayload extends ExternalEvent {
 export interface WorkflowErrorPayload extends ExternalEvent {
   type: string;
   message: string;
+  severity: string;
 }
 
 export interface JobStartPayload extends ExternalEvent {

packages/engine-multi/src/types.ts

@@ -64,6 +64,7 @@ export interface ExecutionContext extends EventEmitter {
 export interface EngineAPI extends EventEmitter {
   callWorker: CallWorker;
   closeWorkers: () => void;
+  purge?: () => void;
 }
 
 export interface RuntimeEngine extends EventEmitter {

packages/engine-multi/src/worker/events.ts

@@ -59,8 +59,9 @@ export interface LogEvent extends InternalEvent {
   message: JSONLog;
 }
 
-export interface ErrorEvent extends InternalEvent {
-  jobId?: string;
+export interface ErrorEvent {
+  threadId?: string;
+  workflowId?: string;
   error: {
     message: string;
     type: string;