Runtime: warn when an expression doesn't return state (#832)

* Runtime: warn when an expression doesn't return state

* test: fix flaky test

* feat: update logs wording

* test: allow test for undefined at step level

* test: warn when an operation does not return state

* refactor: move null-state check to begining of prepare-final-state

* Runtime: update changelog

* versoins: [email protected] [email protected]

---------

Co-authored-by: Joe Clark <[email protected]>

Author: doc-han

integration-tests/execute/CHANGELOG.md

@@ -1,5 +1,12 @@
 # @openfn/integration-tests-execute
 
+## 1.0.9
+
+### Patch Changes
+
+- Updated dependencies [1cbbba0]
+  - @openfn/runtime@1.5.3
+
 ## 1.0.8
 
 ### Patch Changes

integration-tests/execute/package.json

@@ -1,7 +1,7 @@
 {
   "name": "@openfn/integration-tests-execute",
   "private": true,
-  "version": "1.0.8",
+  "version": "1.0.9",
   "description": "Job execution tests",
   "author": "Open Function Group <[email protected]>",
   "license": "ISC",

integration-tests/worker/CHANGELOG.md

@@ -1,5 +1,13 @@
 # @openfn/integration-tests-worker
 
+## 1.0.67
+
+### Patch Changes
+
+- @openfn/engine-multi@1.4.3
+- @openfn/lightning-mock@2.0.24
+- @openfn/ws-worker@1.8.4
+
 ## 1.0.66
 
 ### Patch Changes

integration-tests/worker/package.json

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

packages/cli/CHANGELOG.md

@@ -1,5 +1,13 @@
 # @openfn/cli
 
+## 1.8.10
+
+### Patch Changes
+
+- Warn when an expression doesn't return state
+- Updated dependencies [1cbbba0]
+  - @openfn/runtime@1.5.3
+
 ## 1.8.9
 
 ### Patch Changes

packages/cli/package.json

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

packages/engine-multi/CHANGELOG.md

@@ -1,5 +1,12 @@
 # engine-multi
 
+## 1.4.3
+
+### Patch Changes
+
+- Updated dependencies [1cbbba0]
+  - @openfn/runtime@1.5.3
+
 ## 1.4.2
 
 ### Patch Changes

packages/engine-multi/package.json

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

packages/engine-multi/test/integration.test.ts

@@ -151,19 +151,20 @@ test.serial('trigger workflow-log for job logs', (t) => {
 
     let didLog = false;
 
-    api.execute(plan, emptyState).on('workflow-log', (evt) => {
-      if (evt.name === 'JOB') {
-        didLog = true;
-        t.deepEqual(evt.message, JSON.stringify(['hola']));
-        t.pass('workflow logged');
-      }
-    });
-
-    api.execute(plan, emptyState).on('workflow-complete', (evt) => {
-      t.true(didLog);
-      t.falsy(evt.state.errors);
-      done();
-    });
+    api
+      .execute(plan, emptyState)
+      .on('workflow-log', (evt) => {
+        if (evt.name === 'JOB') {
+          didLog = true;
+          t.deepEqual(evt.message, JSON.stringify(['hola']));
+          t.pass('workflow logged');
+        }
+      })
+      .on('workflow-complete', (evt) => {
+        t.true(didLog);
+        t.falsy(evt.state.errors);
+        done();
+      });
   });
 });
 

packages/lightning-mock/CHANGELOG.md

@@ -1,5 +1,13 @@
 # @openfn/lightning-mock
 
+## 2.0.24
+
+### Patch Changes
+
+- Updated dependencies [1cbbba0]
+  - @openfn/runtime@1.5.3
+  - @openfn/engine-multi@1.4.3
+
 ## 2.0.23
 
 ### Patch Changes

packages/lightning-mock/package.json

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

packages/runtime/CHANGELOG.md

@@ -1,5 +1,11 @@
 # @openfn/runtime
 
+## 1.5.3
+
+### Patch Changes
+
+- 1cbbba0: warn when an expression doesn't return state
+
 ## 1.5.2
 
 ### Patch Changes

