Skip to content

Commit

Permalink
refactor(core): Couple of refactors on WorkflowRunner and ActiveExecu…
Browse files Browse the repository at this point in the history 
…tions (no-changelog) (#8487)
  • Loading branch information
@netroy
netroy authored Feb 6, 2024
1 parent dc068ce commit c04f92f
Show file tree
Hide file tree
Showing 6 changed files with 64 additions and 141 deletions.
80 changes: 37 additions & 43 deletions packages/cli/src/ActiveExecutions.ts
View file
Original file line number Diff line number Diff line change
Expand Up @@ -6,7 +6,7 @@ import type {
IRun,
ExecutionStatus,
} from 'n8n-workflow';
import { ApplicationError, WorkflowOperationError, createDeferredPromise } from 'n8n-workflow';
import { ApplicationError, createDeferredPromise, sleep } from 'n8n-workflow';

import type {
ExecutionPayload,
Expand All @@ -22,7 +22,7 @@ import { Logger } from '@/Logger';
@Service()
export class ActiveExecutions {
private activeExecutions: {
[index: string]: IExecutingWorkflowData;
[executionId: string]: IExecutingWorkflowData;
} = {};

constructor(
Expand Down Expand Up @@ -87,32 +87,17 @@ export class ActiveExecutions {

/**
* Attaches an execution
*
*/

attachWorkflowExecution(executionId: string, workflowExecution: PCancelable<IRun>) {
const execution = this.activeExecutions[executionId];
if (execution === undefined) {
throw new ApplicationError('No active execution found to attach to workflow execution to', {
extra: { executionId },
});
}

execution.workflowExecution = workflowExecution;
this.getExecution(executionId).workflowExecution = workflowExecution;
}

attachResponsePromise(
executionId: string,
responsePromise: IDeferredPromise<IExecuteResponsePromiseData>,
): void {
const execution = this.activeExecutions[executionId];
if (execution === undefined) {
throw new ApplicationError('No active execution found to attach to workflow execution to', {
extra: { executionId },
});
}

execution.responsePromise = responsePromise;
this.getExecution(executionId).responsePromise = responsePromise;
}

resolveResponsePromise(executionId: string, response: IExecuteResponsePromiseData): void {
Expand All @@ -126,7 +111,6 @@ export class ActiveExecutions {

/**
* Remove an active execution
*
*/
remove(executionId: string, fullRunData?: IRun): void {
const execution = this.activeExecutions[executionId];
Expand All @@ -135,7 +119,6 @@ export class ActiveExecutions {
}

// Resolve all the waiting promises

for (const promise of execution.postExecutePromises) {
promise.resolve(fullRunData);
}
Expand All @@ -160,28 +143,17 @@ export class ActiveExecutions {
}

/**
* Returns a promise which will resolve with the data of the execution
* with the given id
*
* @param {string} executionId The id of the execution to wait for
* Returns a promise which will resolve with the data of the execution with the given id
*/
async getPostExecutePromise(executionId: string): Promise<IRun | undefined> {
const execution = this.activeExecutions[executionId];
if (execution === undefined) {
throw new WorkflowOperationError(`There is no active execution with id "${executionId}".`);
}

// Create the promise which will be resolved when the execution finished
const waitPromise = await createDeferredPromise<IRun | undefined>();

execution.postExecutePromises.push(waitPromise);

this.getExecution(executionId).postExecutePromises.push(waitPromise);
return await waitPromise.promise();
}

/**
* Returns all the currently active executions
*
*/
getActiveExecutions(): IExecutionsCurrentSummary[] {
const returnData: IExecutionsCurrentSummary[] = [];
Expand All @@ -203,20 +175,42 @@ export class ActiveExecutions {
return returnData;
}

async setStatus(executionId: string, status: ExecutionStatus): Promise<void> {
const execution = this.activeExecutions[executionId];
if (execution === undefined) {
this.logger.debug(
`There is no active execution with id "${executionId}", can't update status to ${status}.`,
setStatus(executionId: string, status: ExecutionStatus) {
this.getExecution(executionId).status = status;
}

getStatus(executionId: string): ExecutionStatus {
return this.getExecution(executionId).status;
}

/** Wait for all active executions to finish */
async shutdown(cancelAll = false) {
let executionIds = Object.keys(this.activeExecutions);

if (cancelAll) {
const stopPromises = executionIds.map(
async (executionId) => await this.stopExecution(executionId),
);
return;

await Promise.allSettled(stopPromises);
}

execution.status = status;
let count = 0;
while (executionIds.length !== 0) {
if (count++ % 4 === 0) {
this.logger.info(`Waiting for ${executionIds.length} active executions to finish...`);
}

await sleep(500);
executionIds = Object.keys(this.activeExecutions);
}
}

getStatus(executionId: string): ExecutionStatus {
private getExecution(executionId: string): IExecutingWorkflowData {
const execution = this.activeExecutions[executionId];
return execution?.status ?? 'unknown';
if (!execution) {
throw new ApplicationError('No active execution found', { extra: { executionId } });
}
return execution;
}
}
6 changes: 1 addition & 5 deletions packages/cli/src/WorkflowExecuteAdditionalData.ts
View file
Original file line number Diff line number Diff line change
Expand Up @@ -930,11 +930,7 @@ export function setExecutionStatus(status: ExecutionStatus) {
return;
}
logger.debug(`Setting execution status for ${this.executionId} to "${status}"`);
Container.get(ActiveExecutions)
.setStatus(this.executionId, status)
.catch((error) => {
logger.debug(`Setting execution status "${status}" failed: ${error.message}`);
});
Container.get(ActiveExecutions).setStatus(this.executionId, status);
}

export function sendDataToUI(type: string, data: IDataObject | IDataObject[]) {
Expand Down
58 changes: 20 additions & 38 deletions packages/cli/src/WorkflowRunner.ts
View file
Original file line number Diff line number Diff line change
Expand Up @@ -35,7 +35,6 @@ import { Queue } from '@/Queue';
import * as WorkflowHelpers from '@/WorkflowHelpers';
import * as WorkflowExecuteAdditionalData from '@/WorkflowExecuteAdditionalData';
import { generateFailedExecutionFromError } from '@/WorkflowHelpers';
import { initErrorHandling } from '@/ErrorReporting';
import { PermissionChecker } from '@/UserManagement/PermissionChecker';
import { InternalHooks } from '@/InternalHooks';
import { Logger } from '@/Logger';
Expand All @@ -55,7 +54,11 @@ export class WorkflowRunner {
private readonly workflowStaticDataService: WorkflowStaticDataService,
private readonly nodeTypes: NodeTypes,
private readonly permissionChecker: PermissionChecker,
) {}
) {
if (this.executionsMode === 'queue') {
this.jobQueue = Container.get(Queue);
}
}

/** The process did error */
async processError(
Expand Down Expand Up @@ -150,27 +153,21 @@ export class WorkflowRunner {
data: IWorkflowExecutionDataProcess,
loadStaticData?: boolean,
realtime?: boolean,
executionId?: string,
restartExecutionId?: string,
responsePromise?: IDeferredPromise<IExecuteResponsePromiseData>,
): Promise<string> {
await initErrorHandling();

if (this.executionsMode === 'queue') {
this.jobQueue = Container.get(Queue);
// Register a new execution
const executionId = await this.activeExecutions.add(data, restartExecutionId);
if (responsePromise) {
this.activeExecutions.attachResponsePromise(executionId, responsePromise);
}

if (this.executionsMode === 'queue' && data.executionMode !== 'manual') {
// Do not run "manual" executions in bull because sending events to the
// frontend would not be possible
executionId = await this.enqueueExecution(
data,
loadStaticData,
realtime,
executionId,
responsePromise,
);
await this.enqueueExecution(executionId, data, loadStaticData, realtime);
} else {
executionId = await this.runMainProcess(data, loadStaticData, executionId, responsePromise);
await this.runMainProcess(executionId, data, loadStaticData, executionId);
void Container.get(InternalHooks).onWorkflowBeforeExecute(executionId, data);
}

Expand All @@ -185,7 +182,7 @@ export class WorkflowRunner {
postExecutePromise
.then(async (executionData) => {
void Container.get(InternalHooks).onWorkflowPostExecute(
executionId!,
executionId,
data.workflowData,
executionData,
data.userId,
Expand Down Expand Up @@ -214,11 +211,11 @@ export class WorkflowRunner {

/** Run the workflow in current process */
private async runMainProcess(
executionId: string,
data: IWorkflowExecutionDataProcess,
loadStaticData?: boolean,
restartExecutionId?: string,
responsePromise?: IDeferredPromise<IExecuteResponsePromiseData>,
): Promise<string> {
): Promise<void> {
const workflowId = data.workflowData.id;
if (loadStaticData === true && workflowId) {
data.workflowData.staticData =
Expand Down Expand Up @@ -257,10 +254,9 @@ export class WorkflowRunner {
undefined,
workflowTimeout <= 0 ? undefined : Date.now() + workflowTimeout * 1000,
);
// TODO: set this in queue mode as well
additionalData.restartExecutionId = restartExecutionId;

// Register the active execution
const executionId = await this.activeExecutions.add(data, restartExecutionId);
additionalData.executionId = executionId;

this.logger.verbose(
Expand Down Expand Up @@ -290,14 +286,12 @@ export class WorkflowRunner {
);
await additionalData.hooks.executeHookFunctions('workflowExecuteAfter', [failedExecution]);
this.activeExecutions.remove(executionId, failedExecution);
return executionId;
return;
}

additionalData.hooks.hookFunctions.sendResponse = [
async (response: IExecuteResponsePromiseData): Promise<void> => {
if (responsePromise) {
responsePromise.resolve(response);
}
this.activeExecutions.resolveResponsePromise(executionId, response);
},
];

Expand Down Expand Up @@ -391,25 +385,14 @@ export class WorkflowRunner {

throw error;
}

return executionId;
}

private async enqueueExecution(
executionId: string,
data: IWorkflowExecutionDataProcess,
loadStaticData?: boolean,
realtime?: boolean,
restartExecutionId?: string,
responsePromise?: IDeferredPromise<IExecuteResponsePromiseData>,
): Promise<string> {
// TODO: If "loadStaticData" is set to true it has to load data new on worker

// Register the active execution
const executionId = await this.activeExecutions.add(data, restartExecutionId);
if (responsePromise) {
this.activeExecutions.attachResponsePromise(executionId, responsePromise);
}

): Promise<void> {
const jobData: JobData = {
executionId,
loadStaticData: !!loadStaticData,
Expand Down Expand Up @@ -601,6 +584,5 @@ export class WorkflowRunner {
});

this.activeExecutions.attachWorkflowExecution(executionId, workflowExecution);
return executionId;
}
}
23 changes: 2 additions & 21 deletions packages/cli/src/commands/executeBatch.ts
View file
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,7 @@ import { Flags } from '@oclif/core';
import fs from 'fs';
import os from 'os';
import type { IRun, ITaskData } from 'n8n-workflow';
import { ApplicationError, jsonParse, sleep } from 'n8n-workflow';
import { ApplicationError, jsonParse } from 'n8n-workflow';
import { sep } from 'path';
import { diff } from 'json-diff';
import pick from 'lodash/pick';
Expand Down Expand Up @@ -118,28 +118,9 @@ export class ExecuteBatch extends BaseCommand {
}

ExecuteBatch.cancelled = true;
const activeExecutionsInstance = Container.get(ActiveExecutions);
const stopPromises = activeExecutionsInstance
.getActiveExecutions()
.map(async (execution) => await activeExecutionsInstance.stopExecution(execution.id));

await Promise.allSettled(stopPromises);
await Container.get(ActiveExecutions).shutdown(true);

setTimeout(() => process.exit(0), 30000);

let executingWorkflows = activeExecutionsInstance.getActiveExecutions();

let count = 0;
while (executingWorkflows.length !== 0) {
if (count++ % 4 === 0) {
console.log(`Waiting for ${executingWorkflows.length} active executions to finish...`);
executingWorkflows.map((execution) => {
console.log(` - Execution ID ${execution.id}, workflow ID: ${execution.workflowId}`);
});
}
await sleep(500);
executingWorkflows = activeExecutionsInstance.getActiveExecutions();
}
// We may receive true but when called from `process.on`
// we get the signal (SIGINT, etc.)
if (skipExit !== true) {
Expand Down
20 changes: 2 additions & 18 deletions packages/cli/src/commands/start.ts
View file
Original file line number Diff line number Diff line change
Expand Up @@ -8,7 +8,7 @@ import { createReadStream, createWriteStream, existsSync } from 'fs';
import { pipeline } from 'stream/promises';
import replaceStream from 'replacestream';
import glob from 'fast-glob';
import { sleep, jsonParse } from 'n8n-workflow';
import { jsonParse } from 'n8n-workflow';

import config from '@/config';
import { ActiveExecutions } from '@/ActiveExecutions';
Expand Down Expand Up @@ -106,23 +106,7 @@ export class Start extends BaseCommand {

await Container.get(InternalHooks).onN8nStop();

// Wait for active workflow executions to finish
const activeExecutionsInstance = Container.get(ActiveExecutions);
let executingWorkflows = activeExecutionsInstance.getActiveExecutions();

let count = 0;
while (executingWorkflows.length !== 0) {
if (count++ % 4 === 0) {
console.log(`Waiting for ${executingWorkflows.length} active executions to finish...`);

executingWorkflows.map((execution) => {
console.log(` - Execution ID ${execution.id}, workflow ID: ${execution.workflowId}`);
});
}

await sleep(500);
executingWorkflows = activeExecutionsInstance.getActiveExecutions();
}
await Container.get(ActiveExecutions).shutdown();

// Finally shut down Event Bus
await Container.get(MessageEventBus).close();
Expand Down
Loading

0 comments on commit c04f92f

Please sign in to comment.