From c25c613a04a6773fa4014d9a0d290e443bcabbe0 Mon Sep 17 00:00:00 2001 From: Alex Grozav Date: Wed, 29 Jan 2025 13:38:24 +0200 Subject: [PATCH] feat(editor): Always keep at least one executing node indicator in the workflow (#12829) --- .../src/composables/useExecutingNode.test.ts | 109 ++++++++++++++++++ .../src/composables/useExecutingNode.ts | 52 +++++++++ .../editor-ui/src/stores/workflows.store.ts | 15 +-- 3 files changed, 166 insertions(+), 10 deletions(-) create mode 100644 packages/editor-ui/src/composables/useExecutingNode.test.ts create mode 100644 packages/editor-ui/src/composables/useExecutingNode.ts diff --git a/packages/editor-ui/src/composables/useExecutingNode.test.ts b/packages/editor-ui/src/composables/useExecutingNode.test.ts new file mode 100644 index 0000000000..d844e66cdd --- /dev/null +++ b/packages/editor-ui/src/composables/useExecutingNode.test.ts @@ -0,0 +1,109 @@ +import { useExecutingNode } from '@/composables/useExecutingNode'; + +describe('useExecutingNode', () => { + it('should always have at least one executing node during execution', () => { + const { executingNode, executingNodeCompletionQueue, addExecutingNode, removeExecutingNode } = + useExecutingNode(); + + addExecutingNode('node1'); + + expect(executingNode.value).toEqual(['node1']); + expect(executingNodeCompletionQueue.value).toEqual([]); + + addExecutingNode('node2'); + + expect(executingNode.value).toEqual(['node1', 'node2']); + expect(executingNodeCompletionQueue.value).toEqual([]); + + addExecutingNode('node3'); + + expect(executingNode.value).toEqual(['node1', 'node2', 'node3']); + expect(executingNodeCompletionQueue.value).toEqual([]); + + removeExecutingNode('node1'); + + expect(executingNode.value).toEqual(['node2', 'node3']); + expect(executingNodeCompletionQueue.value).toEqual([]); + + removeExecutingNode('node2'); + + expect(executingNode.value).toEqual(['node3']); + expect(executingNodeCompletionQueue.value).toEqual([]); + + removeExecutingNode('node3'); + + expect(executingNode.value).toEqual(['node3']); + expect(executingNodeCompletionQueue.value).toEqual(['node3']); + + addExecutingNode('node4'); + + expect(executingNode.value).toEqual(['node4']); + expect(executingNodeCompletionQueue.value).toEqual([]); + }); + + describe('resolveNodeExecutionQueue', () => { + it('should clear all nodes from the execution queue', () => { + const { executingNode, executingNodeCompletionQueue, resolveNodeExecutionQueue } = + useExecutingNode(); + + executingNode.value = ['node1', 'node2']; + executingNodeCompletionQueue.value = ['node1', 'node2']; + + resolveNodeExecutionQueue(); + + expect(executingNode.value).toEqual([]); + expect(executingNodeCompletionQueue.value).toEqual([]); + }); + + it('should keep the last executing node if keepLastInQueue is true and only one node is executing', () => { + const { executingNode, executingNodeCompletionQueue, resolveNodeExecutionQueue } = + useExecutingNode(); + executingNode.value = ['node1']; + executingNodeCompletionQueue.value = ['node1']; + + resolveNodeExecutionQueue(true); + + expect(executingNode.value).toEqual(['node1']); + expect(executingNodeCompletionQueue.value).toEqual(['node1']); + }); + + it('should remove all nodes except the last one if keepLastInQueue is true and more than one node is executing', () => { + const { executingNode, executingNodeCompletionQueue, resolveNodeExecutionQueue } = + useExecutingNode(); + + executingNode.value = ['node1', 'node2']; + executingNodeCompletionQueue.value = ['node1', 'node2']; + + resolveNodeExecutionQueue(true); + + expect(executingNode.value).toEqual(['node2']); + expect(executingNodeCompletionQueue.value).toEqual(['node2']); + }); + + it('should clear all nodes if keepLastInQueue is false', () => { + const { executingNode, executingNodeCompletionQueue, resolveNodeExecutionQueue } = + useExecutingNode(); + + executingNode.value = ['node1', 'node2']; + executingNodeCompletionQueue.value = ['node1', 'node2']; + + resolveNodeExecutionQueue(false); + + expect(executingNode.value).toEqual([]); + expect(executingNodeCompletionQueue.value).toEqual([]); + }); + + it('should handle empty execution queue gracefully', () => { + const { executingNode, executingNodeCompletionQueue, resolveNodeExecutionQueue } = + useExecutingNode(); + + executingNode.value = []; + executingNodeCompletionQueue.value = []; + + resolveNodeExecutionQueue(); + + expect(executingNode.value).toEqual([]); + expect(executingNodeCompletionQueue.value).toEqual([]); + }); + }); +}); diff --git a/packages/editor-ui/src/composables/useExecutingNode.ts b/packages/editor-ui/src/composables/useExecutingNode.ts new file mode 100644 index 0000000000..fd037c2dba --- /dev/null +++ b/packages/editor-ui/src/composables/useExecutingNode.ts @@ -0,0 +1,52 @@ +import { ref } from 'vue'; + +/** + * Composable to keep track of the currently executing node. + * The queue is used to keep track of the order in which nodes are completed and + * to ensure that there's always at least one node in the executing queue. + * + * The completion queue serves as a workaround for the fact that the execution status of a node + * is not updated in real-time when dealing with large amounts of data, meaning we can end up in a + * state where no node is actively executing, even though the workflow execution is not completed. + */ +export function useExecutingNode() { + const executingNode = ref([]); + const executingNodeCompletionQueue = ref([]); + + function addExecutingNode(nodeName: string) { + resolveNodeExecutionQueue(); + executingNode.value.push(nodeName); + } + + function removeExecutingNode(nodeName: string) { + executingNodeCompletionQueue.value.push(nodeName); + resolveNodeExecutionQueue( + executingNode.value.length <= executingNodeCompletionQueue.value.length, + ); + } + + function resolveNodeExecutionQueue(keepLastInQueue = false) { + const lastExecutingNode = executingNodeCompletionQueue.value.at(-1); + const nodesToRemove = keepLastInQueue + ? executingNodeCompletionQueue.value.slice(0, -1) + : executingNodeCompletionQueue.value; + + executingNode.value = executingNode.value.filter((name) => !nodesToRemove.includes(name)); + executingNodeCompletionQueue.value = + keepLastInQueue && lastExecutingNode ? [lastExecutingNode] : []; + } + + function clearNodeExecutionQueue() { + executingNode.value = []; + executingNodeCompletionQueue.value = []; + } + + return { + executingNode, + executingNodeCompletionQueue, + addExecutingNode, + removeExecutingNode, + resolveNodeExecutionQueue, + clearNodeExecutionQueue, + }; +} diff --git a/packages/editor-ui/src/stores/workflows.store.ts b/packages/editor-ui/src/stores/workflows.store.ts index 684e0ba4c9..ae3b142a57 100644 --- a/packages/editor-ui/src/stores/workflows.store.ts +++ b/packages/editor-ui/src/stores/workflows.store.ts @@ -89,6 +89,7 @@ import { clearPopupWindowState, openFormPopupWindow } from '@/utils/executionUti import { useNodeHelpers } from '@/composables/useNodeHelpers'; import { useUsersStore } from '@/stores/users.store'; import { updateCurrentUserSettings } from '@/api/users'; +import { useExecutingNode } from '@/composables/useExecutingNode'; const defaults: Omit & { settings: NonNullable } = { name: '', @@ -140,7 +141,6 @@ export const useWorkflowsStore = defineStore(STORES.WORKFLOWS, () => { const activeExecutionId = ref(null); const subWorkflowExecutionError = ref(null); const executionWaitingForWebhook = ref(false); - const executingNode = ref([]); const workflowsById = ref>({}); const nodeMetadata = ref({}); const isInDebugMode = ref(false); @@ -149,6 +149,9 @@ export const useWorkflowsStore = defineStore(STORES.WORKFLOWS, () => { const isChatPanelOpen = ref(false); const isLogsPanelOpen = ref(false); + const { executingNode, addExecutingNode, removeExecutingNode, clearNodeExecutionQueue } = + useExecutingNode(); + const workflowName = computed(() => workflow.value.name); const workflowId = computed(() => workflow.value.id); @@ -552,14 +555,6 @@ export const useWorkflowsStore = defineStore(STORES.WORKFLOWS, () => { executionWaitingForWebhook.value = false; } - function addExecutingNode(nodeName: string) { - executingNode.value.push(nodeName); - } - - function removeExecutingNode(nodeName: string) { - executingNode.value = executingNode.value.filter((name) => name !== nodeName); - } - function setWorkflowId(id?: string) { workflow.value.id = !id || id === 'new' ? PLACEHOLDER_EMPTY_WORKFLOW_ID : id; } @@ -1604,7 +1599,7 @@ export const useWorkflowsStore = defineStore(STORES.WORKFLOWS, () => { function markExecutionAsStopped() { activeExecutionId.value = null; - executingNode.value.length = 0; + clearNodeExecutionQueue(); executionWaitingForWebhook.value = false; uiStore.removeActiveAction('workflowRunning'); workflowHelpers.setDocumentTitle(workflowName.value, 'IDLE');