Merge pull request #552 from OpenFn/551-edge-condition

Handle initial edge condition and log edge evaluations

.changeset/green-coins-jam.md

@@ -0,0 +1,5 @@
+---
+'@openfn/runtime': patch
+---
+
+Use 'debug' logging for edge condition evaluation reports

integration-tests/worker/CHANGELOG.md

@@ -1,5 +1,14 @@
 # @openfn/integration-tests-worker
 
+## 1.0.27
+
+### Patch Changes
+
+- Updated dependencies [f228fd5]
+ - @openfn/ws-worker@0.4.0
+ - @openfn/engine-multi@0.2.5
+ - @openfn/lightning-mock@1.1.7
+
 ## 1.0.26
 
 ### Patch Changes

integration-tests/worker/package.json

@@ -1,7 +1,7 @@
 {
  "name": "@openfn/integration-tests-worker",
  "private": true,
- "version": "1.0.26",
+ "version": "1.0.27",
  "description": "Lightning WOrker integration tests",
  "author": "Open Function Group <admin@openfn.org>",
  "license": "ISC",

packages/cli/CHANGELOG.md

@@ -1,5 +1,12 @@
 # @openfn/cli
 
+## 0.4.12
+
+### Patch Changes
+
+- Updated dependencies [f228fd5]
+ - @openfn/runtime@0.2.3
+
 ## 0.4.11
 
 ### Patch Changes

packages/cli/package.json

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

packages/engine-multi/CHANGELOG.md

@@ -1,5 +1,12 @@
 # engine-multi
 
+## 0.2.5
+
+### Patch Changes
+
+- Updated dependencies [f228fd5]
+ - @openfn/runtime@0.2.3
+
 ## 0.2.4
 
 ### Patch Changes

packages/engine-multi/package.json

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

packages/lightning-mock/CHANGELOG.md

@@ -1,5 +1,13 @@
 # @openfn/lightning-mock
 
+## 1.1.7
+
+### Patch Changes
+
+- Updated dependencies [f228fd5]
+ - @openfn/runtime@0.2.3
+ - @openfn/engine-multi@0.2.5
+
 ## 1.1.6
 
 ### Patch Changes

packages/lightning-mock/package.json

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

packages/runtime/CHANGELOG.md

@@ -1,5 +1,11 @@
 # @openfn/runtime
 
+## 0.2.3
+
+### Patch Changes
+
+- f228fd5: Add edge evaluation logging to runtime
+
 ## 0.2.2
 
 ### Patch Changes

packages/runtime/package.json

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

packages/runtime/src/execute/job.ts

@@ -10,6 +10,7 @@ import type {
  JobNodeID,
  State,
 } from '../types';
+import { Logger } from '@openfn/logger';
 import { EdgeConditionError } from '../errors';
 import {
  NOTIFY_INIT_COMPLETE,
@@ -43,7 +44,7 @@ const loadState = async (
  return job.state;
 };
 
-const calculateNext = (job: CompiledJobNode, result: any) => {
+const calculateNext = (job: CompiledJobNode, result: any, logger: Logger) => {
  const next: string[] = [];
  if (job.next) {
  for (const nextJobId in job.next) {
@@ -58,11 +59,17 @@ const calculateNext = (job: CompiledJobNode, result: any) => {
  if (typeof edge.condition === 'function') {
  try {
  if (!edge.condition(result)) {
+ logger.debug(
+ `Edge ${edge.condition.toString()} returned false; ${nextJobId} will NOT be executed`
+ );
  continue;
  }
  } catch (e: any) {
  throw new EdgeConditionError(e.message);
  }
+ logger.debug(
+ `Edge ${edge.condition.toString()} returned true; ${nextJobId} will be executed next`
+ );
  }
  }
  next.push(nextJobId);
@@ -136,7 +143,7 @@ const executeJob = async (
  logger.error(`Failed job ${jobId} after ${duration}`);
  report(state, jobId, error);
 
- next = calculateNext(job, result);
+ next = calculateNext(job, result, logger);
 
  notify(NOTIFY_JOB_ERROR, {
  duration: Date.now() - startTime,
@@ -175,7 +182,7 @@ const executeJob = async (
  `Final memory usage: [job ${humanJobMemory}mb] [system ${humanSystemMemory}mb]`
  );
 
- next = calculateNext(job, result);
+ next = calculateNext(job, result, logger);
  notify(NOTIFY_JOB_COMPLETE, {
  duration: Date.now() - duration,
  state: result,
@@ -189,7 +196,7 @@ const executeJob = async (
  }
  } else {
  // calculate next for trigger nodes
- next = calculateNext(job, result);
+ next = calculateNext(job, result, logger);
  }
 
  if (next.length && !didError && !result) {

packages/ws-worker/CHANGELOG.md

@@ -1,5 +1,17 @@
 # ws-worker
 
+## 0.4.0
+
+### Minor Changes
+
+- f228fd5: Add support for initial edge conditions in worker
+
+### Patch Changes
+
+- Updated dependencies [f228fd5]
+ - @openfn/runtime@0.2.3
+ - @openfn/engine-multi@0.2.5
+
 ## 0.3.2
 
 ### Patch Changes

packages/ws-worker/package.json

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

packages/ws-worker/src/util/convert-attempt.ts

@@ -25,6 +25,14 @@ const mapEdgeCondition = (edge: Edge) => {
  return condition;
 };
 
+const mapTriggerEdgeCondition = (edge: Edge) => {
+ const { condition } = edge;
+ // This handles cron triggers with undefined conditions and the 'always' string.
+ if (condition === undefined || condition === 'always') return true;
+ // Otherwise, we will return the condition and assume it's a valid JS expression.
+ return condition;
+};
+
 export default (
  attempt: Attempt
 ): { plan: ExecutionPlan; options: AttemptOptions } => {
@@ -63,7 +71,7 @@ export default (
  nodes[id].next = connectedEdges.reduce((obj, edge) => {
  if (edge.enabled !== false) {
  // @ts-ignore
- obj[edge.target_job_id] = true;
+ obj[edge.target_job_id] = mapTriggerEdgeCondition(edge);
  }
  return obj;
  }, {});