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

fix(core): Fix ignoring crashed executions without event msgs (#7368)

Browse source 
when the event logs do not contain messages for running executions, the
recovery/crash detection on startup would skip these. this PR fixes
that.
This commit is contained in:
Michael Auerswald 2023-10-11 14:22:43 +02:00 committed by GitHub
parent c77042f2bb
commit 2f4d91b2cd
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
3 changed files with 87 additions and 7 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

36
packages/cli/src/eventbus/MessageEventBus/MessageEventBus.ts
View file

@ -1,6 +1,7 @@
import { LoggerProxy, jsonParse } from 'n8n-workflow';
import type { MessageEventBusDestinationOptions } from 'n8n-workflow';
import type { DeleteResult } from 'typeorm';
import { In } from 'typeorm';
import type {
EventMessageTypes,
EventNamesTypes,
@ -132,7 +133,23 @@ export class MessageEventBus extends EventEmitter {
this.logWriter?.startLogging();
await this.send(unsentAndUnfinished.unsentMessages);
const unfinishedExecutionIds = Object.keys(unsentAndUnfinished.unfinishedExecutions);
let unfinishedExecutionIds = Object.keys(unsentAndUnfinished.unfinishedExecutions);
// if we are in queue mode, running jobs may still be running on a worker despite the main process
// crashing, so we can't just mark them as crashed
if (config.get('executions.mode') !== 'queue') {
const dbUnfinishedExecutionIds = (
await Container.get(ExecutionRepository).find({
where: {
status: In(['running', 'new', 'unknown']),
},
select: ['id'],
})
).map((e) => e.id);
unfinishedExecutionIds = Array.from(
new Set<string>([...unfinishedExecutionIds, ...dbUnfinishedExecutionIds]),
);
}
if (unfinishedExecutionIds.length > 0) {
LoggerProxy.warn(`Found unfinished executions: ${unfinishedExecutionIds.join(', ')}`);
@ -160,11 +177,18 @@ export class MessageEventBus extends EventEmitter {
this.logWriter?.startRecoveryProcess();
for (const executionId of unfinishedExecutionIds) {
LoggerProxy.warn(`Attempting to recover execution ${executionId}`);
await recoverExecutionDataFromEventLogMessages(
executionId,
unsentAndUnfinished.unfinishedExecutions[executionId],
true,
);
if (!unsentAndUnfinished.unfinishedExecutions[executionId]?.length) {
LoggerProxy.debug(
`No event messages found, marking execution ${executionId} as 'crashed'`,
);
await Container.get(ExecutionRepository).markAsCrashed([executionId]);
} else {
await recoverExecutionDataFromEventLogMessages(
executionId,
unsentAndUnfinished.unfinishedExecutions[executionId],
true,
);
}
}
}
// remove the recovery process flag file

2
packages/cli/src/eventbus/MessageEventBus/recoverEvents.ts
View file

@ -11,7 +11,7 @@ import { ExecutionRepository } from '@db/repositories';
export async function recoverExecutionDataFromEventLogMessages(
executionId: string,
messages: EventMessageTypes[],
applyToDb = true,
applyToDb: boolean,
): Promise<IRunExecutionData | undefined> {
const executionEntry = await Container.get(ExecutionRepository).findSingleExecution(executionId, {
includeData: true,

56
packages/cli/test/integration/eventbus.ee.test.ts
View file

@ -24,6 +24,8 @@ import type { MessageEventBusDestinationWebhook } from '@/eventbus/MessageEventB
import type { MessageEventBusDestinationSentry } from '@/eventbus/MessageEventBusDestination/MessageEventBusDestinationSentry.ee';
import { EventMessageAudit } from '@/eventbus/EventMessageClasses/EventMessageAudit';
import type { EventNamesTypes } from '@/eventbus/EventMessageClasses';
import { EventMessageWorkflow } from '@/eventbus/EventMessageClasses/EventMessageWorkflow';
import { EventMessageNode } from '@/eventbus/EventMessageClasses/EventMessageNode';
jest.unmock('@/eventbus/MessageEventBus/MessageEventBus');
jest.mock('axios');
@ -389,3 +391,57 @@ test('DELETE /eventbus/destination delete all destinations by id', async () => {
expect(Object.keys(eventBus.destinations).length).toBe(0);
});
// These two tests are running very flaky on CI due to the logwriter working in a worker
// Mocking everything on the other would defeat the purpose of even testing them... so, skipping in CI for now.
// eslint-disable-next-line n8n-local-rules/no-skipped-tests
test.skip('should not find unfinished executions in recovery process', async () => {
eventBus.logWriter?.putMessage(
new EventMessageWorkflow({
eventName: 'n8n.workflow.started',
payload: { executionId: '509', isManual: false },
}),
);
eventBus.logWriter?.putMessage(
new EventMessageNode({
eventName: 'n8n.node.started',
payload: { executionId: '509', nodeName: 'Set', workflowName: 'test' },
}),
);
eventBus.logWriter?.putMessage(
new EventMessageNode({
eventName: 'n8n.node.finished',
payload: { executionId: '509', nodeName: 'Set', workflowName: 'test' },
}),
);
eventBus.logWriter?.putMessage(
new EventMessageWorkflow({
eventName: 'n8n.workflow.success',
payload: { executionId: '509', success: true },
}),
);
const unfinishedExecutions = await eventBus.getUnfinishedExecutions();
expect(Object.keys(unfinishedExecutions)).toHaveLength(0);
});
// eslint-disable-next-line n8n-local-rules/no-skipped-tests
test.skip('should not find unfinished executions in recovery process', async () => {
eventBus.logWriter?.putMessage(
new EventMessageWorkflow({
eventName: 'n8n.workflow.started',
payload: { executionId: '510', isManual: false },
}),
);
eventBus.logWriter?.putMessage(
new EventMessageNode({
eventName: 'n8n.node.started',
payload: { executionId: '510', nodeName: 'Set', workflowName: 'test' },
}),
);
const unfinishedExecutions = await eventBus.getUnfinishedExecutions();
expect(Object.keys(unfinishedExecutions)).toHaveLength(1);
expect(Object.keys(unfinishedExecutions)).toContain('510');
});