fix(core): Pass storedAt into WorkflowExecute at every callsite (#40495)

This commit is contained in:
Iván Ovejero
2026-10-07 11:21:18 +00:00
committed by GitHub
parent dc33c29d1a
commit 5ab2187416
16 changed files with 170 additions and 28 deletions
@@ -1,6 +1,11 @@
import { TOOL_EXECUTOR_NODE_NAME } from '@n8n/constants';
import * as core from 'n8n-core';
import { DirectedGraph, recreateNodeExecutionStack, WorkflowExecute } from 'n8n-core';
import {
DirectedGraph,
recreateNodeExecutionStack,
type StorageConfig,
WorkflowExecute,
} from 'n8n-core';
import { NodeHelpers, UserError } from 'n8n-workflow';
import type {
Workflow,
@@ -26,7 +31,10 @@ import { ManualExecutionService } from '@/manual-execution.service';
vi.mock('n8n-core');
describe('ManualExecutionService', () => {
const manualExecutionService = new ManualExecutionService(mock());
const manualExecutionService = new ManualExecutionService(
mock(),
mock<StorageConfig>({ modeTag: 'fs' }),
);
describe('getExecutionStartNode', () => {
it('Should return undefined', () => {
@@ -189,6 +197,7 @@ describe('ManualExecutionService', () => {
waitingExecutionSource,
},
}),
'fs',
);
});
@@ -240,6 +249,7 @@ describe('ManualExecutionService', () => {
resultData: expect.any(Object),
executionData: expect.any(Object),
}),
'fs',
);
});
@@ -534,6 +544,7 @@ describe('ManualExecutionService', () => {
pinData: mockPinData,
}),
}),
'fs',
);
});
it('should call runPartialWorkflow2 with runData and empty startNodes', async () => {
@@ -48,6 +48,7 @@ describe('WaitTracker', () => {
const execution = mock<IExecutionResponse>({
id: '123',
finished: false,
storedAt: 'fs',
waitTill: new Date(Date.now() + 1000),
mode: 'manual',
data: mock({
@@ -217,7 +218,7 @@ describe('WaitTracker', () => {
},
false,
false,
{ executionId: execution.id, expectedStatus: 'waiting' },
{ executionId: execution.id, expectedStatus: 'waiting', storedAt: 'fs' },
);
});
@@ -264,7 +265,7 @@ describe('WaitTracker', () => {
}),
false,
false,
{ executionId: execution.id, expectedStatus: 'waiting' },
{ executionId: execution.id, expectedStatus: 'waiting', storedAt: 'fs' },
);
});
@@ -288,6 +289,7 @@ describe('WaitTracker', () => {
const parentExecution = mock<IExecutionResponse>({
id: 'parent_execution_id',
finished: false,
storedAt: 'fs',
status: 'waiting',
data: createRunExecutionData({
executionData: { nodeExecutionStack: parentStack() },
@@ -376,7 +378,7 @@ describe('WaitTracker', () => {
},
false,
false,
{ executionId: execution.id, expectedStatus: 'waiting' },
{ executionId: execution.id, expectedStatus: 'waiting', storedAt: 'fs' },
);
// ACT 1
@@ -397,7 +399,7 @@ describe('WaitTracker', () => {
},
false,
false,
{ executionId: parentExecution.id, expectedStatus: 'waiting' },
{ executionId: parentExecution.id, expectedStatus: 'waiting', storedAt: 'fs' },
);
});
@@ -446,7 +448,7 @@ describe('WaitTracker', () => {
}),
false,
false,
{ executionId: parentExecution.id, expectedStatus: 'waiting' },
{ executionId: parentExecution.id, expectedStatus: 'waiting', storedAt: 'fs' },
);
});
@@ -521,6 +523,7 @@ describe('WaitTracker', () => {
expect(workflowRunner.run).toHaveBeenNthCalledWith(1, expect.any(Object), false, false, {
executionId: execution.id,
expectedStatus: 'waiting',
storedAt: 'fs',
});
// ACT 2
@@ -561,6 +564,7 @@ describe('WaitTracker', () => {
expect(workflowRunner.run).toHaveBeenNthCalledWith(2, expect.any(Object), false, false, {
executionId: parentExecution.id,
expectedStatus: 'waiting',
storedAt: 'db',
});
});
@@ -632,7 +636,7 @@ describe('WaitTracker', () => {
}),
false,
false,
{ executionId: childExecution.id, expectedStatus: 'waiting' },
{ executionId: childExecution.id, expectedStatus: 'waiting', storedAt: 'fs' },
);
});
@@ -659,7 +663,7 @@ describe('WaitTracker', () => {
}),
false,
false,
{ executionId: execution.id, expectedStatus: 'waiting' },
{ executionId: execution.id, expectedStatus: 'waiting', storedAt: 'fs' },
);
// ACT 2 - Child execution goes into waiting state
@@ -765,6 +769,7 @@ describe('WaitTracker', () => {
expect(workflowRunner.run).toHaveBeenNthCalledWith(2, expect.any(Object), false, false, {
executionId: parentExecution.id,
expectedStatus: 'waiting',
storedAt: 'fs',
});
expect(workflowRunner.run).toHaveBeenCalledTimes(4);
expect(logger.error).toHaveBeenCalled();
@@ -793,6 +798,7 @@ describe('WaitTracker', () => {
expect(workflowRunner.run).toHaveBeenLastCalledWith(expect.any(Object), false, false, {
executionId: parentExecution.id,
expectedStatus: 'waiting',
storedAt: 'fs',
});
expect(logger.error).not.toHaveBeenCalled();
});
@@ -817,6 +823,7 @@ describe('WaitTracker', () => {
expect(workflowRunner.run).toHaveBeenCalledWith(expect.any(Object), false, false, {
executionId: parentExecution.id,
expectedStatus: 'waiting',
storedAt: 'fs',
});
expect(logger.error).toHaveBeenCalled();
});
@@ -1021,6 +1028,7 @@ describe('WaitTracker', () => {
expect(workflowRunner.run).toHaveBeenCalledWith(expect.any(Object), false, false, {
executionId: parentId,
expectedStatus: 'waiting',
storedAt: 'db',
});
expect(logger.error).not.toHaveBeenCalled();
},
@@ -1130,6 +1138,7 @@ describe('WaitTracker', () => {
expect(workflowRunner.run).toHaveBeenLastCalledWith(expect.any(Object), false, false, {
executionId: parentId,
expectedStatus: 'waiting',
storedAt: 'db',
});
expect(logger.error).not.toHaveBeenCalled();
});
@@ -1182,6 +1191,7 @@ describe('WaitTracker', () => {
expect(workflowRunner.run).toHaveBeenCalledWith(expect.any(Object), false, false, {
executionId: parentId,
expectedStatus: 'waiting',
storedAt: 'db',
});
});
});
@@ -15,7 +15,7 @@ import { createExecution } from '@test-integration/db/executions';
import { createUser } from '@test-integration/db/users';
import { setupTestServer } from '@test-integration/utils';
import type { Response } from 'express';
import { DirectedGraph, WorkflowExecute, WorkflowHasIssuesError } from 'n8n-core';
import { DirectedGraph, StorageConfig, WorkflowExecute, WorkflowHasIssuesError } from 'n8n-core';
import * as core from 'n8n-core';
import {
type IExecuteData,
@@ -879,6 +879,65 @@ describe('run', () => {
expect(addSpy).toHaveBeenCalledWith(data, existingExecution);
});
describe('storedAt', () => {
function arrangeRun() {
const activeExecutions = Container.get(ActiveExecutions);
vi.spyOn(activeExecutions, 'add').mockResolvedValue('1');
vi.spyOn(activeExecutions, 'attachWorkflowExecution').mockReturnValueOnce();
vi.spyOn(Container.get(CredentialsPermissionChecker), 'check').mockResolvedValueOnce();
vi.spyOn(WorkflowExecuteAdditionalData, 'getBase').mockResolvedValue(
mock<IWorkflowExecuteAdditionalData>(),
);
const processRunExecutionData = vi
.spyOn(WorkflowExecute.prototype, 'processRunExecutionData')
.mockReturnValueOnce(new PCancelable(() => mock<IRun>()));
const data = mock<IWorkflowExecutionDataProcess>({
executionMode: 'webhook',
workflowData: { nodes: [], id: 'workflow-id', settings: undefined, staticData: {} },
executionData: createRunExecutionData({}),
triggerToStartFrom: undefined,
startNodes: undefined,
destinationNode: undefined,
});
return {
data,
getStoredAt: () => {
const [workflowExecute] = processRunExecutionData.mock.contexts as WorkflowExecute[];
return workflowExecute.getFullRunData(new Date()).storedAt;
},
};
}
it('passes the storedAt of a resumed execution into WorkflowExecute', async () => {
const { data, getStoredAt } = arrangeRun();
await runner.run(data, undefined, false, {
executionId: '1',
expectedStatus: 'waiting',
storedAt: 'fs',
});
expect(getStoredAt()).toBe('fs');
});
it('passes the configured mode into WorkflowExecute for a new execution', async () => {
const storageConfig = Container.get(StorageConfig);
storageConfig.mode = 's3';
try {
const { data, getStoredAt } = arrangeRun();
await runner.run(data);
expect(getStoredAt()).toBe('s3');
} finally {
storageConfig.mode = 'database';
}
});
});
describe('engine v2 dispatch', () => {
it('hands the run to the dispatcher without registering a control-plane execution', async () => {
// ARRANGE
@@ -54,7 +54,12 @@ describe('ChatExecutionManager', () => {
describe('runWorkflow', () => {
it('should call WorkflowRunner.run with correct parameters', async () => {
const execution = { id: '1', workflowData: {}, data: {} } as IExecutionResponse;
const execution = {
id: '1',
workflowData: {},
data: {},
storedAt: 'fs',
} as IExecutionResponse;
const message = {
sessionId: '123',
action: 'sendMessage',
@@ -70,6 +75,7 @@ describe('ChatExecutionManager', () => {
expect(workflowRunner.run).toHaveBeenCalledWith(runData, true, true, {
executionId: '1',
expectedStatus: 'waiting',
storedAt: 'fs',
});
});
@@ -56,6 +56,7 @@ export class ChatExecutionManager {
await this.workflowRunner.run(await this.getRunData(execution, message), true, true, {
executionId: execution.id,
expectedStatus: 'waiting',
storedAt: execution.storedAt,
});
}
@@ -16,7 +16,12 @@ import type { WorkflowRunner } from '@/workflow-runner';
const project = mock<Project>({ id: 'project-1' });
const enqueuedExecution = (id: string) =>
mock<IExecutionResponse>({ id, mode: 'webhook', workflowId: `workflow-for-${id}` });
mock<IExecutionResponse>({
id,
mode: 'webhook',
workflowId: `workflow-for-${id}`,
storedAt: 'fs',
});
describe('EnqueuedExecutionRecoveryService', () => {
const logger = mockLogger();
@@ -71,11 +76,11 @@ describe('EnqueuedExecutionRecoveryService', () => {
expect.objectContaining({ executionMode: 'webhook', projectId: 'project-1' }),
undefined,
false,
{ executionId: '1', expectedStatus: 'new' },
{ executionId: '1', expectedStatus: 'new', storedAt: 'fs' },
);
expect(workflowRunner.run.mock.calls.map((call) => call[3])).toEqual([
{ executionId: '1', expectedStatus: 'new' },
{ executionId: '2', expectedStatus: 'new' },
{ executionId: '1', expectedStatus: 'new', storedAt: 'fs' },
{ executionId: '2', expectedStatus: 'new', storedAt: 'fs' },
]);
});
@@ -121,7 +126,7 @@ describe('EnqueuedExecutionRecoveryService', () => {
expect.anything(),
undefined,
false,
{ executionId: '1', expectedStatus: 'new' },
{ executionId: '1', expectedStatus: 'new', storedAt: 'fs' },
);
});
@@ -165,6 +170,7 @@ describe('EnqueuedExecutionRecoveryService', () => {
{
executionId: '2',
expectedStatus: 'new',
storedAt: 'fs',
},
);
});
@@ -87,7 +87,11 @@ export class EnqueuedExecutionRecoveryService {
// do not block - each execution either runs concurrently or is queued
void this.workflowRunner
.run(data, undefined, false, { executionId, expectedStatus: 'new' })
.run(data, undefined, false, {
executionId,
expectedStatus: 'new',
storedAt: execution.storedAt,
})
.catch(async (error) => await this.crashUnstartableExecution(executionId, error));
} catch (error) {
await this.crashUnstartableExecution(executionId, error);
+3
View File
@@ -23,6 +23,7 @@ import type {
IWorkflowExecutionDataProcess,
IExecutionContext,
WorkflowExecutionSource,
ExecutionStorageLocation,
} from 'n8n-workflow';
import type PCancelable from 'p-cancelable';
@@ -141,6 +142,8 @@ export interface ResumableExecution {
executionId: string;
/** `waiting` for a row being resumed, `new` for one enqueued before a restart. */
expectedStatus: Extract<ExecutionStatus, 'new' | 'waiting'>;
/** Store recorded on the row. A resume keeps writing there, whatever the configured mode. */
storedAt?: ExecutionStorageLocation;
}
export interface IExecutingWorkflowData {
+20 -3
View File
@@ -6,11 +6,13 @@ import {
DirectedGraph,
filterDisabledNodes,
recreateNodeExecutionStack,
StorageConfig,
WorkflowExecute,
rewireGraph,
} from 'n8n-core';
import { NodeHelpers, UserError, createRunExecutionData } from 'n8n-workflow';
import type {
ExecutionStorageLocation,
IExecuteData,
IPinData,
IRun,
@@ -24,7 +26,10 @@ import type PCancelable from 'p-cancelable';
@Service()
export class ManualExecutionService {
constructor(private readonly logger: Logger) {}
constructor(
private readonly logger: Logger,
private readonly storageConfig: StorageConfig,
) {}
getExecutionStartNode(data: IWorkflowExecutionDataProcess, workflow: Workflow) {
let startNode;
@@ -52,6 +57,7 @@ export class ManualExecutionService {
additionalData: IWorkflowExecuteAdditionalData,
executionId: string,
pinData?: IPinData,
storedAt: ExecutionStorageLocation = this.storageConfig.modeTag,
): PCancelable<IRun> {
if (data.triggerToStartFrom?.data && data.startNodes?.length) {
this.logger.debug(
@@ -101,6 +107,7 @@ export class ManualExecutionService {
additionalData,
data.executionMode,
executionData,
storedAt,
);
return workflowExecute.processRunExecutionData(workflow);
} else if (data.runData === undefined || data.executionMode === 'evaluation') {
@@ -168,7 +175,12 @@ export class ManualExecutionService {
}
// Can execute without webhook so go on
const workflowExecute = new WorkflowExecute(additionalData, data.executionMode);
const workflowExecute = new WorkflowExecute(
additionalData,
data.executionMode,
undefined,
storedAt,
);
return workflowExecute.run({
workflow,
startNode,
@@ -186,7 +198,12 @@ export class ManualExecutionService {
// Partial Execution
this.logger.debug(`Execution ID ${executionId} is a partial execution.`, { executionId });
// Execute only the nodes between start and destination nodes
const workflowExecute = new WorkflowExecute(additionalData, data.executionMode);
const workflowExecute = new WorkflowExecute(
additionalData,
data.executionMode,
undefined,
storedAt,
);
return workflowExecute.runPartialWorkflow2(
workflow,
@@ -308,6 +308,7 @@ describe('JobProcessor', () => {
const pinData: IPinData = { pinned: [] };
const execution = mock<IExecutionResponse>({
mode: 'manual',
storedAt: 's3',
workflowData: { id: 'workflow-id', nodes: [], pinData, staticData: {} },
data: mock<IRunExecutionData>({
resultData: {
@@ -357,6 +358,7 @@ describe('JobProcessor', () => {
additionalData,
executionId,
pinData,
's3',
);
});
@@ -461,6 +463,7 @@ describe('JobProcessor', () => {
executionPersistence.findSingleExecution.mockResolvedValue(
mock<IExecutionResponse>({
mode,
storedAt: 's3',
workflowData: { nodes: [], staticData: {} },
data: executionData,
}),
@@ -486,7 +489,7 @@ describe('JobProcessor', () => {
await jobProcessor.processJob(mock<Job>());
// Assert the constructor and method were called
expect(MockedWorkflowExecute).toHaveBeenCalledWith(additionalData, mode, executionData);
expect(MockedWorkflowExecute).toHaveBeenCalledWith(additionalData, mode, executionData, 's3');
expect(processRunExecutionDataMock).toHaveBeenCalled();
},
);
+7 -1
View File
@@ -308,7 +308,12 @@ export class JobProcessor {
const { startData, resultData, manualData } = execution.data;
if (execution.data?.executionData) {
workflowExecute = new WorkflowExecute(additionalData, execution.mode, execution.data);
workflowExecute = new WorkflowExecute(
additionalData,
execution.mode,
execution.data,
execution.storedAt,
);
workflowRun = workflowExecute.processRunExecutionData(workflow);
} else {
const data: IWorkflowExecutionDataProcess = {
@@ -331,6 +336,7 @@ export class JobProcessor {
additionalData,
executionId,
resultData.pinData,
execution.storedAt,
);
} catch (error) {
if (error instanceof WorkflowHasIssuesError) {
+1
View File
@@ -184,6 +184,7 @@ export class WaitTracker {
await this.workflowRunner.run(data, false, false, {
executionId,
expectedStatus: 'waiting',
storedAt: fullExecutionData.storedAt,
});
const { parentExecution } = fullExecutionData.data;
@@ -445,6 +445,8 @@ export class WaitingWebhooks implements IWebhookManager {
}
resolve(data);
},
undefined,
{ storedAt: execution.storedAt },
).catch(reject); // ensure the Promise settles even if executeWebhook throws
});
} finally {
+6 -1
View File
@@ -39,6 +39,7 @@ import type {
WebhookResponseData,
IDestinationNode,
IUser,
ExecutionStorageLocation,
} from 'n8n-workflow';
import {
CHAT_TRIGGER_NODE_TYPE,
@@ -975,6 +976,8 @@ export async function executeWebhook(
* a node that establishes its own carrier below still wins.
*/
encryptedRunnerIdentity?: string;
/** Store recorded on the execution being resumed. Unset for a new execution. */
storedAt?: ExecutionStorageLocation;
},
): Promise<string | undefined> {
const responder = new WebhookResponder(responseCallback);
@@ -1291,7 +1294,9 @@ export async function executeWebhook(
true,
!responder.hasResponded && !shouldDeferOnReceivedResponse,
// An execution id here means we are resuming one that is waiting on this webhook
executionId ? { executionId, expectedStatus: 'waiting' } : undefined,
executionId
? { executionId, expectedStatus: 'waiting', storedAt: options?.storedAt }
: undefined,
responsePromise,
);
@@ -9,7 +9,7 @@ import { Time } from '@n8n/constants';
import { ExecutionRepository, WorkflowRepository } from '@n8n/db';
import { Container } from '@n8n/di';
import type { JSONSchema7 } from 'json-schema';
import { ExternalSecretsProxy, WorkflowExecute } from 'n8n-core';
import { ExternalSecretsProxy, StorageConfig, WorkflowExecute } from 'n8n-core';
import type {
AiEvent,
EnvProviderState,
@@ -696,6 +696,7 @@ async function startExecution(
additionalDataIntegrated,
runData.executionMode,
runExecutionData,
Container.get(StorageConfig).modeTag,
);
const execution = workflowExecute.processRunExecutionData(workflow);
activeExecutions.attachWorkflowExecution(executionId, execution);
+11 -4
View File
@@ -21,6 +21,7 @@ import {
} from 'n8n-core';
import type {
ExecutionError,
ExecutionStorageLocation,
IExecuteResponsePromiseData,
INode,
IPinData,
@@ -135,6 +136,7 @@ export class WorkflowRunner {
executionMode: WorkflowExecuteMode,
executionId: string,
hooks?: ExecutionLifecycleHooks,
storedAt: ExecutionStorageLocation = this.storageConfig.modeTag,
) {
// This means the execution was probably cancelled and has already
// been cleaned up.
@@ -311,7 +313,7 @@ export class WorkflowRunner {
startedAt,
stoppedAt: new Date(),
status: error instanceof MaxStalledCountError ? 'crashed' : 'error',
storedAt: this.storageConfig.modeTag,
storedAt,
};
// Remove from active execution with empty data. That will
@@ -478,7 +480,7 @@ export class WorkflowRunner {
executionId,
data,
shouldReloadStaticData,
existingExecution?.executionId,
existingExecution,
executionWorkflow,
);
}
@@ -563,7 +565,7 @@ export class WorkflowRunner {
executionId: string,
data: IWorkflowExecutionDataProcess,
loadStaticData?: boolean,
restartExecutionId?: string,
existingExecution?: ResumableExecution,
executionWorkflow?: Workflow,
): Promise<void> {
const workflowId = data.workflowData.id;
@@ -605,13 +607,15 @@ export class WorkflowRunner {
workflowTimeout <= 0 ? undefined : Date.now() + workflowTimeout * 1000,
workflowSettings,
});
additionalData.restartExecutionId = restartExecutionId;
additionalData.restartExecutionId = existingExecution?.executionId;
additionalData.streamingEnabled = data.streamingEnabled;
additionalData.encryptedRunnerIdentity = data.encryptedRunnerIdentity;
additionalData.executionId = executionId;
additionalData.evaluationRunId = data.evaluationRunId;
const storedAt = existingExecution?.storedAt ?? this.storageConfig.modeTag;
this.logger.debug(
`Execution for workflow ${data.workflowData.name} was assigned id ${executionId}`,
{ executionId },
@@ -654,6 +658,7 @@ export class WorkflowRunner {
additionalData,
data.executionMode,
data.executionData,
storedAt,
);
workflowExecution = workflowExecute.processRunExecutionData(workflow);
} else {
@@ -711,6 +716,7 @@ export class WorkflowRunner {
data.executionMode,
executionId,
additionalData.hooks,
storedAt,
),
);
} catch (error) {
@@ -725,6 +731,7 @@ export class WorkflowRunner {
data.executionMode,
executionId,
additionalData.hooks,
storedAt,
);
throw error;