import * as PCancelable from 'p-cancelable' ; import { Command , flags } from '@oclif/command' ; import { BinaryDataManager , IBinaryDataConfig , UserSettings , WorkflowExecute } from 'n8n-core' ; import { IExecuteResponsePromiseData , INodeTypes , IRun , Workflow , LoggerProxy } from 'n8n-workflow' ; import { FindOneOptions } from 'typeorm' ; import * as Bull from 'bull' ; import { CredentialsOverwrites , CredentialTypes , Db , ExternalHooks , GenericHelpers , IBullJobData , IBullJobResponse , IBullWebhookResponse , IExecutionFlattedDb , InternalHooksManager , LoadNodesAndCredentials , NodeTypes , ResponseHelper , WebhookHelpers , WorkflowExecuteAdditionalData , } from '../src' ; import { getLogger } from '../src/Logger' ; import * as config from '../config' ; import * as Queue from '../src/Queue' ; export class Worker extends Command { static description = '\nStarts a n8n worker' ; static examples = [ `$ n8n worker --concurrency=5` ] ; static flags = { help : flags . help ( { char : 'h' } ) , concurrency : flags . integer ( { default : 10 , description : 'How many jobs can run in parallel.' , } ) , } ; static runningJobs : { [ key ] : PCancelable < IRun > ; } = { } ; static jobQueue : Bull . Queue ; static processExistCode = 0 ; /**
     * Stoppes the n8n in a graceful way.
     * Make for example sure that all the webhooks from third party services
     * get removed.
     */ static async stopProcess ( ) { LoggerProxy . info ( `Stopping n8n...` ) ; Worker . jobQueue . pause ( true ) ; try { const externalHooks = ExternalHooks ( ) ; await externalHooks . run ( 'n8n.stop' , [ ] ) ; const maxStopTime = 30000 ; const stopTime = new Date ( ) . getTime ( ) + maxStopTime ; setTimeout ( ( ) => { process . exit ( Worker . processExistCode ) ; } , maxStopTime ) ; let count = 0 ; while ( Object . keys ( Worker . runningJobs ) . length !== 0 ) { if ( count ++ % 4 === 0 ) { const waitLeft = Math . ceil ( ( stopTime - new Date ( ) . getTime ( ) ) / 1000 ) ; LoggerProxy . info ( `Waiting for ${ Object . keys ( Worker . runningJobs ) . length } active executions to finish... (wait ${ waitLeft } more seconds)` ) ; } await new Promise ( ( resolve ) => { setTimeout ( resolve , 500 ) ; } ) ; } } catch ( error ) { LoggerProxy . error ( 'There was an error shutting down n8n.' , error ) ; } process . exit ( Worker . processExistCode ) ; } async runJob ( job , nodeTypes ) { const jobData = job . data as IBullJobData ; const executionDb = ( await Db . collections . Execution ! . findOne ( jobData . executionId ) ) as IExecutionFlattedDb ; const currentExecutionDb = ResponseHelper . unflattenExecutionData ( executionDb ) ; LoggerProxy . info ( `Start job: ${ job . id } (Workflow ID: ${ currentExecutionDb . workflowData . id } | Execution: ${ jobData . executionId })` ) ; let { staticData } = currentExecutionDb . workflowData ; if ( jobData . loadStaticData ) { const findOptions = { select : [ 'id' , 'staticData' ] , } as FindOneOptions ; const workflowData = await Db . collections . Workflow ! . findOne ( currentExecutionDb . workflowData . id , findOptions ) ; if ( workflowData === undefined ) { throw new Error ( `The workflow with the ID "${ currentExecutionDb . workflowData . id }" could not be found` ) ; } staticData = workflowData . staticData ; } let workflowTimeout = config . get ( 'executions.timeout' ) as number ; if ( currentExecutionDb . workflowData . settings && currentExecutionDb . workflowData . settings . executionTimeout ) { workflowTimeout = currentExecutionDb . workflowData . settings . executionTimeout as number ; } let executionTimeoutTimestamp ; if ( workflowTimeout > 0 ) { workflowTimeout = Math . min ( workflowTimeout , config . get ( 'executions.maxTimeout' ) as number ) ; executionTimeoutTimestamp = Date . now ( ) + workflowTimeout * 1000 ; } const workflow = new Workflow ( { id : currentExecutionDb . workflowData . id as string , name : currentExecutionDb . workflowData . name , nodes : currentExecutionDb . workflowData . nodes , connections : currentExecutionDb . workflowData . connections , active : currentExecutionDb . workflowData . active , nodeTypes , staticData , settings : currentExecutionDb . workflowData . settings , } ) ; const additionalData = await WorkflowExecuteAdditionalData . getBase ( undefined , executionTimeoutTimestamp ) ; additionalData . hooks = WorkflowExecuteAdditionalData . getWorkflowHooksWorkerExecuter ( currentExecutionDb . mode , job . data . executionId , currentExecutionDb . workflowData , { retryOf : currentExecutionDb . retryOf as string } ) ; additionalData . hooks . hookFunctions . sendResponse = [ async ( response ) : Promise < void > => { await job . progress ( { executionId : job . data . executionId as string , response : WebhookHelpers . encodeWebhookResponse ( response ) , } as IBullWebhookResponse ) ; } , ] ; additionalData . executionId = jobData . executionId ; let workflowExecute ; let workflowRun ; if ( currentExecutionDb . data !== undefined ) { workflowExecute = new WorkflowExecute ( additionalData , currentExecutionDb . mode , currentExecutionDb . data ) ; workflowRun = workflowExecute . processRunExecutionData ( workflow ) ; } else { workflowExecute = new WorkflowExecute ( additionalData , currentExecutionDb . mode ) ; workflowRun = workflowExecute . run ( workflow ) ; } Worker . runningJobs [ job . id ] = workflowRun ; await workflowRun ; delete Worker . runningJobs [ job . id ] ; return { success : true , } ; } async run ( ) { const logger = getLogger ( ) ; LoggerProxy . init ( logger ) ; console . info ( 'Starting n8n worker...' ) ; process . on ( 'SIGTERM' , Worker . stopProcess ) ; process . on ( 'SIGINT' , Worker . stopProcess ) ; await ( async ( ) => { try { const { flags } = this . parse ( Worker ) ; const startDbInitPromise = Db . init ( ) . catch ( ( error ) => { logger . error ( `There was an error initializing DB: "${ error . message }"` ) ; Worker . processExistCode = 1 ; process . emit ( 'SIGINT' ) ; process . exit ( 1 ) ; } ) ; await UserSettings . prepareUserSettings ( ) ; const loadNodesAndCredentials = LoadNodesAndCredentials ( ) ; await loadNodesAndCredentials . init ( ) ; const credentialsOverwrites = CredentialsOverwrites ( ) ; await credentialsOverwrites . init ( ) ; const externalHooks = ExternalHooks ( ) ; await externalHooks . init ( ) ; const nodeTypes = NodeTypes ( ) ; await nodeTypes . init ( loadNodesAndCredentials . nodeTypes ) ; const credentialTypes = CredentialTypes ( ) ; await credentialTypes . init ( loadNodesAndCredentials . credentialTypes ) ; await startDbInitPromise ; const redisConnectionTimeoutLimit = config . get ( 'queue.bull.redis.timeoutThreshold' ) ; Worker . jobQueue = Queue . getInstance ( ) . getBullObjectInstance ( ) ; Worker . jobQueue . process ( flags . concurrency , async ( job ) => this . runJob ( job , nodeTypes ) ) ; const versions = await GenericHelpers . getVersions ( ) ; const instanceId = await UserSettings . getInstanceId ( ) ; InternalHooksManager . init ( instanceId , versions . cli , nodeTypes ) ; const binaryDataConfig = config . get ( 'binaryDataManager' ) as IBinaryDataConfig ; await BinaryDataManager . init ( binaryDataConfig ) ; console . info ( '\nn8n worker is now ready' ) ; console . info ( ` * Version: ${ versions . cli }` ) ; console . info ( ` * Concurrency: ${ flags . concurrency }` ) ; console . info ( '' ) ; Worker . jobQueue . on ( 'global:progress' , ( jobId , progress ) => { if ( progress === - 1 ) { if ( Worker . runningJobs [ jobId ] !== undefined ) { Worker . runningJobs [ jobId ] . cancel ( ) ; delete Worker . runningJobs [ jobId ] ; } } } ) ; let lastTimer = 0 ; let cumulativeTimeout = 0 ; Worker . jobQueue . on ( 'error' , ( error ) => { if ( error . toString ( ) . includes ( 'ECONNREFUSED' ) ) { const now = Date . now ( ) ; if ( now - lastTimer > 30000 ) { lastTimer = now ; cumulativeTimeout = 0 ; } else { cumulativeTimeout += now - lastTimer ; lastTimer = now ; if ( cumulativeTimeout > redisConnectionTimeoutLimit ) { logger . error ( `Unable to connect to Redis after ${ redisConnectionTimeoutLimit }. Exiting process.` ) ; process . exit ( 1 ) ; } } logger . warn ( 'Redis unavailable - trying to reconnect...' ) ; } else if ( error . toString ( ) . includes ( 'Error initializing Lua scripts' ) ) { logger . error ( 'Error initializing worker.' ) ; process . exit ( 2 ) ; } else { logger . error ( 'Error from queue: ' , error ) ; } } ) ; } catch ( error ) { logger . error ( `Worker process cannot continue. "${ error . message }"` ) ; Worker . processExistCode = 1 ; process . emit ( 'SIGINT' ) ; process . exit ( 1 ) ; } } ) ( ) ; } }