mirror of
https://github.com/n8n-io/n8n.git
synced 2025-03-05 20:50:17 -08:00
fix(core): Handle max stalled count error better (#12824)
This commit is contained in:
parent
648c6f9315
commit
eabf160957
|
@ -5,9 +5,12 @@ import { ApplicationError } from 'n8n-workflow';
|
|||
*/
|
||||
export class MaxStalledCountError extends ApplicationError {
|
||||
constructor(cause: Error) {
|
||||
super('The execution has reached the maximum number of attempts and will no longer retry.', {
|
||||
level: 'warning',
|
||||
cause,
|
||||
});
|
||||
super(
|
||||
'This execution failed to be processed too many times and will no longer retry. To allow this execution to complete, please break down your workflow or scale up your workers or adjust your worker settings.',
|
||||
{
|
||||
level: 'warning',
|
||||
cause,
|
||||
},
|
||||
);
|
||||
}
|
||||
}
|
||||
|
|
|
@ -17,7 +17,6 @@ import config from '@/config';
|
|||
import { HIGHEST_SHUTDOWN_PRIORITY, Time } from '@/constants';
|
||||
import { ExecutionRepository } from '@/databases/repositories/execution.repository';
|
||||
import { OnShutdown } from '@/decorators/on-shutdown';
|
||||
import { MaxStalledCountError } from '@/errors/max-stalled-count.error';
|
||||
import { EventService } from '@/events/event.service';
|
||||
import { OrchestrationService } from '@/services/orchestration.service';
|
||||
import { assertNever } from '@/utils';
|
||||
|
@ -271,10 +270,6 @@ export class ScalingService {
|
|||
this.queue.on('error', (error: Error) => {
|
||||
if ('code' in error && error.code === 'ECONNREFUSED') return; // handled by RedisClientService.retryStrategy
|
||||
|
||||
if (error.message.includes('job stalled more than maxStalledCount')) {
|
||||
throw new MaxStalledCountError(error);
|
||||
}
|
||||
|
||||
/**
|
||||
* Non-recoverable error on worker start with Redis unavailable.
|
||||
* Even if Redis recovers, worker will remain unable to process jobs.
|
||||
|
|
|
@ -37,6 +37,8 @@ import * as WorkflowExecuteAdditionalData from '@/workflow-execute-additional-da
|
|||
import { generateFailedExecutionFromError } from '@/workflow-helpers';
|
||||
import { WorkflowStaticDataService } from '@/workflows/workflow-static-data.service';
|
||||
|
||||
import { MaxStalledCountError } from './errors/max-stalled-count.error';
|
||||
|
||||
@Service()
|
||||
export class WorkflowRunner {
|
||||
private scalingService: ScalingService;
|
||||
|
@ -416,6 +418,13 @@ export class WorkflowRunner {
|
|||
try {
|
||||
await job.finished();
|
||||
} catch (error) {
|
||||
if (
|
||||
error instanceof Error &&
|
||||
error.message.includes('job stalled more than maxStalledCount')
|
||||
) {
|
||||
error = new MaxStalledCountError(error);
|
||||
}
|
||||
|
||||
// We use "getWorkflowHooksWorkerExecuter" as "getWorkflowHooksWorkerMain" does not contain the
|
||||
// "workflowExecuteAfter" which we require.
|
||||
const hooks = getWorkflowHooksWorkerExecuter(
|
||||
|
@ -424,6 +433,7 @@ export class WorkflowRunner {
|
|||
data.workflowData,
|
||||
{ retryOf: data.retryOf ? data.retryOf.toString() : undefined },
|
||||
);
|
||||
|
||||
await this.processError(error, new Date(), data.executionMode, executionId, hooks);
|
||||
|
||||
reject(error);
|
||||
|
|
Loading…
Reference in a new issue