additional tests for saveExecutionProgress

packages/cli/src/execution-lifecycle/__tests__/save-execution-progress.test.ts

@@ -8,63 +8,142 @@ import { mockInstance } from '@test/mocking';
 
 import { saveExecutionProgress } from '../save-execution-progress';
 
-mockInstance(Logger);
+describe('saveExecutionProgress', () => {
-const errorReporter = mockInstance(ErrorReporter);
+	mockInstance(Logger);
-const executionRepository = mockInstance(ExecutionRepository);
+	const errorReporter = mockInstance(ErrorReporter);
+	const executionRepository = mockInstance(ExecutionRepository);
 
-afterEach(() => {
+	afterEach(() => {
-	jest.clearAllMocks();
+		jest.resetAllMocks();
-});
+	});
 
-const commonArgs: [string, string, string, ITaskData, IRunExecutionData] = [
+	const commonArgs: [string, string, string, ITaskData, IRunExecutionData] = [
-	'some-workflow-id',
+		'some-workflow-id',
-	'some-execution-id',
+		'some-execution-id',
-	'My Node',
+		'My Node',
-	{} as ITaskData,
+		{} as ITaskData,
-	{} as IRunExecutionData,
+		{} as IRunExecutionData,
-];
+	];
 
-test('should ignore on leftover async call', async () => {
+	test('should not try to update non-existent executions', async () => {
-	executionRepository.findSingleExecution.mockResolvedValue({
+		executionRepository.findSingleExecution.mockResolvedValue(undefined);
-		finished: true,
-	} as IExecutionResponse);
 
-	await saveExecutionProgress(...commonArgs);
+		await saveExecutionProgress(...commonArgs);
+		expect(executionRepository.updateExistingExecution).not.toHaveBeenCalled();
+	});
 
-	expect(executionRepository.updateExistingExecution).not.toHaveBeenCalled();
+	test('should handle DB errors on execution lookup', async () => {
-});
+		const error = new Error('Something went wrong');
+		executionRepository.findSingleExecution.mockImplementation(() => {
+			throw error;
+		});
 
-test('should update execution when saving progress is enabled', async () => {
+		await saveExecutionProgress(...commonArgs);
-	executionRepository.findSingleExecution.mockResolvedValue({} as IExecutionResponse);
 
-	await saveExecutionProgress(...commonArgs);
+		expect(executionRepository.updateExistingExecution).not.toHaveBeenCalled();
+		expect(errorReporter.error).toHaveBeenCalledWith(error);
+	});
 
-	expect(executionRepository.updateExistingExecution).toHaveBeenCalledWith('some-execution-id', {
+	test('should handle DB errors when updating the execution', async () => {
-		data: {
+		const error = new Error('Something went wrong');
-			executionData: undefined,
+		executionRepository.findSingleExecution.mockResolvedValue({} as IExecutionResponse);
-			resultData: {
+		executionRepository.updateExistingExecution.mockImplementation(() => {
-				lastNodeExecuted: 'My Node',
+			throw error;
-				runData: {
+		});
-					'My Node': [{}],
+
+		await saveExecutionProgress(...commonArgs);
+
+		expect(executionRepository.findSingleExecution).toHaveBeenCalled();
+		expect(executionRepository.updateExistingExecution).toHaveBeenCalled();
+		expect(errorReporter.error).toHaveBeenCalledWith(error);
+	});
+
+	test('should not try to update finished executions', async () => {
+		executionRepository.findSingleExecution.mockResolvedValue({
+			finished: true,
+		} as IExecutionResponse);
+
+		await saveExecutionProgress(...commonArgs);
+
+		expect(executionRepository.updateExistingExecution).not.toHaveBeenCalled();
+	});
+
+	test('should populate `.data` when it is missing', async () => {
+		const fullExecutionData = {} as IExecutionResponse;
+		executionRepository.findSingleExecution.mockResolvedValue(fullExecutionData);
+
+		await saveExecutionProgress(...commonArgs);
+
+		expect(fullExecutionData).toEqual({
+			data: {
+				executionData: undefined,
+				resultData: {
+					lastNodeExecuted: 'My Node',
+					runData: {
+						'My Node': [{}],
+					},
+				},
+				startData: {},
+			},
+			status: 'running',
+		});
+
+		expect(executionRepository.updateExistingExecution).toHaveBeenCalledWith(
+			'some-execution-id',
+			fullExecutionData,
+		);
+
+		expect(errorReporter.error).not.toHaveBeenCalled();
+	});
+
+	test('should augment `.data` if it already exists', async () => {
+		const fullExecutionData = {
+			data: {
+				startData: {},
+				resultData: {
+					runData: {
+						'My Node': [{}],
+					},
 				},
 			},
-			startData: {},
+		} as unknown as IExecutionResponse;
-		},
+		executionRepository.findSingleExecution.mockResolvedValue(fullExecutionData);
-		status: 'running',
+
+		await saveExecutionProgress(...commonArgs);
+
+		expect(fullExecutionData).toEqual({
+			data: {
+				executionData: undefined,
+				resultData: {
+					lastNodeExecuted: 'My Node',
+					runData: {
+						'My Node': [{}, {}],
+					},
+				},
+				startData: {},
+			},
+			status: 'running',
+		});
+
+		expect(executionRepository.updateExistingExecution).toHaveBeenCalledWith(
+			'some-execution-id',
+			fullExecutionData,
+		);
 	});
 
-	expect(errorReporter.error).not.toHaveBeenCalled();
+	test('should set last executed node correctly', async () => {
-});
+		const fullExecutionData = {
+			data: {
+				resultData: {
+					lastNodeExecuted: 'Another Node',
+					runData: {},
+				},
+			},
+		} as unknown as IExecutionResponse;
+		executionRepository.findSingleExecution.mockResolvedValue(fullExecutionData);
 
-test('should report error on failure', async () => {
+		await saveExecutionProgress(...commonArgs);
-	const error = new Error('Something went wrong');
 
-	executionRepository.findSingleExecution.mockImplementation(() => {
+		expect(fullExecutionData.data.resultData.lastNodeExecuted).toEqual('My Node');
-		throw error;
 	});
-
-	await saveExecutionProgress(...commonArgs);
-
-	expect(executionRepository.updateExistingExecution).not.toHaveBeenCalled();
-	expect(errorReporter.error).toHaveBeenCalledWith(error);
 });

packages/cli/src/execution-lifecycle/save-execution-progress.ts

@@ -12,6 +12,8 @@ export async function saveExecutionProgress(
 	executionData: IRunExecutionData,
 ) {
 	const logger = Container.get(Logger);
+	const executionRepository = Container.get(ExecutionRepository);
+	const errorReporter = Container.get(ErrorReporter);
 
 	try {
 		logger.debug(`Save execution progress to database for execution ID ${executionId} `, {
@@ -19,13 +21,10 @@ export async function saveExecutionProgress(
 			nodeName,
 		});
 
-		const fullExecutionData = await Container.get(ExecutionRepository).findSingleExecution(
+		const fullExecutionData = await executionRepository.findSingleExecution(executionId, {
-			executionId,
+			includeData: true,
-			{
+			unflattenData: true,
-				includeData: true,
+		});
-				unflattenData: true,
-			},
-		);
 
 		if (!fullExecutionData) {
 			// Something went badly wrong if this happens.
@@ -40,29 +39,22 @@ export async function saveExecutionProgress(
 			return;
 		}
 
-		if (fullExecutionData.data === undefined) {
+		fullExecutionData.data ??= {
-			fullExecutionData.data = {
+			startData: {},
-				startData: {},
+			resultData: {
-				resultData: {
+				runData: {},
-					runData: {},
+			},
-				},
+			executionData: {
-				executionData: {
+				contextData: {},
-					contextData: {},
+				metadata: {},
-					metadata: {},
+				nodeExecutionStack: [],
-					nodeExecutionStack: [],
+				waitingExecution: {},
-					waitingExecution: {},
+				waitingExecutionSource: {},
-					waitingExecutionSource: {},
+			},
-				},
+		};
-			};
-		}
 
-		if (Array.isArray(fullExecutionData.data.resultData.runData[nodeName])) {
+		const { runData } = fullExecutionData.data.resultData;
-			// Append data if array exists
+		(runData[nodeName] ??= []).push(data);
-			fullExecutionData.data.resultData.runData[nodeName].push(data);
-		} else {
-			// Initialize array and save data
-			fullExecutionData.data.resultData.runData[nodeName] = [data];
-		}
 
 		fullExecutionData.data.executionData = executionData.executionData;
 
@@ -71,14 +63,11 @@ export async function saveExecutionProgress(
 
 		fullExecutionData.status = 'running';
 
-		await Container.get(ExecutionRepository).updateExistingExecution(
+		await executionRepository.updateExistingExecution(executionId, fullExecutionData);
-			executionId,
-			fullExecutionData,
-		);
 	} catch (e) {
 		const error = e instanceof Error ? e : new Error(`${e}`);
 
-		Container.get(ErrorReporter).error(error);
+		errorReporter.error(error);
 		// TODO: Improve in the future!
 		// Errors here might happen because of database access
 		// For busy machines, we may get "Database is locked" errors.
@@ -86,11 +75,7 @@ export async function saveExecutionProgress(
 		// We do this to prevent crashes and executions ending in `unknown` state.
 		logger.error(
 			`Failed saving execution progress to database for execution ID ${executionId} (hookFunctionsPreExecute, nodeExecuteAfter)`,
-			{
+			{ error, executionId, workflowId },
-				...error,
-				executionId,
-				workflowId,
-			},
 		);
 	}
 }