From 152cad7c45e132da100b8312b40480ea4f264ad4 Mon Sep 17 00:00:00 2001 From: Guillaume Jacquart Date: Fri, 24 Jul 2026 23:09:34 +0200 Subject: [PATCH] feat(core): Forward sub-workflow node progress to the parent execution --- packages/@n8n/api-types/src/push/execution.ts | 57 ++- .../execution-lifecycle-hooks.test.ts | 325 +++++++++++++++++- .../execution-lifecycle-hooks.ts | 179 +++++++++- .../src/workflow-execute-additional-data.ts | 2 +- 4 files changed, 557 insertions(+), 6 deletions(-) diff --git a/packages/@n8n/api-types/src/push/execution.ts b/packages/@n8n/api-types/src/push/execution.ts index a518a91fa64..a28e5c8353f 100644 --- a/packages/@n8n/api-types/src/push/execution.ts +++ b/packages/@n8n/api-types/src/push/execution.ts @@ -119,6 +119,58 @@ export type NodeExecuteAfterData = { }; }; +/** + * Sent once when a sub-workflow execution begins. Lets the editor display a + * live progress overlay on the parent's "Execute Sub-workflow" node. + */ +export type SubworkflowExecutionStarted = { + type: 'subworkflowExecutionStarted'; + data: { + /** Root execution id whose UI session should display the overlay. */ + parentExecutionId: string; + /** Name of the "Execute Sub-workflow" node in the parent workflow. */ + parentNodeName: string; + /** Child sub-execution id. */ + executionId: string; + /** Total nodes in the child workflow — used to render "X / Y". */ + totalNodes: number; + }; +}; + +/** + * Sent on each child node start/finish during a sub-workflow execution. + * Lightweight (no item data) so it can fire at high frequency. + */ +export type SubworkflowNodeProgress = { + type: 'subworkflowNodeProgress'; + data: { + parentExecutionId: string; + parentNodeName: string; + executionId: string; + /** Currently-running node name in the child workflow. */ + currentNodeName: string; + /** 1-based index of the node in the child workflow's execution order. */ + currentNodeIndex: number; + totalNodes: number; + /** 'running' on nodeExecuteBefore; 'success' | 'error' on nodeExecuteAfter. */ + phase: 'running' | 'success' | 'error'; + }; +}; + +/** + * Sent when a sub-workflow execution reaches a terminal state. The editor uses + * this to clear the per-node progress overlay. + */ +export type SubworkflowExecutionFinished = { + type: 'subworkflowExecutionFinished'; + data: { + parentExecutionId: string; + parentNodeName: string; + executionId: string; + status: ExecutionStatus; + }; +}; + export type ExecutionPushMessage = | ExecutionStarted | ExecutionWaiting @@ -126,4 +178,7 @@ export type ExecutionPushMessage = | ExecutionRecovered | NodeExecuteBefore | NodeExecuteAfter - | NodeExecuteAfterData; + | NodeExecuteAfterData + | SubworkflowExecutionStarted + | SubworkflowNodeProgress + | SubworkflowExecutionFinished; diff --git a/packages/cli/src/execution-lifecycle/__tests__/execution-lifecycle-hooks.test.ts b/packages/cli/src/execution-lifecycle/__tests__/execution-lifecycle-hooks.test.ts index 2b0517f9f91..88b4e09217f 100644 --- a/packages/cli/src/execution-lifecycle/__tests__/execution-lifecycle-hooks.test.ts +++ b/packages/cli/src/execution-lifecycle/__tests__/execution-lifecycle-hooks.test.ts @@ -25,6 +25,7 @@ import type { } from 'n8n-workflow'; import { mock } from 'vitest-mock-extended'; +import { ActiveExecutions } from '@/active-executions'; import { EventService } from '@/events/event.service'; import { ExecutionPersistence } from '@/executions/execution-persistence'; import { ExecutionRedactionServiceProxy } from '@/executions/execution-redaction-proxy.service'; @@ -62,6 +63,7 @@ describe('Execution Lifecycle Hooks', () => { const userRepository = mockInstance(UserRepository); const redactionProxy = mockInstance(ExecutionRedactionServiceProxy); const workflowHookContext = mockInstance(WorkflowHookContextService); + const activeExecutions = mockInstance(ActiveExecutions); const nodeName = 'Test Node'; const nodeType = 'n8n-nodes-base.testNode'; @@ -1656,7 +1658,7 @@ describe('Execution Lifecycle Hooks', () => { executionId, workflowData, undefined, - parentExecution, + { parentExecution }, ); }); @@ -1726,5 +1728,326 @@ describe('Execution Lifecycle Hooks', () => { expect(binaryDataService.duplicateBinaryData).not.toHaveBeenCalled(); }); }); + + describe('subworkflow progress push', () => { + const parentWorkflowId = 'parent-workflow-id'; + const parentExecutionId = 'parent-execution-id'; + const parentExecution = { + workflowId: parentWorkflowId, + executionId: parentExecutionId, + }; + const parentNode: INode = { + id: 'parent-node-id', + name: 'Execute Sub-workflow', + type: 'n8n-nodes-base.executeWorkflow', + typeVersion: 1, + position: [0, 0], + parameters: {}, + }; + const rootPushRef = 'root-push-ref'; + + function buildHooks(childWorkflowData: IWorkflowBase = workflowData) { + return getLifecycleHooksForSubExecutions( + 'integrated', + executionId, + childWorkflowData, + undefined, + { parentExecution, parentNode }, + ); + } + + function stubActiveExecution(id: string, opts: { pushRef?: string; parentId?: string }) { + activeExecutions.has.mockImplementation((requested) => requested === id); + activeExecutions.getExecutionOrFail.mockImplementation((requested) => { + if (requested !== id) throw new Error(`Unexpected execution lookup: ${requested}`); + return { + executionData: { + pushRef: opts.pushRef, + executionData: opts.parentId + ? { parentExecution: { executionId: opts.parentId, workflowId: 'wf' } } + : undefined, + }, + } as never; + }); + } + + beforeEach(() => { + push.send.mockReset(); + activeExecutions.has.mockReset(); + activeExecutions.getExecutionOrFail.mockReset(); + }); + + it('emits subworkflowExecutionStarted on workflowExecuteBefore with the parent pushRef', async () => { + stubActiveExecution(parentExecutionId, { pushRef: rootPushRef }); + const hooks = buildHooks(); + + await hooks.runHook('workflowExecuteBefore', [workflow, runExecutionData]); + + expect(push.send).toHaveBeenCalledWith( + { + type: 'subworkflowExecutionStarted', + data: { + parentExecutionId, + parentNodeName: parentNode.name, + executionId, + totalNodes: workflowData.nodes.length, + }, + }, + rootPushRef, + ); + }); + + it('emits subworkflowNodeProgress with running phase on nodeExecuteBefore', async () => { + stubActiveExecution(parentExecutionId, { pushRef: rootPushRef }); + const hooks = buildHooks(); + + await hooks.runHook('nodeExecuteBefore', [nodeName, taskStartedData]); + + expect(push.send).toHaveBeenCalledWith( + expect.objectContaining({ + type: 'subworkflowNodeProgress', + data: expect.objectContaining({ + parentExecutionId, + parentNodeName: parentNode.name, + executionId, + currentNodeName: nodeName, + currentNodeIndex: 1, + phase: 'running', + }), + }), + rootPushRef, + ); + }); + + it('emits success phase on nodeExecuteAfter for a successful node', async () => { + stubActiveExecution(parentExecutionId, { pushRef: rootPushRef }); + const hooks = buildHooks(); + + const success = mock({ error: undefined }); + await hooks.runHook('nodeExecuteAfter', [nodeName, success, runExecutionData]); + + expect(push.send).toHaveBeenCalledWith( + expect.objectContaining({ + type: 'subworkflowNodeProgress', + data: expect.objectContaining({ phase: 'success' }), + }), + rootPushRef, + ); + }); + + it('emits error phase on nodeExecuteAfter when task data contains an error', async () => { + stubActiveExecution(parentExecutionId, { pushRef: rootPushRef }); + const hooks = buildHooks(); + + const errored = mock({ error: expressionError }); + await hooks.runHook('nodeExecuteAfter', [nodeName, errored, runExecutionData]); + + expect(push.send).toHaveBeenCalledWith( + expect.objectContaining({ + type: 'subworkflowNodeProgress', + data: expect.objectContaining({ phase: 'error' }), + }), + rootPushRef, + ); + }); + + it('emits subworkflowExecutionFinished with the run status', async () => { + stubActiveExecution(parentExecutionId, { pushRef: rootPushRef }); + const hooks = buildHooks(); + + await hooks.runHook('workflowExecuteAfter', [successfulRun, {}]); + + expect(push.send).toHaveBeenCalledWith( + { + type: 'subworkflowExecutionFinished', + data: { + parentExecutionId, + parentNodeName: parentNode.name, + executionId, + status: 'success', + }, + }, + rootPushRef, + ); + }); + + it('does not register push hooks when parent node is omitted', async () => { + const hooks = getLifecycleHooksForSubExecutions( + 'integrated', + executionId, + workflowData, + undefined, + { parentExecution }, + ); + + await hooks.runHook('workflowExecuteBefore', [workflow, runExecutionData]); + await hooks.runHook('nodeExecuteBefore', [nodeName, taskStartedData]); + await hooks.runHook('nodeExecuteAfter', [nodeName, taskData, runExecutionData]); + await hooks.runHook('workflowExecuteAfter', [successfulRun, {}]); + + expect(push.send).not.toHaveBeenCalled(); + }); + + it('does not emit when no ancestor in the chain has a pushRef', async () => { + stubActiveExecution(parentExecutionId, { pushRef: undefined }); + const hooks = buildHooks(); + + await hooks.runHook('workflowExecuteBefore', [workflow, runExecutionData]); + await hooks.runHook('nodeExecuteBefore', [nodeName, taskStartedData]); + await hooks.runHook('nodeExecuteAfter', [nodeName, taskData, runExecutionData]); + await hooks.runHook('workflowExecuteAfter', [successfulRun, {}]); + + expect(push.send).not.toHaveBeenCalled(); + }); + + it('walks up nested sub-execution chains to the root pushRef', async () => { + const grandparentId = 'grandparent-execution-id'; + + activeExecutions.has.mockImplementation((id) => + [parentExecutionId, grandparentId].includes(id), + ); + activeExecutions.getExecutionOrFail.mockImplementation((id) => { + if (id === parentExecutionId) { + return { + executionData: { + pushRef: undefined, + executionData: { + parentExecution: { executionId: grandparentId, workflowId: 'wf' }, + }, + }, + } as never; + } + if (id === grandparentId) { + return { + executionData: { + pushRef: rootPushRef, + executionData: undefined, + }, + } as never; + } + throw new Error(`Unexpected lookup: ${id}`); + }); + + const hooks = buildHooks(); + + await hooks.runHook('workflowExecuteBefore', [workflow, runExecutionData]); + + expect(push.send).toHaveBeenCalledWith( + expect.objectContaining({ type: 'subworkflowExecutionStarted' }), + rootPushRef, + ); + }); + + it('retries pushRef resolution after a transient miss', async () => { + activeExecutions.has.mockReturnValue(false); + const hooks = buildHooks(); + + await hooks.runHook('workflowExecuteBefore', [workflow, runExecutionData]); + expect(push.send).not.toHaveBeenCalled(); + + stubActiveExecution(parentExecutionId, { pushRef: rootPushRef }); + await hooks.runHook('nodeExecuteBefore', [nodeName, taskStartedData]); + + expect(push.send).toHaveBeenCalledWith( + expect.objectContaining({ type: 'subworkflowNodeProgress' }), + rootPushRef, + ); + }); + + it('excludes disabled nodes and sticky notes from totalNodes', async () => { + stubActiveExecution(parentExecutionId, { pushRef: rootPushRef }); + const activeNode = workflowData.nodes[0]; + const hooks = buildHooks({ + ...workflowData, + nodes: [ + activeNode, + { ...activeNode, id: 'active-2', name: 'Second Node' }, + { ...activeNode, id: 'disabled-1', name: 'Disabled Node', disabled: true }, + { ...activeNode, id: 'sticky-1', name: 'Sticky', type: 'n8n-nodes-base.stickyNote' }, + ], + }); + + await hooks.runHook('workflowExecuteBefore', [workflow, runExecutionData]); + + expect(push.send).toHaveBeenCalledWith( + expect.objectContaining({ + type: 'subworkflowExecutionStarted', + data: expect.objectContaining({ totalNodes: 2 }), + }), + rootPushRef, + ); + }); + + it('counts unique nodes, not executions, so loops never exceed the total', async () => { + stubActiveExecution(parentExecutionId, { pushRef: rootPushRef }); + const hooks = buildHooks(); + + // A loop revisits node A: A → B → A + await hooks.runHook('nodeExecuteBefore', ['Node A', taskStartedData]); + await hooks.runHook('nodeExecuteAfter', ['Node A', taskData, runExecutionData]); + await hooks.runHook('nodeExecuteBefore', ['Node B', taskStartedData]); + await hooks.runHook('nodeExecuteAfter', ['Node B', taskData, runExecutionData]); + await hooks.runHook('nodeExecuteBefore', ['Node A', taskStartedData]); + + const lastCall = push.send.mock.calls.at(-1)?.[0]; + expect(lastCall).toMatchObject({ + type: 'subworkflowNodeProgress', + data: { currentNodeName: 'Node A', currentNodeIndex: 2 }, + }); + }); + + describe('throttling', () => { + it('suppresses same-node running repeats within the throttle window', async () => { + stubActiveExecution(parentExecutionId, { pushRef: rootPushRef }); + const hooks = buildHooks(); + + await hooks.runHook('nodeExecuteBefore', [nodeName, taskStartedData]); + await hooks.runHook('nodeExecuteBefore', [nodeName, taskStartedData]); + + expect(push.send).toHaveBeenCalledTimes(1); + }); + + it('emits same-node running repeats once the throttle window has elapsed', async () => { + stubActiveExecution(parentExecutionId, { pushRef: rootPushRef }); + const hooks = buildHooks(); + + await hooks.runHook('nodeExecuteBefore', [nodeName, taskStartedData]); + vi.advanceTimersByTime(150); + await hooks.runHook('nodeExecuteBefore', [nodeName, taskStartedData]); + + expect(push.send).toHaveBeenCalledTimes(2); + }); + + it('always emits on node-name change regardless of the window', async () => { + stubActiveExecution(parentExecutionId, { pushRef: rootPushRef }); + const hooks = buildHooks(); + + await hooks.runHook('nodeExecuteBefore', ['Node A', taskStartedData]); + await hooks.runHook('nodeExecuteBefore', ['Node B', taskStartedData]); + + expect(push.send).toHaveBeenCalledTimes(2); + expect(push.send.mock.calls.at(-1)?.[0]).toMatchObject({ + data: { currentNodeName: 'Node B', phase: 'running' }, + }); + }); + + it('always emits on phase change regardless of the window', async () => { + stubActiveExecution(parentExecutionId, { pushRef: rootPushRef }); + const hooks = buildHooks(); + + // before → after → before on the same node, all within the window: + // the after (phase success) and the second before (phase running) + // must both emit so the UI never shows a stale phase. + await hooks.runHook('nodeExecuteBefore', [nodeName, taskStartedData]); + await hooks.runHook('nodeExecuteAfter', [nodeName, taskData, runExecutionData]); + await hooks.runHook('nodeExecuteBefore', [nodeName, taskStartedData]); + + expect(push.send).toHaveBeenCalledTimes(3); + expect(push.send.mock.calls.at(-1)?.[0]).toMatchObject({ + data: { phase: 'running' }, + }); + }); + }); + }); }); }); diff --git a/packages/cli/src/execution-lifecycle/execution-lifecycle-hooks.ts b/packages/cli/src/execution-lifecycle/execution-lifecycle-hooks.ts index 65765480fe6..a2a4ce4fbee 100644 --- a/packages/cli/src/execution-lifecycle/execution-lifecycle-hooks.ts +++ b/packages/cli/src/execution-lifecycle/execution-lifecycle-hooks.ts @@ -11,8 +11,10 @@ import { FileLocation, InstanceSettings, } from 'n8n-core'; +import { STICKY_NODE_TYPE } from 'n8n-workflow'; import type { ExecutionStatus, + INode, IRun, IRunData, IRunExecutionData, @@ -23,6 +25,7 @@ import type { } from 'n8n-workflow'; import { runDataAttemptedDynamicCredentials, runDataUsedDynamicCredentials } from 'n8n-workflow'; +import { ActiveExecutions } from '@/active-executions'; import { EventService } from '@/events/event.service'; import { ExecutionPersistence } from '@/executions/execution-persistence'; import type { RedactableExecution } from '@/executions/execution-redaction'; @@ -466,6 +469,168 @@ function hookFunctionsPush( }); } +/** + * Walks up the active-execution chain starting at `executionId` and returns + * the pushRef of the root execution that owns the editor UI session. Returns + * undefined if no ancestor in the chain has a pushRef (e.g. the root was a + * trigger/webhook/API-initiated run, or the root execution has already left + * the in-memory ActiveExecutions registry). + */ +function findRootPushRef(executionId: string): string | undefined { + const activeExecutions = Container.get(ActiveExecutions); + const seen = new Set(); + let currentId: string | undefined = executionId; + while (currentId && !seen.has(currentId)) { + seen.add(currentId); + if (!activeExecutions.has(currentId)) return undefined; + const entry = activeExecutions.getExecutionOrFail(currentId); + const pushRef = entry.executionData.pushRef; + if (pushRef) return pushRef; + currentId = entry.executionData.executionData?.parentExecution?.executionId; + } + return undefined; +} + +/** + * Push hooks for a sub-workflow execution. Forwards a lightweight progress + * stream up to the root parent's pushRef so the editor can render a live + * overlay on the parent's "Execute Sub-workflow" node. + * + * No-ops when invoked without a parent execution / node, or when no ancestor + * in the chain holds a pushRef (e.g. workflow started via the public API). + */ +function hookFunctionsPushSubExecution( + hooks: ExecutionLifecycleHooks, + workflowData: IWorkflowBase, + parentExecution: RelatedExecution, + parentNode: INode, +) { + // Sticky notes and disabled nodes never execute, so they must not count + // towards the "X / Y" indicator's denominator. + const totalNodes = workflowData.nodes.filter( + (node) => !node.disabled && node.type !== STICKY_NODE_TYPE, + ).length; + + // Push + pushRef are resolved lazily on first emit. This keeps hook + // registration free of DI side effects (cli tests globally `jest.mock` + // `@/push`, which strips its `@Service` metadata) and lets the freshest + // ancestor state determine the target pushRef. A failed resolution is + // retried on the next emit rather than cached, so a transient miss + // doesn't silence the whole stream. + let cached: { pushInstance: Push; pushRef: string } | undefined; + const resolveTarget = (): typeof cached => { + if (cached) return cached; + const pushRef = findRootPushRef(parentExecution.executionId); + if (!pushRef) return undefined; + try { + cached = { pushInstance: Container.get(Push), pushRef }; + } catch { + return undefined; + } + return cached; + }; + + // Counts unique nodes reached, not node executions: in a looping child + // workflow the same node runs many times, and execution count would make + // the indicator exceed its denominator (e.g. "30 / 3"). + const reachedNodeNames = new Set(); + // Coalesces high-frequency repeats on the same node. Any phase change + // or node-name change always emits, so the UI never lags behind which + // node is actually running. + const throttleMs = totalNodes >= 50 ? 250 : 100; + let lastEmitAt = 0; + let lastEmittedNodeName: string | undefined; + let lastEmittedPhase: 'running' | 'success' | 'error' | undefined; + + hooks.addHandler('workflowExecuteBefore', function () { + const target = resolveTarget(); + if (!target) return; + target.pushInstance.send( + { + type: 'subworkflowExecutionStarted', + data: { + parentExecutionId: parentExecution.executionId, + parentNodeName: parentNode.name, + executionId: this.executionId, + totalNodes, + }, + }, + target.pushRef, + ); + }); + + hooks.addHandler('nodeExecuteBefore', function (nodeName) { + const target = resolveTarget(); + if (!target) return; + reachedNodeNames.add(nodeName); + const now = Date.now(); + const isNewNode = nodeName !== lastEmittedNodeName; + const isPhaseChange = lastEmittedPhase !== 'running'; + // Always emit when the displayed state would actually change. + if (!isNewNode && !isPhaseChange && now - lastEmitAt < throttleMs) return; + lastEmitAt = now; + lastEmittedNodeName = nodeName; + lastEmittedPhase = 'running'; + target.pushInstance.send( + { + type: 'subworkflowNodeProgress', + data: { + parentExecutionId: parentExecution.executionId, + parentNodeName: parentNode.name, + executionId: this.executionId, + currentNodeName: nodeName, + currentNodeIndex: reachedNodeNames.size, + totalNodes, + phase: 'running', + }, + }, + target.pushRef, + ); + }); + + hooks.addHandler('nodeExecuteAfter', function (nodeName, data) { + const target = resolveTarget(); + if (!target) return; + reachedNodeNames.add(nodeName); + const phase: 'success' | 'error' = data?.error ? 'error' : 'success'; + lastEmitAt = Date.now(); + lastEmittedNodeName = nodeName; + lastEmittedPhase = phase; + target.pushInstance.send( + { + type: 'subworkflowNodeProgress', + data: { + parentExecutionId: parentExecution.executionId, + parentNodeName: parentNode.name, + executionId: this.executionId, + currentNodeName: nodeName, + currentNodeIndex: reachedNodeNames.size, + totalNodes, + phase, + }, + }, + target.pushRef, + ); + }); + + hooks.addHandler('workflowExecuteAfter', function (fullRunData) { + const target = resolveTarget(); + if (!target) return; + target.pushInstance.send( + { + type: 'subworkflowExecutionFinished', + data: { + parentExecutionId: parentExecution.executionId, + parentNodeName: parentNode.name, + executionId: this.executionId, + status: fullRunData.status, + }, + }, + target.pushRef, + ); + }); +} + function hookFunctionsExternalHooks(hooks: ExecutionLifecycleHooks) { const externalHooks = Container.get(ExternalHooks); const workflowContext = Container.get(WorkflowHookContextService); @@ -760,10 +925,15 @@ export function getLifecycleHooksForSubExecutions( executionId: string, workflowData: IWorkflowBase, userId?: string, - parentExecution?: RelatedExecution, - projectId?: string, - projectName?: string, + options: { + parentExecution?: RelatedExecution; + projectId?: string; + projectName?: string; + /** The "Execute Sub-workflow" node in the parent workflow that spawned this execution. */ + parentNode?: INode; + } = {}, ): ExecutionLifecycleHooks { + const { parentExecution, projectId, projectName, parentNode } = options; const hooks = new ExecutionLifecycleHooks(mode, executionId, workflowData); const saveSettings = toSaveSettings(workflowData.settings); hookFunctionsWorkflowEvents(hooks, userId, projectId, projectName); @@ -773,6 +943,9 @@ export function getLifecycleHooksForSubExecutions( hookFunctionsSaveProgress(hooks, { saveSettings }); hookFunctionsStatistics(hooks); hookFunctionsExternalHooks(hooks); + if (parentExecution && parentNode) { + hookFunctionsPushSubExecution(hooks, workflowData, parentExecution, parentNode); + } Container.get(ModulesHooksRegistry).addHooks(hooks); return hooks; } diff --git a/packages/cli/src/workflow-execute-additional-data.ts b/packages/cli/src/workflow-execute-additional-data.ts index 8c0b52d06c4..9e0065ee260 100644 --- a/packages/cli/src/workflow-execute-additional-data.ts +++ b/packages/cli/src/workflow-execute-additional-data.ts @@ -563,7 +563,7 @@ async function startExecution( executionId, workflowData, additionalData.userId, - options.parentExecution, + { parentExecution: options.parentExecution, parentNode: options.node }, ); additionalDataIntegrated.executionId = executionId; additionalDataIntegrated.parentCallbackManager = options.parentCallbackManager;