mudhorn/n8n
1
0
Fork 0
mirror of https://github.com/n8n-io/n8n.git synced 2024-12-25 04:34:06 -08:00
Code Issues Actions Packages Projects Releases Wiki Activity

fix(core): Prevent executions from becoming forever running (#7569)

Browse source 
Fixes CP-867
Possibly also fixes PAY-323 and PAY-412
This commit is contained in:
कारतोफ्फेलस्क्रिप्ट™ 2023-11-01 13:51:13 +01:00 committed by GitHub
parent 0746783e02
commit 9bdb85c4ce
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
4 changed files with 18 additions and 25 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

15
packages/cli/src/InternalHooks.ts
View file

@ -27,7 +27,6 @@ import type { User } from '@db/entities/User';
import { N8N_VERSION } from '@/constants';
import { NodeTypes } from './NodeTypes';
import type { ExecutionMetadata } from '@db/entities/ExecutionMetadata';
import { ExecutionRepository } from '@db/repositories';
import { RoleService } from './services/role.service';
import type { EventPayloadWorkflow } from './eventbus/EventMessageClasses/EventMessageWorkflow';
import { determineFinalExecutionStatus } from './executionLifecycleHooks/shared/sharedHookFunctions';
@ -55,7 +54,6 @@ export class InternalHooks implements IInternalHooksClass {
private telemetry: Telemetry,
private nodeTypes: NodeTypes,
private roleService: RoleService,
private executionRepository: ExecutionRepository,
eventsService: EventsService,
private readonly instanceSettings: InstanceSettings,
) {
@ -256,15 +254,10 @@ export class InternalHooks implements IInternalHooksClass {
workflowName: (data as IWorkflowBase).name,
};
}
void Promise.all([
this.executionRepository.updateExistingExecution(executionId, {
status: 'running',
}),
eventBus.sendWorkflowEvent({
eventName: 'n8n.workflow.started',
payload,
}),
]);
void eventBus.sendWorkflowEvent({
eventName: 'n8n.workflow.started',
payload,
});
}
async onWorkflowCrashed(

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

@ -306,13 +306,9 @@ export class WorkflowRunner {
{ executionId },
);
let workflowExecution: PCancelable<IRun>;
await Container.get(ExecutionRepository).updateStatus(executionId, 'running');
try {
this.logger.verbose(
`Execution for workflow ${data.workflowData.name} was assigned id ${executionId}`,
{ executionId },
);
additionalData.hooks = WorkflowExecuteAdditionalData.getWorkflowHooksMain(
data,
executionId,
@ -701,6 +697,7 @@ export class WorkflowRunner {
const executionId = await this.activeExecutions.add(data, subprocess, restartExecutionId);
(data as unknown as IWorkflowExecutionDataProcessWithExecution).executionId = executionId;
await Container.get(ExecutionRepository).updateStatus(executionId, 'running');
const workflowHooks = WorkflowExecuteAdditionalData.getWorkflowHooksMain(data, executionId);

13
packages/cli/src/commands/worker.ts
View file

@ -112,13 +112,11 @@ export class Worker extends BaseCommand {
async runJob(job: Job, nodeTypes: INodeTypes): Promise<JobResponse> {
const { executionId, loadStaticData } = job.data;
const fullExecutionData = await Container.get(ExecutionRepository).findSingleExecution(
executionId,
{
includeData: true,
unflattenData: true,
},
);
const executionRepository = Container.get(ExecutionRepository);
const fullExecutionData = await executionRepository.findSingleExecution(executionId, {
includeData: true,
unflattenData: true,
});
if (!fullExecutionData) {
this.logger.error(
@ -133,6 +131,7 @@ export class Worker extends BaseCommand {
this.logger.info(
`Start job: ${job.id} (Workflow ID: ${workflowId} | Execution: ${executionId})`,
);
await executionRepository.updateStatus(executionId, 'running');
const workflowOwner = await Container.get(OwnershipService).getWorkflowOwnerCached(workflowId);

8
packages/cli/src/databases/repositories/execution.repository.ts
View file

@ -17,7 +17,7 @@ import type {
SelectQueryBuilder,
} from 'typeorm';
import { parse, stringify } from 'flatted';
import type { IExecutionsSummary, IRunExecutionData } from 'n8n-workflow';
import type { ExecutionStatus, IExecutionsSummary, IRunExecutionData } from 'n8n-workflow';
import { BinaryDataService } from 'n8n-core';
import type {
ExecutionPayload,
@ -298,11 +298,15 @@ export class ExecutionRepository extends Repository<ExecutionEntity> {
return Promise.all([this.delete(ids.executionId), this.binaryDataService.deleteMany([ids])]);
}
async updateStatus(executionId: string, status: ExecutionStatus) {
await this.update({ id: executionId }, { status });
}
async updateExistingExecution(executionId: string, execution: Partial<IExecutionResponse>) {
// Se isolate startedAt because it must be set when the execution starts and should never change.
// So we prevent updating it, if it's sent (it usually is and causes problems to executions that
// are resumed after waiting for some time, as a new startedAt is set)
const { id, data, workflowData, startedAt, ...executionInformation } = execution;
const { id, data, workflowId, workflowData, startedAt, ...executionInformation } = execution;
if (Object.keys(executionInformation).length > 0) {
await this.update({ id: executionId }, executionInformation);
}