import { BinaryDataManager , IProcessMessage , WorkflowExecute } from 'n8n-core' ; import { ExecutionError , IDeferredPromise , IExecuteResponsePromiseData , IRun , LoggerProxy as Logger , Workflow , WorkflowExecuteMode , WorkflowHooks , WorkflowOperationError , } from 'n8n-workflow' ; import * as PCancelable from 'p-cancelable' ; import { join as pathJoin } from 'path' ; import { fork } from 'child_process' ; import * as Bull from 'bull' ; import * as config from '../config' ; import { ActiveExecutions , CredentialsOverwrites , CredentialTypes , Db , ExternalHooks , IBullJobData , IBullJobResponse , ICredentialsOverwrite , ICredentialsTypeData , IExecutionFlattedDb , IProcessMessageDataHook , ITransferNodeTypes , IWorkflowExecutionDataProcess , IWorkflowExecutionDataProcessWithExecution , NodeTypes , Push , ResponseHelper , WebhookHelpers , WorkflowExecuteAdditionalData , WorkflowHelpers , } from '.' ; import * as Queue from './Queue' ; import { InternalHooksManager } from './InternalHooksManager' ; export class WorkflowRunner { activeExecutions : ActiveExecutions . ActiveExecutions ; credentialsOverwrites : ICredentialsOverwrite ; push : Push . Push ; jobQueue : Bull . Queue ; constructor ( ) { this . push = Push . getInstance ( ) ; this . activeExecutions = ActiveExecutions . getInstance ( ) ; this . credentialsOverwrites = CredentialsOverwrites ( ) . getAll ( ) ; const executionsMode = config . get ( 'executions.mode' ) as string ; if ( executionsMode === 'queue' ) { this . jobQueue = Queue . getInstance ( ) . getBullObjectInstance ( ) ; } } param { WorkflowHooks } workflowHooks param { IProcessMessageDataHook } hookData memberof processHookMessage ( workflowHooks , hookData ) { workflowHooks . executeHookFunctions ( hookData . hook , hookData . parameters ) ; } param { ExecutionError } error param { Date } startedAt param { WorkflowExecuteMode } executionMode param { string } executionId memberof async processError ( error , startedAt , executionMode , executionId , hooks ? ) { const fullRunData = { data : { resultData : { error : { ... error , message : error . message , stack : error . stack , } , runData : { } , } , } , finished : false , mode : executionMode , startedAt , stoppedAt : new Date ( ) , } ; this . activeExecutions . remove ( executionId , fullRunData ) ; if ( hooks ) { await hooks . executeHookFunctions ( 'workflowExecuteAfter' , [ fullRunData ] ) ; } } param { IWorkflowExecutionDataProcess } data param { boolean } loadStaticData returns { Promise < string > } memberof async run ( data , loadStaticData ? , realtime ? , executionId ? , responsePromise ? ) { const executionsProcess = config . get ( 'executions.process' ) as string ; const executionsMode = config . get ( 'executions.mode' ) as string ; if ( executionsMode === 'queue' && data . executionMode !== 'manual' ) { executionId = await this . runBull ( data , loadStaticData , realtime , executionId , responsePromise ) ; } else if ( executionsProcess === 'main' ) { executionId = await this . runMainProcess ( data , loadStaticData , executionId , responsePromise ) ; } else { executionId = await this . runSubprocess ( data , loadStaticData , executionId , responsePromise ) ; } const postExecutePromise = this . activeExecutions . getPostExecutePromise ( executionId ) ; const externalHooks = ExternalHooks ( ) ; postExecutePromise . then ( async ( executionData ) => { void InternalHooksManager . getInstance ( ) . onWorkflowPostExecute ( executionId ! , data . workflowData , executionData ) ; } ) . catch ( ( error ) => { console . error ( 'There was a problem running internal hook "onWorkflowPostExecute"' , error ) ; } ) ; if ( externalHooks . exists ( 'workflow.postExecute' ) ) { postExecutePromise . then ( async ( executionData ) => { await externalHooks . run ( 'workflow.postExecute' , [ executionData , data . workflowData , executionId , ] ) ; } ) . catch ( ( error ) => { console . error ( 'There was a problem running hook "workflow.postExecute"' , error ) ; } ) ; } return executionId ; } param { IWorkflowExecutionDataProcess } data param { boolean } loadStaticData returns { Promise < string > } memberof async runMainProcess ( data , loadStaticData ? , restartExecutionId ? , responsePromise ? ) { if ( loadStaticData === true && data . workflowData . id ) { data . workflowData . staticData = await WorkflowHelpers . getStaticDataById ( data . workflowData . id as string ) ; } const nodeTypes = NodeTypes ( ) ; let executionTimeout ; let workflowTimeout = config . get ( 'executions.timeout' ) as number ; if ( data . workflowData . settings && data . workflowData . settings . executionTimeout ) { workflowTimeout = data . workflowData . settings . executionTimeout as number ; } if ( workflowTimeout > 0 ) { workflowTimeout = Math . min ( workflowTimeout , config . get ( 'executions.maxTimeout' ) as number ) ; } const workflow = new Workflow ( { id : data . workflowData . id as string | undefined , name : data . workflowData . name , nodes : data . workflowData . nodes , connections : data . workflowData . connections , active : data . workflowData . active , nodeTypes , staticData : data . workflowData . staticData , } ) ; const additionalData = await WorkflowExecuteAdditionalData . getBase ( undefined , workflowTimeout <= 0 ? undefined : Date . now ( ) + workflowTimeout * 1000 ) ; const executionId = await this . activeExecutions . add ( data , undefined , restartExecutionId ) ; additionalData . executionId = executionId ; Logger . verbose ( `Execution for workflow ${ data . workflowData . name } was assigned id ${ executionId }` , { executionId } ) ; let workflowExecution ; try { Logger . verbose ( `Execution for workflow ${ data . workflowData . name } was assigned id ${ executionId }` , { executionId } ) ; additionalData . hooks = WorkflowExecuteAdditionalData . getWorkflowHooksMain ( data , executionId , true ) ; additionalData . hooks . hookFunctions . sendResponse = [ async ( response ) : Promise < void > => { if ( responsePromise ) { responsePromise . resolve ( response ) ; } } , ] ; additionalData . sendMessageToUI = WorkflowExecuteAdditionalData . sendMessageToUI . bind ( { sessionId : data . sessionId , } ) ; if ( data . executionData !== undefined ) { Logger . debug ( `Execution ID ${ executionId } had Execution data. Running with payload.` , { executionId , } ) ; const workflowExecute = new WorkflowExecute ( additionalData , data . executionMode , data . executionData ) ; workflowExecution = workflowExecute . processRunExecutionData ( workflow ) ; } else if ( data . runData === undefined || data . startNodes === undefined || data . startNodes . length === 0 || data . destinationNode === undefined ) { Logger . debug ( `Execution ID ${ executionId } will run executing all nodes.` , { executionId } ) ; const workflowExecute = new WorkflowExecute ( additionalData , data . executionMode ) ; workflowExecution = workflowExecute . run ( workflow , undefined , data . destinationNode ) ; } else { Logger . debug ( `Execution ID ${ executionId } is a partial execution.` , { executionId } ) ; const workflowExecute = new WorkflowExecute ( additionalData , data . executionMode ) ; workflowExecution = workflowExecute . runPartialWorkflow ( workflow , data . runData , data . startNodes , data . destinationNode ) ; } this . activeExecutions . attachWorkflowExecution ( executionId , workflowExecution ) ; if ( workflowTimeout > 0 ) { const timeout = Math . min ( workflowTimeout , config . get ( 'executions.maxTimeout' ) as number ) * 1000 ; executionTimeout = setTimeout ( ( ) => { this . activeExecutions . stopExecution ( executionId , 'timeout' ) ; } , timeout ) ; } workflowExecution . then ( ( fullRunData ) => { clearTimeout ( executionTimeout ) ; if ( workflowExecution . isCanceled ) { fullRunData . finished = false ; } this . activeExecutions . remove ( executionId , fullRunData ) ; } ) . catch ( ( error ) => { this . processError ( error , new Date ( ) , data . executionMode , executionId , additionalData . hooks ) ; } ) ; } catch ( error ) { await this . processError ( error , new Date ( ) , data . executionMode , executionId , additionalData . hooks ) ; throw error ; } return executionId ; } async runBull ( data , loadStaticData ? , realtime ? , restartExecutionId ? , responsePromise ? ) { const executionId = await this . activeExecutions . add ( data , undefined , restartExecutionId ) ; if ( responsePromise ) { this . activeExecutions . attachResponsePromise ( executionId , responsePromise ) ; } const jobData = { executionId , loadStaticData : ! ! loadStaticData , } ; let priority = 100 ; if ( realtime === true ) { priority = 50 ; } const jobOptions = { priority , removeOnComplete : true , removeOnFail : true , } ; let job ; let hooks ; try { job = await this . jobQueue . add ( jobData , jobOptions ) ; console . log ( `Started with job ID: ${ job . id . toString ( ) } (Execution ID: ${ executionId })` ) ; hooks = WorkflowExecuteAdditionalData . getWorkflowHooksWorkerMain ( data . executionMode , executionId , data . workflowData , { retryOf : data . retryOf ? data . retryOf . toString ( ) : undefined } ) ; hooks . executeHookFunctions ( 'workflowExecuteBefore' , [ ] ) ; } catch ( error ) { const hooks = WorkflowExecuteAdditionalData . getWorkflowHooksWorkerExecuter ( data . executionMode , executionId , data . workflowData , { retryOf : data . retryOf ? data . retryOf . toString ( ) : undefined } ) ; await this . processError ( error , new Date ( ) , data . executionMode , executionId , hooks ) ; throw error ; } const workflowExecution = new PCancelable ( async ( resolve , reject , onCancel ) => { onCancel . shouldReject = false ; onCancel ( async ( ) => { await Queue . getInstance ( ) . stopJob ( job ) ; const hooksWorker = WorkflowExecuteAdditionalData . getWorkflowHooksWorkerExecuter ( data . executionMode , executionId , data . workflowData , { retryOf : data . retryOf ? data . retryOf . toString ( ) : undefined } ) ; const error = new WorkflowOperationError ( 'Workflow-Execution has been canceled!' ) ; await this . processError ( error , new Date ( ) , data . executionMode , executionId , hooksWorker ) ; reject ( error ) ; } ) ; const jobData = job . finished ( ) ; const queueRecoveryInterval = config . get ( 'queue.bull.queueRecoveryInterval' ) as number ; const racingPromises = [ jobData ] ; let clearWatchdogInterval ; if ( queueRecoveryInterval > 0 ) { /** ***********************************************
                 * Long explanation about what this solves:      *
                 * This only happens in a very specific scenario *
                 * when Redis crashes and recovers shortly       *
                 * but during this time, some execution(s)       *
                 * finished. The end result is that the main     *
                 * process will wait indefinitively and never    *
                 * get a response. This adds an active polling to*
                 * the queue that allows us to identify that the *
                 * execution finished and get information from   *
                 * the database.                                 *
                 ************************************************ */ let watchDogInterval ; const watchDog = new Promise ( ( res ) => { watchDogInterval = setInterval ( async ( ) => { const currentJob = await this . jobQueue . getJob ( job . id ) ; if ( currentJob === null ) { res ( { success : true } ) ; } } , queueRecoveryInterval * 1000 ) ; } ) ; racingPromises . push ( watchDog ) ; clearWatchdogInterval = ( ) => { if ( watchDogInterval ) { clearInterval ( watchDogInterval ) ; watchDogInterval = undefined ; } } ; } try { await Promise . race ( racingPromises ) ; if ( clearWatchdogInterval !== undefined ) { clearWatchdogInterval ( ) ; } } catch ( error ) { const hooks = WorkflowExecuteAdditionalData . getWorkflowHooksWorkerExecuter ( data . executionMode , executionId , data . workflowData , { retryOf : data . retryOf ? data . retryOf . toString ( ) : undefined } ) ; Logger . error ( `Problem with execution ${ executionId }: ${ error . message }. Aborting.` ) ; if ( clearWatchdogInterval !== undefined ) { clearWatchdogInterval ( ) ; } await this . processError ( error , new Date ( ) , data . executionMode , executionId , hooks ) ; reject ( error ) ; } const executionDb = ( await Db . collections . Execution ! . findOne ( executionId ) ) as IExecutionFlattedDb ; const fullExecutionData = ResponseHelper . unflattenExecutionData ( executionDb ) ; const runData = { data : fullExecutionData . data , finished : fullExecutionData . finished , mode : fullExecutionData . mode , startedAt : fullExecutionData . startedAt , stoppedAt : fullExecutionData . stoppedAt , } as IRun ; this . activeExecutions . remove ( executionId , runData ) ; hooks . executeHookFunctions ( 'workflowExecuteAfter' , [ runData ] ) ; try { let saveDataErrorExecution = config . get ( 'executions.saveDataOnError' ) as string ; let saveDataSuccessExecution = config . get ( 'executions.saveDataOnSuccess' ) as string ; if ( data . workflowData . settings !== undefined ) { saveDataErrorExecution = ( data . workflowData . settings . saveDataErrorExecution as string ) || saveDataErrorExecution ; saveDataSuccessExecution = ( data . workflowData . settings . saveDataSuccessExecution as string ) || saveDataSuccessExecution ; } const workflowDidSucceed = ! runData . data . resultData . error ; if ( ( workflowDidSucceed && saveDataSuccessExecution === 'none' ) || ( ! workflowDidSucceed && saveDataErrorExecution === 'none' ) ) { await Db . collections . Execution ! . delete ( executionId ) ; await BinaryDataManager . getInstance ( ) . markDataForDeletionByExecutionId ( executionId ) ; } } catch ( err ) { console . log ( 'Error removing saved execution from database. More details: ' , err ) ; } resolve ( runData ) ; } ) ; this . activeExecutions . attachWorkflowExecution ( executionId , workflowExecution ) ; return executionId ; } param { IWorkflowExecutionDataProcess } data param { boolean } loadStaticData returns { Promise < string > } memberof async runSubprocess ( data , loadStaticData ? , restartExecutionId ? , responsePromise ? ) { let startedAt = new Date ( ) ; const subprocess = fork ( pathJoin ( __dirname , 'WorkflowRunnerProcess.js' ) ) ; if ( loadStaticData === true && data . workflowData . id ) { data . workflowData . staticData = await WorkflowHelpers . getStaticDataById ( data . workflowData . id as string ) ; } const executionId = await this . activeExecutions . add ( data , subprocess , restartExecutionId ) ; let loadAllNodeTypes = false ; for ( const node of data . workflowData . nodes ) { if ( node . type === 'n8n-nodes-base.executeWorkflow' ) { loadAllNodeTypes = true ; break ; } } let nodeTypeData ; let credentialTypeData ; let credentialsOverwrites = this . credentialsOverwrites ; if ( loadAllNodeTypes ) { nodeTypeData = WorkflowHelpers . getAllNodeTypeData ( ) ; const credentialTypes = CredentialTypes ( ) ; credentialTypeData = credentialTypes . credentialTypes ; } else { nodeTypeData = WorkflowHelpers . getNodeTypeData ( data . workflowData . nodes ) ; credentialTypeData = WorkflowHelpers . getCredentialsDataByNodes ( data . workflowData . nodes ) ; credentialsOverwrites = { } ; for ( const credentialName of Object . keys ( credentialTypeData ) ) { if ( this . credentialsOverwrites [ credentialName ] !== undefined ) { credentialsOverwrites [ credentialName ] = this . credentialsOverwrites [ credentialName ] ; } } } ( data as unknown as IWorkflowExecutionDataProcessWithExecution ) . executionId = executionId ; ( data as unknown as IWorkflowExecutionDataProcessWithExecution ) . nodeTypeData = nodeTypeData ; ( data as unknown as IWorkflowExecutionDataProcessWithExecution ) . credentialsOverwrite = this . credentialsOverwrites ; ( data as unknown as IWorkflowExecutionDataProcessWithExecution ) . credentialsTypeData = credentialTypeData ; const workflowHooks = WorkflowExecuteAdditionalData . getWorkflowHooksMain ( data , executionId ) ; try { subprocess . send ( { type : 'startWorkflow' , data } as IProcessMessage ) ; } catch ( error ) { await this . processError ( error , new Date ( ) , data . executionMode , executionId , workflowHooks ) ; return executionId ; } let executionTimeout ; let workflowTimeout = config . get ( 'executions.timeout' ) as number ; if ( data . workflowData . settings && data . workflowData . settings . executionTimeout ) { workflowTimeout = data . workflowData . settings . executionTimeout as number ; } const processTimeoutFunction = ( timeout ) => { this . activeExecutions . stopExecution ( executionId , 'timeout' ) ; executionTimeout = setTimeout ( ( ) => subprocess . kill ( ) , Math . max ( timeout * 0.2 , 5000 ) ) ; } ; if ( workflowTimeout > 0 ) { workflowTimeout = Math . min ( workflowTimeout , config . get ( 'executions.maxTimeout' ) as number ) * 1000 ; executionTimeout = setTimeout ( processTimeoutFunction , Math . max ( 5000 , workflowTimeout ) , workflowTimeout ) ; } const childExecutionIds = [ ] ; subprocess . on ( 'message' , async ( message ) => { Logger . debug ( `Received child process message of type ${ message . type } for execution ID ${ executionId }.` , { executionId } ) ; if ( message . type === 'start' ) { startedAt = new Date ( ) ; if ( workflowTimeout > 0 ) { clearTimeout ( executionTimeout ) ; executionTimeout = setTimeout ( processTimeoutFunction , workflowTimeout , workflowTimeout ) ; } } else if ( message . type === 'end' ) { clearTimeout ( executionTimeout ) ; this . activeExecutions . remove ( executionId , message . data . runData ) ; } else if ( message . type === 'sendResponse' ) { if ( responsePromise ) { responsePromise . resolve ( WebhookHelpers . decodeWebhookResponse ( message . data . response ) ) ; } } else if ( message . type === 'sendMessageToUI' ) { WorkflowExecuteAdditionalData . sendMessageToUI . bind ( { sessionId : data . sessionId } ) ( message . data . source , message . data . message ) ; } else if ( message . type === 'processError' ) { clearTimeout ( executionTimeout ) ; const executionError = message . data . executionError as ExecutionError ; await this . processError ( executionError , startedAt , data . executionMode , executionId , workflowHooks ) ; } else if ( message . type === 'processHook' ) { this . processHookMessage ( workflowHooks , message . data as IProcessMessageDataHook ) ; } else if ( message . type === 'timeout' ) { const timeoutError = new WorkflowOperationError ( 'Workflow execution timed out!' ) ; this . processError ( timeoutError , startedAt , data . executionMode , executionId ) ; } else if ( message . type === 'startExecution' ) { const executionId = await this . activeExecutions . add ( message . data . runData ) ; childExecutionIds . push ( executionId ) ; subprocess . send ( { type : 'executionId' , data : { executionId } } as IProcessMessage ) ; } else if ( message . type === 'finishExecution' ) { const executionIdIndex = childExecutionIds . indexOf ( message . data . executionId ) ; if ( executionIdIndex !== - 1 ) { childExecutionIds . splice ( executionIdIndex , 1 ) ; } await this . activeExecutions . remove ( message . data . executionId , message . data . result ) ; } } ) ; subprocess . on ( 'exit' , async ( code , signal ) => { if ( signal === 'SIGTERM' ) { Logger . debug ( `Subprocess for execution ID ${ executionId } timed out.` , { executionId } ) ; const timeoutError = new WorkflowOperationError ( 'Workflow execution timed out!' ) ; await this . processError ( timeoutError , startedAt , data . executionMode , executionId , workflowHooks ) ; } else if ( code !== 0 ) { Logger . debug ( `Subprocess for execution ID ${ executionId } finished with error code ${ code }.` , { executionId } ) ; const executionError = new WorkflowOperationError ( 'Workflow execution process did crash for an unknown reason!' ) ; await this . processError ( executionError , startedAt , data . executionMode , executionId , workflowHooks ) ; } for ( const executionId of childExecutionIds ) { await this . activeExecutions . remove ( executionId ) ; } clearTimeout ( executionTimeout ) ; } ) ; return executionId ; } }