import * as PCancelable from 'p-cancelable' ; import { ExecutionError , IConnection , IDataObject , IExecuteData , INode , INodeConnections , INodeExecutionData , IRun , IRunData , IRunExecutionData , ITaskData , ITaskDataConnections , IWaitingForExecution , IWorkflowExecuteAdditionalData , LoggerProxy as Logger , NodeApiError , NodeOperationError , Workflow , WorkflowExecuteMode , WorkflowOperationError , } from 'n8n-workflow' ; import { get } from 'lodash' ; import { NodeExecuteFunctions } from '.' ; export class WorkflowExecute { runExecutionData : IRunExecutionData ; private additionalData : IWorkflowExecuteAdditionalData ; private mode : WorkflowExecuteMode ; constructor ( additionalData , mode , runExecutionData ? ) { this . additionalData = additionalData ; this . mode = mode ; this . runExecutionData = runExecutionData || { startData : { } , resultData : { runData : { } , } , executionData : { contextData : { } , nodeExecutionStack : [ ] , waitingExecution : { } , } , } ; } param { Workflow } workflow param { INode [ ] } startNodes param { string } destinationNode returns { ( Promise < string > ) } memberof run ( workflow , startNode ? , destinationNode ? ) { startNode = startNode || workflow . getStartNode ( destinationNode ) ; if ( startNode === undefined ) { throw new Error ( 'No node to start the workflow from could be found!' ) ; } let runNodeFilter ; if ( destinationNode ) { runNodeFilter = workflow . getParentNodes ( destinationNode ) ; runNodeFilter . push ( destinationNode ) ; } const nodeExecutionStack = [ { node : startNode , data : { main : [ [ { json : { } , } , ] , ] , } , } , ] ; this . runExecutionData = { startData : { destinationNode , runNodeFilter , } , resultData : { runData : { } , } , executionData : { contextData : { } , nodeExecutionStack , waitingExecution : { } , } , } ; return this . processRunExecutionData ( workflow ) ; } param { Workflow } workflow param { IRunData } runData param { string [ ] } startNodes param { string } destinationNode returns { ( Promise < string > ) } memberof runPartialWorkflow ( workflow , runData , startNodes , destinationNode ) { let incomingNodeConnections ; let connection ; const runIndex = 0 ; const nodeExecutionStack = [ ] ; const waitingExecution = { } ; for ( const startNode of startNodes ) { incomingNodeConnections = workflow . connectionsByDestinationNode [ startNode ] ; const incomingData = [ ] ; if ( incomingNodeConnections === undefined ) { incomingData . push ( [ { json : { } , } , ] ) ; } else { for ( const connections of incomingNodeConnections . main ) { for ( let inputIndex = 0 ; inputIndex < connections . length ; inputIndex ++ ) { connection = connections [ inputIndex ] ; incomingData . push ( runData [ connection . node ] [ runIndex ] . data ! [ connection . type ] [ connection . index ] ! ) ; } } } const executeData = { node : workflow . getNode ( startNode ) as INode , data : { main : incomingData , } , } ; nodeExecutionStack . push ( executeData ) ; incomingNodeConnections = workflow . connectionsByDestinationNode [ destinationNode ] ; if ( incomingNodeConnections !== undefined ) { for ( const connections of incomingNodeConnections . main ) { for ( let inputIndex = 0 ; inputIndex < connections . length ; inputIndex ++ ) { connection = connections [ inputIndex ] ; if ( waitingExecution [ destinationNode ] === undefined ) { waitingExecution [ destinationNode ] = { } ; } if ( waitingExecution [ destinationNode ] [ runIndex ] === undefined ) { waitingExecution [ destinationNode ] [ runIndex ] = { } ; } if ( waitingExecution [ destinationNode ] [ runIndex ] [ connection . type ] === undefined ) { waitingExecution [ destinationNode ] [ runIndex ] [ connection . type ] = [ ] ; } if ( runData [ connection . node ] !== undefined ) { waitingExecution [ destinationNode ] [ runIndex ] [ connection . type ] . push ( runData [ connection . node ] [ runIndex ] . data ! [ connection . type ] [ connection . index ] ) ; } else { waitingExecution [ destinationNode ] [ runIndex ] [ connection . type ] . push ( null ) ; } } } } } let runNodeFilter ; runNodeFilter = workflow . getParentNodes ( destinationNode ) ; runNodeFilter . push ( destinationNode ) ; this . runExecutionData = { startData : { destinationNode , runNodeFilter , } , resultData : { runData , } , executionData : { contextData : { } , nodeExecutionStack , waitingExecution , } , } ; return this . processRunExecutionData ( workflow ) ; } param { string } hookName param { any [ ] } parameters returns { Promise < IRun > } memberof async executeHook ( hookName , parameters ) { if ( this . additionalData . hooks === undefined ) { return ; } return this . additionalData . hooks . executeHookFunctions ( hookName , parameters ) ; } /**
     * Checks the incoming connection does not receive any data
     */ incomingConnectionIsEmpty ( runData , inputConnections , runIndex ) { for ( const inputConnection of inputConnections ) { const nodeIncomingData = get ( runData , `[${ inputConnection . node }][${ runIndex }].data.main[${ inputConnection . index }]` ) ; if ( nodeIncomingData !== undefined && ( nodeIncomingData as object [ ] ) . length !== 0 ) { return false ; } } return true ; } addNodeToBeExecuted ( workflow , connectionData , outputIndex , parentNodeName , nodeSuccessData , runIndex ) { let stillDataMissing = false ; if ( workflow . connectionsByDestinationNode [ connectionData . node ] . main . length > 1 ) { let nodeWasWaiting = true ; if ( this . runExecutionData . executionData ! . waitingExecution [ connectionData . node ] === undefined ) { this . runExecutionData . executionData ! . waitingExecution [ connectionData . node ] = { } ; nodeWasWaiting = false ; } if ( this . runExecutionData . executionData ! . waitingExecution [ connectionData . node ] [ runIndex ] === undefined ) { this . runExecutionData . executionData ! . waitingExecution [ connectionData . node ] [ runIndex ] = { main : [ ] , } ; for ( let i = 0 ; i < workflow . connectionsByDestinationNode [ connectionData . node ] . main . length ; i ++ ) { this . runExecutionData . executionData ! . waitingExecution [ connectionData . node ] [ runIndex ] . main . push ( null ) ; } } if ( nodeSuccessData === null ) { this . runExecutionData . executionData ! . waitingExecution [ connectionData . node ] [ runIndex ] . main [ connectionData . index ] = null ; } else { this . runExecutionData . executionData ! . waitingExecution [ connectionData . node ] [ runIndex ] . main [ connectionData . index ] = nodeSuccessData [ outputIndex ] ; } let thisExecutionData ; let allDataFound = true ; for ( let i = 0 ; i < this . runExecutionData . executionData ! . waitingExecution [ connectionData . node ] [ runIndex ] . main . length ; i ++ ) { thisExecutionData = this . runExecutionData . executionData ! . waitingExecution [ connectionData . node ] [ runIndex ] . main [ i ] ; if ( thisExecutionData === null ) { allDataFound = false ; break ; } } if ( allDataFound ) { this . runExecutionData . executionData ! . nodeExecutionStack . push ( { node : workflow . nodes [ connectionData . node ] , data : this . runExecutionData . executionData ! . waitingExecution [ connectionData . node ] [ runIndex ] , } ) ; delete this . runExecutionData . executionData ! . waitingExecution [ connectionData . node ] [ runIndex ] ; if ( Object . keys ( this . runExecutionData . executionData ! . waitingExecution [ connectionData . node ] ) . length === 0 ) { delete this . runExecutionData . executionData ! . waitingExecution [ connectionData . node ] ; } return ; } stillDataMissing = true ; if ( ! nodeWasWaiting ) { const checkOutputNodes = [ ] ; for ( const outputIndexParent in workflow . connectionsBySourceNode [ parentNodeName ] . main ) { if ( ! workflow . connectionsBySourceNode [ parentNodeName ] . main . hasOwnProperty ( outputIndexParent ) ) { continue ; } for ( const connectionDataCheck of workflow . connectionsBySourceNode [ parentNodeName ] . main [ outputIndexParent ] ) { checkOutputNodes . push ( connectionDataCheck . node ) ; } } for ( let inputIndex = 0 ; inputIndex < workflow . connectionsByDestinationNode [ connectionData . node ] . main . length ; inputIndex ++ ) { for ( const inputData of workflow . connectionsByDestinationNode [ connectionData . node ] . main [ inputIndex ] ) { if ( inputData . node === parentNodeName ) { continue ; } const executionStackNodes = this . runExecutionData . executionData ! . nodeExecutionStack . map ( ( stackData ) => stackData . node . name ) ; if ( inputData . node !== parentNodeName && checkOutputNodes . includes ( inputData . node ) ) { if ( ! this . incomingConnectionIsEmpty ( this . runExecutionData . resultData . runData , workflow . connectionsByDestinationNode [ inputData . node ] . main [ 0 ] , runIndex ) ) { continue ; } } if ( executionStackNodes . includes ( inputData . node ) ) { continue ; } if ( this . runExecutionData . resultData . runData [ inputData . node ] !== undefined ) { continue ; } const parentNodes = workflow . getParentNodes ( inputData . node , 'main' , - 1 ) ; let nodeToAdd = inputData . node ; parentNodes . push ( inputData . node ) ; parentNodes . reverse ( ) ; for ( const parentNode of parentNodes ) { if ( inputData . node !== parentNode && checkOutputNodes . includes ( parentNode ) ) { nodeToAdd = undefined ; break ; } if ( executionStackNodes . includes ( parentNode ) ) { nodeToAdd = undefined ; break ; } if ( this . runExecutionData . resultData . runData [ parentNode ] !== undefined ) { break ; } nodeToAdd = parentNode ; } const parentNodesNodeToAdd = workflow . getParentNodes ( nodeToAdd as string ) ; if ( parentNodesNodeToAdd . includes ( parentNodeName ) && nodeSuccessData [ outputIndex ] . length === 0 ) { nodeToAdd = undefined ; } if ( nodeToAdd === undefined ) { continue ; } let addEmptyItem = false ; if ( workflow . connectionsByDestinationNode [ nodeToAdd ] === undefined ) { addEmptyItem = true ; } else if ( this . incomingConnectionIsEmpty ( this . runExecutionData . resultData . runData , workflow . connectionsByDestinationNode [ nodeToAdd ] . main [ 0 ] , runIndex ) ) { addEmptyItem = true ; } if ( addEmptyItem ) { this . runExecutionData . executionData ! . nodeExecutionStack . push ( { node : workflow . getNode ( nodeToAdd ) as INode , data : { main : [ [ { json : { } , } , ] , ] , } , } ) ; } } } } } const connectionDataArray = [ ] ; for ( let i = connectionData . index ; i >= 0 ; i -- ) { connectionDataArray [ i ] = null ; } if ( nodeSuccessData === null ) { connectionDataArray [ connectionData . index ] = null ; } else { connectionDataArray [ connectionData . index ] = nodeSuccessData [ outputIndex ] ; } if ( stillDataMissing ) { if ( ! this . runExecutionData . executionData ! . waitingExecution . hasOwnProperty ( connectionData . node ) ) { this . runExecutionData . executionData ! . waitingExecution [ connectionData . node ] = { } ; } this . runExecutionData . executionData ! . waitingExecution [ connectionData . node ] [ runIndex ] = { main : connectionDataArray , } ; } else { this . runExecutionData . executionData ! . nodeExecutionStack . push ( { node : workflow . nodes [ connectionData . node ] , data : { main : connectionDataArray , } , } ) ; } } param { Workflow } workflow returns { Promise < string > } memberof processRunExecutionData ( workflow ) { Logger . verbose ( 'Workflow execution started' , { workflowId : workflow . id } ) ; const startedAt = new Date ( ) ; const startNode = this . runExecutionData . executionData ! . nodeExecutionStack [ 0 ] . node . name ; let destinationNode ; if ( this . runExecutionData . startData && this . runExecutionData . startData . destinationNode ) { destinationNode = this . runExecutionData . startData . destinationNode ; } const workflowIssues = workflow . checkReadyForExecution ( { startNode , destinationNode } ) ; if ( workflowIssues !== null ) { throw new Error ( 'The workflow has issues and can for that reason not be executed. Please fix them first.' ) ; } let executionData ; let executionError ; let executionNode ; let nodeSuccessData ; let runIndex ; let startTime ; let taskData ; if ( this . runExecutionData . startData === undefined ) { this . runExecutionData . startData = { } ; } if ( this . runExecutionData . waitTill ) { const lastNodeExecuted = this . runExecutionData . resultData . lastNodeExecuted as string ; this . runExecutionData . executionData ! . nodeExecutionStack [ 0 ] . node . disabled = true ; this . runExecutionData . waitTill = undefined ; this . runExecutionData . resultData . runData [ lastNodeExecuted ] . pop ( ) ; } let currentExecutionTry = '' ; let lastExecutionTry = '' ; return new PCancelable ( async ( resolve , reject , onCancel ) => { let gotCancel = false ; onCancel . shouldReject = false ; onCancel ( ( ) => { gotCancel = true ; } ) ; const returnPromise = ( async ( ) => { try { await this . executeHook ( 'workflowExecuteBefore' , [ workflow ] ) ; } catch ( error ) { executionError = { ... ( error as NodeOperationError | NodeApiError ) , message : ( error as NodeOperationError | NodeApiError ) . message , stack : ( error as NodeOperationError | NodeApiError ) . stack , } ; executionData = this . runExecutionData . executionData ! . nodeExecutionStack [ 0 ] ; this . runExecutionData . resultData = { runData : { [ executionData . node . name ] : [ { startTime , executionTime : new Date ( ) . getTime ( ) - startTime , data : { main : executionData . data . main , } as ITaskDataConnections , } , ] , } , lastNodeExecuted : executionData . node . name , error : executionError , } ; throw error ; } executionLoop : while ( this . runExecutionData . executionData ! . nodeExecutionStack . length !== 0 ) { if ( this . additionalData . executionTimeoutTimestamp !== undefined && Date . now ( ) >= this . additionalData . executionTimeoutTimestamp ) { gotCancel = true ; } if ( gotCancel ) { return Promise . resolve ( ) ; } nodeSuccessData = null ; executionError = undefined ; executionData = this . runExecutionData . executionData ! . nodeExecutionStack . shift ( ) as IExecuteData ; executionNode = executionData . node ; Logger . debug ( `Start processing node "${ executionNode . name }"` , { node : executionNode . name , workflowId : workflow . id , } ) ; await this . executeHook ( 'nodeExecuteBefore' , [ executionNode . name ] ) ; runIndex = 0 ; if ( this . runExecutionData . resultData . runData . hasOwnProperty ( executionNode . name ) ) { runIndex = this . runExecutionData . resultData . runData [ executionNode . name ] . length ; } currentExecutionTry = `${ executionNode . name }:${ runIndex }` ; if ( currentExecutionTry === lastExecutionTry ) { throw new Error ( 'Did stop execution because execution seems to be in endless loop.' ) ; } if ( this . runExecutionData . startData ! . runNodeFilter !== undefined && this . runExecutionData . startData ! . runNodeFilter . indexOf ( executionNode . name ) === - 1 ) { continue ; } if ( workflow . connectionsByDestinationNode . hasOwnProperty ( executionNode . name ) ) { if ( workflow . connectionsByDestinationNode [ executionNode . name ] . hasOwnProperty ( 'main' ) ) { let inputConnections ; let connectionIndex ; inputConnections = workflow . connectionsByDestinationNode [ executionNode . name ] . main ; for ( connectionIndex = 0 ; connectionIndex < inputConnections . length ; connectionIndex ++ ) { if ( workflow . getHighestNode ( executionNode . name , 'main' , connectionIndex ) . length === 0 ) { continue ; } if ( ! executionData . data . hasOwnProperty ( 'main' ) ) { this . runExecutionData . executionData ! . nodeExecutionStack . push ( executionData ) ; lastExecutionTry = currentExecutionTry ; continue executionLoop ; } if ( executionData . data . main ! . length < connectionIndex || executionData . data . main ! [ connectionIndex ] === null ) { this . runExecutionData . executionData ! . nodeExecutionStack . push ( executionData ) ; lastExecutionTry = currentExecutionTry ; continue executionLoop ; } } } } startTime = new Date ( ) . getTime ( ) ; let maxTries = 1 ; if ( executionData . node . retryOnFail === true ) { maxTries = Math . min ( 5 , Math . max ( 2 , executionData . node . maxTries || 3 ) ) ; } let waitBetweenTries = 0 ; if ( executionData . node . retryOnFail === true ) { waitBetweenTries = Math . min ( 5000 , Math . max ( 0 , executionData . node . waitBetweenTries || 1000 ) ) ; } for ( let tryIndex = 0 ; tryIndex < maxTries ; tryIndex ++ ) { if ( gotCancel ) { return Promise . resolve ( ) ; } try { if ( tryIndex !== 0 ) { executionError = undefined ; if ( waitBetweenTries !== 0 ) { await new Promise ( ( resolve ) => { setTimeout ( ( ) => { resolve ( undefined ) ; } , waitBetweenTries ) ; } ) ; } } Logger . debug ( `Running node "${ executionNode . name }" started` , { node : executionNode . name , workflowId : workflow . id , } ) ; nodeSuccessData = await workflow . runNode ( executionData . node , executionData . data , this . runExecutionData , runIndex , this . additionalData , NodeExecuteFunctions , this . mode ) ; Logger . debug ( `Running node "${ executionNode . name }" finished successfully` , { node : executionNode . name , workflowId : workflow . id , } ) ; if ( nodeSuccessData === undefined ) { nodeSuccessData = null ; } else { this . runExecutionData . resultData . lastNodeExecuted = executionData . node . name ; } if ( nodeSuccessData === null || nodeSuccessData [ 0 ] [ 0 ] === undefined ) { if ( executionData . node . alwaysOutputData === true ) { nodeSuccessData = nodeSuccessData || [ ] ; nodeSuccessData [ 0 ] = [ { json : { } , } , ] ; } } if ( nodeSuccessData === null && ! this . runExecutionData . waitTill ! ) { continue executionLoop ; } break ; } catch ( error ) { this . runExecutionData . resultData . lastNodeExecuted = executionData . node . name ; executionError = { ... ( error as NodeOperationError | NodeApiError ) , message : ( error as NodeOperationError | NodeApiError ) . message , stack : ( error as NodeOperationError | NodeApiError ) . stack , } ; Logger . debug ( `Running node "${ executionNode . name }" finished with error` , { node : executionNode . name , workflowId : workflow . id , } ) ; } } if ( ! this . runExecutionData . resultData . runData . hasOwnProperty ( executionNode . name ) ) { this . runExecutionData . resultData . runData [ executionNode . name ] = [ ] ; } taskData = { startTime , executionTime : new Date ( ) . getTime ( ) - startTime , } ; if ( executionError !== undefined ) { taskData . error = executionError ; if ( executionData . node . continueOnFail === true ) { if ( executionData . data . hasOwnProperty ( 'main' ) && executionData . data . main . length > 0 ) { if ( executionData . data . main [ 0 ] !== null ) { nodeSuccessData = [ executionData . data . main [ 0 ] ] ; } } } else { this . runExecutionData . resultData . runData [ executionNode . name ] . push ( taskData ) ; this . runExecutionData . executionData ! . nodeExecutionStack . unshift ( executionData ) ; await this . executeHook ( 'nodeExecuteAfter' , [ executionNode . name , taskData , this . runExecutionData , ] ) ; break ; } } for ( const execution of nodeSuccessData ! ) { for ( const lineResult of execution ) { if ( lineResult . json !== undefined && lineResult . json . $error !== undefined && lineResult . json . $json !== undefined ) { lineResult . error = lineResult . json . $error as NodeApiError | NodeOperationError ; lineResult . json = { error : ( lineResult . json . $error as NodeApiError | NodeOperationError ) . message , } ; } else if ( lineResult . error !== undefined ) { lineResult . json = { error : lineResult . error . message } ; } } } taskData . data = { main : nodeSuccessData , } as ITaskDataConnections ; this . runExecutionData . resultData . runData [ executionNode . name ] . push ( taskData ) ; if ( this . runExecutionData . waitTill ! ) { await this . executeHook ( 'nodeExecuteAfter' , [ executionNode . name , taskData , this . runExecutionData , ] ) ; this . runExecutionData . executionData ! . nodeExecutionStack . unshift ( executionData ) ; break ; } if ( this . runExecutionData . startData && this . runExecutionData . startData . destinationNode && this . runExecutionData . startData . destinationNode === executionNode . name ) { await this . executeHook ( 'nodeExecuteAfter' , [ executionNode . name , taskData , this . runExecutionData , ] ) ; continue ; } if ( workflow . connectionsBySourceNode . hasOwnProperty ( executionNode . name ) ) { if ( workflow . connectionsBySourceNode [ executionNode . name ] . hasOwnProperty ( 'main' ) ) { let outputIndex ; let connectionData ; for ( outputIndex in workflow . connectionsBySourceNode [ executionNode . name ] . main ) { if ( ! workflow . connectionsBySourceNode [ executionNode . name ] . main . hasOwnProperty ( outputIndex ) ) { continue ; } for ( connectionData of workflow . connectionsBySourceNode [ executionNode . name ] . main [ outputIndex ] ) { if ( ! workflow . nodes . hasOwnProperty ( connectionData . node ) ) { return Promise . reject ( new Error ( `The node "${ executionNode . name }" connects to not found node "${ connectionData . node }"` ) ) ; } if ( nodeSuccessData ! [ outputIndex ] && ( nodeSuccessData ! [ outputIndex ] . length !== 0 || connectionData . index > 0 ) ) { this . addNodeToBeExecuted ( workflow , connectionData , parseInt ( outputIndex , 10 ) , executionNode . name , nodeSuccessData ! , runIndex ) ; } } } } } await this . executeHook ( 'nodeExecuteAfter' , [ executionNode . name , taskData , this . runExecutionData , ] ) ; } return Promise . resolve ( ) ; } ) ( ) . then ( async ( ) => { if ( gotCancel && executionError === undefined ) { return this . processSuccessExecution ( startedAt , workflow , new WorkflowOperationError ( 'Workflow has been canceled or timed out!' ) ) ; } return this . processSuccessExecution ( startedAt , workflow , executionError ) ; } ) . catch ( async ( error ) => { const fullRunData = this . getFullRunData ( startedAt ) ; fullRunData . data . resultData . error = { ... error , message : error . message , stack : error . stack , } ; let newStaticData ; if ( workflow . staticData . __dataChanged === true ) { newStaticData = workflow . staticData ; } await this . executeHook ( 'workflowExecuteAfter' , [ fullRunData , newStaticData ] ) . catch ( ( error ) => { console . error ( 'There was a problem running hook "workflowExecuteAfter"' , error ) ; } ) ; return fullRunData ; } ) ; return returnPromise . then ( resolve ) ; } ) ; } async processSuccessExecution ( startedAt , workflow , executionError ? ) { const fullRunData = this . getFullRunData ( startedAt ) ; if ( executionError !== undefined ) { Logger . verbose ( `Workflow execution finished with error` , { error : executionError , workflowId : workflow . id , } ) ; fullRunData . data . resultData . error = { ... executionError , message : executionError . message , stack : executionError . stack , } as ExecutionError ; } else if ( this . runExecutionData . waitTill ! ) { Logger . verbose ( `Workflow execution will wait until ${ this . runExecutionData . waitTill }` , { workflowId : workflow . id , } ) ; fullRunData . waitTill = this . runExecutionData . waitTill ; } else { Logger . verbose ( `Workflow execution finished successfully` , { workflowId : workflow . id } ) ; fullRunData . finished = true ; } let newStaticData ; if ( workflow . staticData . __dataChanged === true ) { newStaticData = workflow . staticData ; } await this . executeHook ( 'workflowExecuteAfter' , [ fullRunData , newStaticData ] ) ; return fullRunData ; } getFullRunData ( startedAt ) { const fullRunData = { data : this . runExecutionData , mode : this . mode , startedAt , stoppedAt : new Date ( ) , } ; return fullRunData ; } }