Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

fix(core): Ensure job processor does not reprocess amended executions #11438

Merged
merged 2 commits into from
Oct 29, 2024
Merged

fix(core): Ensure job processor does not reprocess amended executions #11438

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
a74fe5a
fix(core): Ensure job processor does not reprocess amended executions
ivov Oct 28, 2024
3ffc4b7
Remove check from #11282
ivov Oct 28, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
21 changes: 21 additions & 0 deletions packages/cli/src/scaling/__tests__/job-processor.service.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,21 @@
import { mock } from 'jest-mock-extended';

import type { ExecutionRepository } from '@/databases/repositories/execution.repository';
import type { IExecutionResponse } from '@/interfaces';

import { JobProcessor } from '../job-processor';
import type { Job } from '../scaling.types';

describe('JobProcessor', () => {
it('should refrain from processing a crashed execution', async () => {
const executionRepository = mock<ExecutionRepository>();
executionRepository.findSingleExecution.mockResolvedValue(
mock<IExecutionResponse>({ status: 'crashed' }),
);
const jobProcessor = new JobProcessor(mock(), executionRepository, mock(), mock(), mock());

const result = await jobProcessor.processJob(mock<Job>());

expect(result).toEqual({ success: false });
});
});
7 changes: 7 additions & 0 deletions packages/cli/src/scaling/job-processor.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -58,6 +58,13 @@ export class JobProcessor {
);
}

/**
* Bull's implicit retry mechanism and n8n's execution recovery mechanism may
* cause a crashed execution to be enqueued. We refrain from processing it,
* until we have reworked both mechanisms to prevent this scenario.
*/
if (execution.status === 'crashed') return { success: false };
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Are we using the return value for anything?

I only found this function being called here, and there we're not using the result:

n8n/packages/cli/src/scaling/scaling.service.ts

Line 94 in ad29235

await this.jobProcessor.processJob(job);

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Historically we never used the result, we've only relied on promise resolution or rejection. The important bit is the early return to resolve the promise without any reprocessing of the crashed execution. I went for { success: false } to keep the return type simple, aligned with what is already there.


const workflowId = execution.workflowData.id;

this.logger.info(`Worker started execution ${executionId} (job ${job.id})`, {
Expand Down
12 changes: 0 additions & 12 deletions packages/cli/src/workflow-runner.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,7 +14,6 @@ import type {
IWorkflowExecutionDataProcess,
} from 'n8n-workflow';
import {
ApplicationError,
ErrorReporterProxy as ErrorReporter,
ExecutionCancelledError,
Workflow,
Expand Down Expand Up @@ -381,17 +380,6 @@ export class WorkflowRunner {
let job: Job;
let hooks: WorkflowHooks;
try {
// check to help diagnose PAY-2100
if (
data.executionData?.executionData?.nodeExecutionStack?.length === 0 &&
config.getEnv('deployment.type') === 'internal'
) {
await this.executionRepository.setRunning(executionId); // set `startedAt` so we display it correctly in UI
throw new ApplicationError('Execution to enqueue has empty node execution stack', {
extra: { executionData: data.executionData },
});
}

job = await this.scalingService.addJob(jobData, { priority: realtime ? 50 : 100 });

hooks = WorkflowExecuteAdditionalData.getWorkflowHooksWorkerMain(
Expand Down
Loading