mirror of
https://github.com/n8n-io/n8n.git
synced 2025-03-05 20:50:17 -08:00
fix(core): Fix execution cancellation issues in scaling mode (#12343)
This commit is contained in:
parent
870f8640c7
commit
e26b406665
|
@ -242,8 +242,8 @@ describe('ExecutionService', () => {
|
||||||
*/
|
*/
|
||||||
expect(waitTracker.stopExecution).not.toHaveBeenCalled();
|
expect(waitTracker.stopExecution).not.toHaveBeenCalled();
|
||||||
expect(activeExecutions.stopExecution).toHaveBeenCalled();
|
expect(activeExecutions.stopExecution).toHaveBeenCalled();
|
||||||
expect(scalingService.findJobsByStatus).toHaveBeenCalled();
|
expect(scalingService.findJobsByStatus).not.toHaveBeenCalled();
|
||||||
expect(scalingService.stopJob).toHaveBeenCalled();
|
expect(scalingService.stopJob).not.toHaveBeenCalled();
|
||||||
expect(executionRepository.stopDuringRun).toHaveBeenCalled();
|
expect(executionRepository.stopDuringRun).toHaveBeenCalled();
|
||||||
});
|
});
|
||||||
|
|
||||||
|
@ -268,8 +268,8 @@ describe('ExecutionService', () => {
|
||||||
* Assert
|
* Assert
|
||||||
*/
|
*/
|
||||||
expect(waitTracker.stopExecution).toHaveBeenCalledWith(execution.id);
|
expect(waitTracker.stopExecution).toHaveBeenCalledWith(execution.id);
|
||||||
expect(scalingService.findJobsByStatus).toHaveBeenCalled();
|
expect(scalingService.findJobsByStatus).not.toHaveBeenCalled();
|
||||||
expect(scalingService.stopJob).toHaveBeenCalled();
|
expect(scalingService.stopJob).not.toHaveBeenCalled();
|
||||||
expect(executionRepository.stopDuringRun).toHaveBeenCalled();
|
expect(executionRepository.stopDuringRun).toHaveBeenCalled();
|
||||||
});
|
});
|
||||||
});
|
});
|
||||||
|
|
|
@ -16,7 +16,7 @@ import {
|
||||||
Workflow,
|
Workflow,
|
||||||
WorkflowOperationError,
|
WorkflowOperationError,
|
||||||
} from 'n8n-workflow';
|
} from 'n8n-workflow';
|
||||||
import { Container, Service } from 'typedi';
|
import { Service } from 'typedi';
|
||||||
|
|
||||||
import { ActiveExecutions } from '@/active-executions';
|
import { ActiveExecutions } from '@/active-executions';
|
||||||
import { ConcurrencyControlService } from '@/concurrency/concurrency-control.service';
|
import { ConcurrencyControlService } from '@/concurrency/concurrency-control.service';
|
||||||
|
@ -477,18 +477,6 @@ export class ExecutionService {
|
||||||
this.waitTracker.stopExecution(execution.id);
|
this.waitTracker.stopExecution(execution.id);
|
||||||
}
|
}
|
||||||
|
|
||||||
const { ScalingService } = await import('@/scaling/scaling.service');
|
|
||||||
const scalingService = Container.get(ScalingService);
|
|
||||||
const jobs = await scalingService.findJobsByStatus(['active', 'waiting']);
|
|
||||||
|
|
||||||
const job = jobs.find(({ data }) => data.executionId === execution.id);
|
|
||||||
|
|
||||||
if (job) {
|
|
||||||
await scalingService.stopJob(job);
|
|
||||||
} else {
|
|
||||||
this.logger.debug('Job to stop not in queue', { executionId: execution.id });
|
|
||||||
}
|
|
||||||
|
|
||||||
return await this.executionRepository.stopDuringRun(execution);
|
return await this.executionRepository.stopDuringRun(execution);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
|
@ -66,7 +66,7 @@ export class WorkflowRunner {
|
||||||
//
|
//
|
||||||
// FIXME: This is a quick fix. The proper fix would be to not remove
|
// FIXME: This is a quick fix. The proper fix would be to not remove
|
||||||
// the execution from the active executions while it's still running.
|
// the execution from the active executions while it's still running.
|
||||||
if (error instanceof ExecutionNotFoundError) {
|
if (error instanceof ExecutionNotFoundError || error instanceof ExecutionCancelledError) {
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
|
@ -2,7 +2,7 @@ import type { NodeOptions } from '@sentry/node';
|
||||||
import { close } from '@sentry/node';
|
import { close } from '@sentry/node';
|
||||||
import type { ErrorEvent, EventHint } from '@sentry/types';
|
import type { ErrorEvent, EventHint } from '@sentry/types';
|
||||||
import { AxiosError } from 'axios';
|
import { AxiosError } from 'axios';
|
||||||
import { ApplicationError, type ReportingOptions } from 'n8n-workflow';
|
import { ApplicationError, ExecutionCancelledError, type ReportingOptions } from 'n8n-workflow';
|
||||||
import { createHash } from 'node:crypto';
|
import { createHash } from 'node:crypto';
|
||||||
import { Service } from 'typedi';
|
import { Service } from 'typedi';
|
||||||
|
|
||||||
|
@ -143,6 +143,7 @@ export class ErrorReporter {
|
||||||
}
|
}
|
||||||
|
|
||||||
error(e: unknown, options?: ReportingOptions) {
|
error(e: unknown, options?: ReportingOptions) {
|
||||||
|
if (e instanceof ExecutionCancelledError) return;
|
||||||
const toReport = this.wrap(e);
|
const toReport = this.wrap(e);
|
||||||
if (toReport) this.report(toReport, options);
|
if (toReport) this.report(toReport, options);
|
||||||
}
|
}
|
||||||
|
|
Loading…
Reference in a new issue