packages/runtime/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@openfn/runtime",
-  "version": "1.5.2",
+  "version": "1.5.3",
   "description": "Job processing runtime.",
   "type": "module",
   "exports": {

packages/runtime/src/execute/expression.ts

@@ -18,6 +18,11 @@ import {
 } from '../errors';
 import type { JobModule, ExecutionContext } from '../types';
 import { ModuleInfoMap } from '../modules/linker';
+import {
+  clearNullState,
+  isNullState,
+  createNullState,
+} from '../util/null-state';
 
 export type ExecutionErrorWrapper = {
   state: any;
@@ -105,8 +110,21 @@ export const wrapOperation = (
   return async (state: State) => {
     logger.debug(`Starting operation ${name}`);
     const start = new Date().getTime();
+    if (isNullState(state)) {
+      clearNullState(state);
+      logger.warn(
+        `WARNING: No state was passed into operation ${name}. Did the previous operation return state?`
+      );
+    }
     const newState = immutableState ? clone(state) : state;
-    const result = await fn(newState);
+
+    let result = await fn(newState);
+
+    if (!result) {
+      logger.debug(`Warning: operation ${name} did not return state`);
+      result = createNullState();
+    }
+
     // TODO should we warn if an operation does not return state?
     // the trick is saying WHICH operation without source mapping
     const duration = printDuration(new Date().getTime() - start);

packages/runtime/src/execute/step.ts

@@ -15,7 +15,7 @@ import {
   NOTIFY_JOB_ERROR,
   NOTIFY_JOB_START,
 } from '../events';
-import stringify from 'fast-safe-stringify';
+import { isNullState } from '../util/null-state';
 
 const loadCredentials = async (
   job: Job,
@@ -80,6 +80,7 @@ const prepareFinalState = (
   logger: Logger,
   statePropsToRemove?: string[]
 ) => {
+  if (isNullState(state)) return undefined;
   if (state) {
     if (!statePropsToRemove) {
       // As a strict default, remove the configuration key
@@ -94,12 +95,12 @@ const prepareFinalState = (
         removedProps.push(prop);
       }
     });
-    logger.debug(
-      `Cleaning up state. Removing keys: ${removedProps.join(', ')}`
-    );
+    if (removedProps.length)
+      logger.debug(
+        `Cleaning up state. Removing keys: ${removedProps.join(', ')}`
+      );
 
-    const cleanState = stringify(state);
-    return JSON.parse(cleanState);
+    return clone(state);
   }
   return state;
 };

packages/runtime/src/util/null-state.ts

@@ -0,0 +1,18 @@
+// This module manages a special state object with a hidden null symbol.
+// Used to track when operations and jobs do not return their own state
+
+const NULL_STATE = Symbol('null_state');
+
+// The good thing about using a Symbol is that even if we forget to clean the object.
+// it's still represented as {}, because symbols aren't visible as keys
+export function createNullState() {
+  return { [NULL_STATE]: true };
+}
+
+export function isNullState(state: any) {
+  return typeof state === 'object' && state[NULL_STATE] === true;
+}
+
+export function clearNullState(state: any) {
+  if (typeof state === 'object') delete state[NULL_STATE];
+}

packages/runtime/test/execute/expression.test.ts

@@ -5,6 +5,7 @@ import type { Operation, State } from '@openfn/lexicon';
 
 import execute, { mergeLinkerOptions } from '../../src/execute/expression';
 import type { ExecutionContext } from '../../src/types';
+import { isNullState } from '../../src/util/null-state';
 
 type TestState = State & {
   data: {
@@ -139,16 +140,43 @@ test.serial('async operations run in series', async (t) => {
   t.is(result.data.x, 12);
 });
 
-test.serial('jobs can return undefined', async (t) => {
+test.serial(
+  'jobs return null-state instead of undefined or null',
+  async (t) => {
+    // @ts-ignore violating the operation contract here
+    const job = [() => undefined] as Operation[];
+
+    const state = createState() as TestState;
+    const context = createContext();
+
+    const result = (await execute(context, job, state, {})) as TestState;
+
+    t.assert(isNullState(result));
+  }
+);
+
+test.serial('warn when an operation does not return state', async (t) => {
   // @ts-ignore violating the operation contract here
-  const job = [() => undefined] as Operation[];
+  const job = [
+    (s) => s,
+    () => {},
+    (s) => {
+      s.data = { a: 'a' };
+      return s;
+    },
+  ] as Operation[];
 
   const state = createState() as TestState;
   const context = createContext();
 
   const result = (await execute(context, job, state, {})) as TestState;
+  t.deepEqual(result, { data: { a: 'a' } });
+
+  const debugLog = logger._find('debug', /did not return state/);
+  t.truthy(debugLog);
 
-  t.assert(result === undefined);
+  const warningLog = logger._find('warn', /No state was passed into operation/);
+  t.truthy(warningLog);
 });
 
 test.serial('jobs can mutate the original state', async (t) => {

packages/runtime/test/security.test.ts

@@ -1,7 +1,7 @@
 // a suite of tests with various security concerns in mind
 import test from 'ava';
 import { createMockLogger } from '@openfn/logger';
-import type { ExecutionPlan, State } from '@openfn/lexicon';
+import type { ExecutionPlan } from '@openfn/lexicon';
 
 import run from '../src/runtime';
 
@@ -64,12 +64,12 @@ test.serial(
 );
 
 test.serial('jobs should not have access to global scope', async (t) => {
-  const src = 'export default [() => globalThis.x]';
+  const src = 'export default [() => ({x: globalThis.x, y: "some-val"})]';
   // @ts-ignore
   globalThis.x = 42;
 
   const result: any = await run(src);
-  t.falsy(result);
+  t.deepEqual(result, { y: 'some-val' });
 
   // @ts-ignore
   delete globalThis.x;
@@ -90,16 +90,17 @@ test.serial('jobs should be able to mutate global state', async (t) => {
 });
 
 test.serial('jobs should each run in their own context', async (t) => {
-  const src1 = 'export default [() => { globalThis.x = 1; return 1;}]';
-  const src2 = 'export default [() => globalThis.x]';
+  const src1 =
+    'export default [() => { globalThis.x = 1; return { x: globalThis.x }}]';
+  const src2 = 'export default [() => { return { x: globalThis.x }}]';
 
   await run(src1);
 
   const r1 = (await run(src1)) as any;
-  t.is(r1, 1);
+  t.deepEqual(r1, { x: 1 });
 
   const r2 = (await run(src2)) as any;
-  t.is(r2, undefined);
+  t.deepEqual(r2, {});
 });
 
 test.serial('jobs should not have a process object', async (t) => {

packages/ws-worker/CHANGELOG.md

@@ -1,5 +1,14 @@
 # ws-worker
 
+## 1.8.4
+
+### Patch Changes
+
+- Warn when an expression doesn't return state
+- Updated dependencies [1cbbba0]
+  - @openfn/runtime@1.5.3
+  - @openfn/engine-multi@1.4.3
+
 ## 1.8.3
 
 ### Patch Changes

packages/ws-worker/package.json

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