mudhorn/n8n
1
0
Fork 0
mirror of https://github.com/n8n-io/n8n.git synced 2025-01-11 12:57:29 -08:00
Code Issues Actions 2 Packages Projects Releases Wiki Activity

⚡ Simplify bull error handling (#1983)

Browse source
This commit is contained in:
Jan 2021-07-12 10:12:45 +02:00 committed by GitHub
parent 2daece5f9d
commit cce804a534
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
1 changed files with 15 additions and 33 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

48
packages/cli/src/WorkflowRunner.ts
View file

@ -275,11 +275,11 @@ export class WorkflowRunner {
// data to editor-UI is not needed. // data to editor-UI is not needed.
hooks.executeHookFunctions('workflowExecuteBefore', []); hooks.executeHookFunctions('workflowExecuteBefore', []);
} catch (error) { } catch (error) {
// We use "getWorkflowHooksIntegrated" here as we are just integrated in the "workflowExecuteAfter" // We use "getWorkflowHooksWorkerExecuter" as "getWorkflowHooksWorkerMain" does not contain the
// hook anyway and other get so ignored // "workflowExecuteAfter" which we require.
const hooks = WorkflowExecuteAdditionalData.getWorkflowHooksIntegrated(data.executionMode, executionId, data.workflowData, { retryOf: data.retryOf ? data.retryOf.toString() : undefined }); 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); await this.processError(error, new Date(), data.executionMode, executionId, hooks);
return executionId; throw error;
} }
const workflowExecution: PCancelable<IRun> = new PCancelable(async (resolve, reject, onCancel) => { const workflowExecution: PCancelable<IRun> = new PCancelable(async (resolve, reject, onCancel) => {
@ -287,19 +287,14 @@ export class WorkflowRunner {
onCancel(async () => { onCancel(async () => {
await Queue.getInstance().stopJob(job); await Queue.getInstance().stopJob(job);
const fullRunData :IRun = { // We use "getWorkflowHooksWorkerExecuter" as "getWorkflowHooksWorkerMain" does not contain the
data: { // "workflowExecuteAfter" which we require.
resultData: { const hooksWorker = WorkflowExecuteAdditionalData.getWorkflowHooksWorkerExecuter(data.executionMode, executionId, data.workflowData, { retryOf: data.retryOf ? data.retryOf.toString() : undefined });
error: new WorkflowOperationError('Workflow-Execution has been canceled!'),
runData: {}, const error = new WorkflowOperationError('Workflow-Execution has been canceled!');
}, await this.processError(error, new Date(), data.executionMode, executionId, hooksWorker);
},
mode: data.executionMode, reject(error);
startedAt: new Date(),
stoppedAt: new Date(),
};
this.activeExecutions.remove(executionId, fullRunData);
resolve(fullRunData);
}); });
const jobData: Promise<IBullJobResponse> = job.finished(); const jobData: Promise<IBullJobResponse> = job.finished();
@ -351,6 +346,8 @@ export class WorkflowRunner {
clearWatchdogInterval(); clearWatchdogInterval();
} }
} catch (error) { } catch (error) {
// We use "getWorkflowHooksWorkerExecuter" as "getWorkflowHooksWorkerMain" does not contain the
// "workflowExecuteAfter" which we require.
const hooks = WorkflowExecuteAdditionalData.getWorkflowHooksWorkerExecuter(data.executionMode, executionId, data.workflowData, { retryOf: data.retryOf ? data.retryOf.toString() : undefined }); 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.`); Logger.error(`Problem with execution ${executionId}: ${error.message}. Aborting.`);
if (clearWatchdogInterval !== undefined) { if (clearWatchdogInterval !== undefined) {
@ -358,24 +355,9 @@ export class WorkflowRunner {
} }
await this.processError(error, new Date(), data.executionMode, executionId, hooks); await this.processError(error, new Date(), data.executionMode, executionId, hooks);
const fullRunData :IRun = { reject(error);
data: {
resultData: {
error,
runData: {},
},
},
mode: data.executionMode,
startedAt: new Date(),
stoppedAt: new Date(),
};
this.activeExecutions.remove(executionId, fullRunData);
resolve(fullRunData);
return;
} }
const executionDb = await Db.collections.Execution!.findOne(executionId) as IExecutionFlattedDb; const executionDb = await Db.collections.Execution!.findOne(executionId) as IExecutionFlattedDb;
const fullExecutionData = ResponseHelper.unflattenExecutionData(executionDb) as IExecutionResponse; const fullExecutionData = ResponseHelper.unflattenExecutionData(executionDb) as IExecutionResponse;
const runData = { const runData = {