Merge pull request #24 from pufflyai/23-block-node-execution-if-paren…

…t-wasnt-previously-run Block node execution if parent wasn't previously run
pufflyai · Aug 27, 2023 · 428ae85 · 428ae85
2 parents 3d350dc + c8e94d8
commit 428ae85
Show file tree

Hide file tree

Showing 3 changed files with 46 additions and 0 deletions.
diff --git a/packages/@pufflig/ps-chains/src/engines/dataflow/__snapshots__/runFlow.test.ts.snap b/packages/@pufflig/ps-chains/src/engines/dataflow/__snapshots__/runFlow.test.ts.snap
@@ -68,6 +68,16 @@ exports[`correctly map output from one node onto another 1`] = `
 }
 `;
 
+exports[`do not run an executable node unless the parent was run already 1`] = `
+{
+  "n2": {
+    "input": {
+      "data": "",
+    },
+  },
+}
+`;
+
 exports[`ignore inputs from nested unreachable nodes 1`] = `
 {
   "n1": {

diff --git a/packages/@pufflig/ps-chains/src/engines/dataflow/runFlow.test.ts b/packages/@pufflig/ps-chains/src/engines/dataflow/runFlow.test.ts
@@ -349,3 +349,27 @@ test("run nodes connected through execution nodes", async () => {
   expect(onNodeRunError).toHaveBeenCalledTimes(0);
   expect(res).toMatchSnapshot();
 });
+
+/**
+ * (1) => (2) => (3)
+ *         ^
+ */
+test("do not run an executable node unless the parent was run already", async () => {
+  const onNodeInputUpdate = jest.fn();
+  const onNodeRunComplete = jest.fn();
+  const onNodeRunError = jest.fn();
+  const res = await runFlow(
+    simpleExec,
+    "n2",
+    {},
+    {
+      onNodeRunComplete,
+      onNodeInputUpdate,
+      onNodeRunError,
+    }
+  );
+  expect(onNodeInputUpdate).toHaveBeenCalledTimes(1);
+  expect(onNodeRunComplete).toHaveBeenCalledTimes(0);
+  expect(onNodeRunError).toHaveBeenCalledTimes(0);
+  expect(res).toMatchSnapshot();
+});
diff --git a/packages/@pufflig/ps-chains/src/engines/dataflow/runFlow.ts b/packages/@pufflig/ps-chains/src/engines/dataflow/runFlow.ts
@@ -71,6 +71,18 @@ export async function runFlow(flow: Flow, nodeId: string, input: Record<string,
     }
     // ---
 
+    // if the node is executable, only run it if the parent is complete
+    const parentNodeId = Object.values(flow.definition.edges).find(
+      (e) => e.target === nodeId && e.targetHandle.startsWith(executionPrefix)
+    )?.source;
+    const isParentComplete = parentNodeId ? runs[parentNodeId] > 0 : true;
+    const isExecutableNode = !!node.execution?.inputs.length;
+
+    if (!isParentComplete && isExecutableNode) {
+      logger.debug("Parent node is not complete, skip execution");
+      return;
+    }
+
     // resolve variables to use during the run
     const newInput = newState[nodeId]?.input;
     let resolvedInput: Record<string, ParamValue> = {};