mirror of
https://github.com/n8n-io/n8n.git
synced 2026-10-12 02:19:43 +00:00
981 lines
33 KiB
TypeScript
981 lines
33 KiB
TypeScript
/* eslint-disable @typescript-eslint/no-unsafe-argument */
|
|
|
|
import { Logger } from '@n8n/backend-common';
|
|
import { EventService } from '@n8n/backend-services';
|
|
import { ExecutionsConfig } from '@n8n/config';
|
|
import { Time } from '@n8n/constants';
|
|
import { ExecutionRepository } from '@n8n/db';
|
|
import type { IExecutionResponse } from '@n8n/db';
|
|
import { Container, Service } from '@n8n/di';
|
|
import { ensureError } from '@n8n/utils/errors/ensure-error';
|
|
import type { IDeferredPromise } from '@n8n/utils/promise/deferred-promise';
|
|
import { sleep } from '@n8n/utils/sleep';
|
|
import type { ExecutionLifecycleHooks } from 'n8n-core';
|
|
import {
|
|
ErrorReporter,
|
|
establishExecutionContext,
|
|
InstanceSettings,
|
|
StorageConfig,
|
|
WorkflowExecute,
|
|
} from 'n8n-core';
|
|
import type {
|
|
ExecutionError,
|
|
ExecutionStorageLocation,
|
|
IExecuteResponsePromiseData,
|
|
INode,
|
|
IPinData,
|
|
IRun,
|
|
WorkflowExecuteMode,
|
|
IWorkflowExecutionDataProcess,
|
|
} from 'n8n-workflow';
|
|
import {
|
|
createRunExecutionData,
|
|
ExecutionCancelledError,
|
|
isTerminalExecutionStatus,
|
|
ManualExecutionCancelledError,
|
|
TimeoutExecutionCancelledError,
|
|
Workflow,
|
|
WorkflowOperationError,
|
|
} from 'n8n-workflow';
|
|
import PCancelable from 'p-cancelable';
|
|
|
|
import { ActiveExecutions } from '@/active-executions';
|
|
import { ExecutionNotFoundError } from '@/errors/execution-not-found-error';
|
|
import { MaxStalledCountError } from '@/errors/max-stalled-count.error';
|
|
import { PreExecuteBlockedError } from '@/errors/pre-execute-blocked.error';
|
|
// `no-cycle` still reports a cycle here, but only through the dynamic import
|
|
// in `execute-error-workflow`, which creates no evaluation-order edge.
|
|
|
|
import {
|
|
getLifecycleHooksForRegularMain,
|
|
getLifecycleHooksForScalingWorker,
|
|
getLifecycleHooksForScalingMain,
|
|
} from '@/execution-lifecycle/execution-lifecycle-hooks';
|
|
import { toSaveSettings } from '@/execution-lifecycle/to-save-settings';
|
|
import { ExecutionCrashService } from '@/executions/execution-crash.service';
|
|
import { ExecutionPersistence } from '@/executions/execution-persistence';
|
|
import { FailedRunFactory } from '@/executions/failed-run-factory';
|
|
import {
|
|
CredentialsPermissionChecker,
|
|
WorkflowPreExecute,
|
|
} from '@/executions/pre-execution-checks';
|
|
import { ExternalHooks } from '@/external-hooks';
|
|
import type { ResumableExecution } from '@/interfaces';
|
|
import { ManualExecutionService } from '@/manual-execution.service';
|
|
import { NodeTypes } from '@/node-types';
|
|
import type { PoolConfigService } from '@/scaling/pool-config.service.ee';
|
|
import type { ScalingService } from '@/scaling/scaling.service';
|
|
import type { Job, JobData } from '@/scaling/scaling.types';
|
|
import { EngineV2Dispatcher } from '@/services/engine-v2-dispatcher.service';
|
|
import { StreamingWebhookResponseHeartbeat } from '@/webhooks/streaming-webhook-response-heartbeat';
|
|
import * as WorkflowExecuteAdditionalData from '@/workflow-execute-additional-data';
|
|
import { WorkflowStaticDataService } from '@/workflows/workflow-static-data.service';
|
|
|
|
/** How long to keep rechecking the execution status after a max-stalled-count error before failing the run */
|
|
const MAX_STALLED_COUNT_GRACE_WINDOW_MS = 30 * Time.seconds.toMilliseconds;
|
|
|
|
/** Delay between execution status rechecks inside the max-stalled-count grace window */
|
|
const MAX_STALLED_COUNT_RECHECK_INTERVAL_MS = 1 * Time.seconds.toMilliseconds;
|
|
|
|
/** Rechecks that fit in the grace window, on top of the first read */
|
|
const MAX_STALLED_COUNT_RECHECK_ATTEMPTS = Math.floor(
|
|
MAX_STALLED_COUNT_GRACE_WINDOW_MS / MAX_STALLED_COUNT_RECHECK_INTERVAL_MS,
|
|
);
|
|
|
|
/**
|
|
* Symmetric spread applied to each recheck delay (0.2 = plus or minus 20%). Correlated stalls
|
|
* put every affected execution on the same recheck cadence, so the delay is spread to keep
|
|
* them off a single lockstep read against an already unhealthy instance. Same convention as
|
|
* the scheduler timeline.
|
|
*/
|
|
const MAX_STALLED_COUNT_RECHECK_JITTER_RATIO = 0.2;
|
|
|
|
/**
|
|
* Flush the response through the compression middleware.
|
|
* The `flush` method is added at runtime by the Express `compression` middleware
|
|
* and is not part of the standard Response type.
|
|
*/
|
|
function flushResponse(res: { flush?: () => void }) {
|
|
if (typeof res.flush === 'function') {
|
|
res.flush();
|
|
}
|
|
}
|
|
|
|
@Service()
|
|
export class WorkflowRunner {
|
|
private scalingService: ScalingService;
|
|
|
|
private poolConfigService: PoolConfigService;
|
|
|
|
constructor(
|
|
private readonly logger: Logger,
|
|
private readonly errorReporter: ErrorReporter,
|
|
private readonly activeExecutions: ActiveExecutions,
|
|
private readonly executionRepository: ExecutionRepository,
|
|
private readonly executionPersistence: ExecutionPersistence,
|
|
private readonly workflowStaticDataService: WorkflowStaticDataService,
|
|
private readonly nodeTypes: NodeTypes,
|
|
private readonly credentialsPermissionChecker: CredentialsPermissionChecker,
|
|
private readonly instanceSettings: InstanceSettings,
|
|
private readonly manualExecutionService: ManualExecutionService,
|
|
private readonly failedRunFactory: FailedRunFactory,
|
|
private readonly eventService: EventService,
|
|
private readonly executionsConfig: ExecutionsConfig,
|
|
private readonly storageConfig: StorageConfig,
|
|
private readonly externalHooks: ExternalHooks,
|
|
private readonly engineV2Dispatcher: EngineV2Dispatcher,
|
|
private readonly workflowPreExecute: WorkflowPreExecute,
|
|
private readonly executionCrashService: ExecutionCrashService,
|
|
) {}
|
|
|
|
/** The process did error */
|
|
async processError(
|
|
error: ExecutionError | ExecutionNotFoundError,
|
|
startedAt: Date,
|
|
executionMode: WorkflowExecuteMode,
|
|
executionId: string,
|
|
hooks?: ExecutionLifecycleHooks,
|
|
storedAt: ExecutionStorageLocation = this.storageConfig.modeTag,
|
|
) {
|
|
// This means the execution was probably cancelled and has already
|
|
// been cleaned up.
|
|
//
|
|
// FIXME: This is a quick fix. The proper fix would be to not remove
|
|
// the execution from the active executions while it's still running.
|
|
if (
|
|
error instanceof ExecutionNotFoundError ||
|
|
error instanceof ExecutionCancelledError ||
|
|
(typeof error.message === 'string' && error.message.includes('cancelled'))
|
|
) {
|
|
return;
|
|
}
|
|
|
|
const isQueueMode = this.executionsConfig.mode === 'queue';
|
|
|
|
// in queue mode, first do a sanity run for the edge case that the execution was not marked as stalled
|
|
// by Bull even though it executed successfully, see https://github.com/OptimalBits/bull/issues/1415
|
|
|
|
if (isQueueMode) {
|
|
const isStalled = error instanceof MaxStalledCountError;
|
|
const rechecks = isStalled ? MAX_STALLED_COUNT_RECHECK_ATTEMPTS : 0;
|
|
const recheckUntil = Date.now() + (isStalled ? MAX_STALLED_COUNT_GRACE_WINDOW_MS : 0);
|
|
|
|
for (let recheck = 0; recheck <= rechecks; recheck++) {
|
|
const executionWithoutData = await this.executionRepository.findSingleExecution(
|
|
executionId,
|
|
{ includeData: false },
|
|
);
|
|
const status = executionWithoutData?.status;
|
|
|
|
// A `waiting` row means the worker finished its segment and no resume has claimed
|
|
// the execution yet, so the pause is left intact for the wait tracker to resume.
|
|
if (status === 'success' || status === 'waiting') {
|
|
// false positive, the execution was not lost
|
|
let storedRunData: IRun | undefined;
|
|
|
|
try {
|
|
const fullExecutionData = await this.executionPersistence.findSingleExecution(
|
|
executionId,
|
|
{
|
|
includeData: true,
|
|
unflattenData: true,
|
|
},
|
|
);
|
|
|
|
if (fullExecutionData?.data) {
|
|
storedRunData = {
|
|
// oxlint-disable-next-line typescript/no-deprecated
|
|
finished: fullExecutionData.finished,
|
|
mode: fullExecutionData.mode,
|
|
startedAt: fullExecutionData.startedAt,
|
|
stoppedAt: fullExecutionData.stoppedAt,
|
|
status: fullExecutionData.status,
|
|
waitTill: fullExecutionData.waitTill,
|
|
data: fullExecutionData.data,
|
|
storedAt: fullExecutionData.storedAt,
|
|
};
|
|
}
|
|
|
|
// No lifecycle hooks ran for this execution, so make the retention
|
|
// decision they would have made, regardless of data readability.
|
|
if (fullExecutionData && status === 'success') {
|
|
try {
|
|
const saveSettings = toSaveSettings(fullExecutionData.workflowData?.settings);
|
|
const isManualExecution = fullExecutionData.mode === 'manual';
|
|
if (isManualExecution && !saveSettings.manual) {
|
|
await this.executionRepository.softDelete(executionId);
|
|
} else if (!isManualExecution && !saveSettings.success) {
|
|
await this.executionPersistence.deleteInFlightExecution({
|
|
workflowId: fullExecutionData.workflowId,
|
|
executionId,
|
|
storedAt: fullExecutionData.storedAt,
|
|
});
|
|
}
|
|
} catch (pruneError) {
|
|
this.logger.warn('Could not prune a recovered false-positive success', {
|
|
executionId,
|
|
error: ensureError(pruneError),
|
|
});
|
|
}
|
|
}
|
|
} catch (readError) {
|
|
this.logger.warn('Could not read execution data for a recovered execution', {
|
|
executionId,
|
|
error: ensureError(readError),
|
|
});
|
|
}
|
|
|
|
const unreadableRunData: IRun =
|
|
status === 'waiting'
|
|
? {
|
|
data: createRunExecutionData({ resultData: { runData: {} } }),
|
|
finished: false,
|
|
mode: executionMode,
|
|
startedAt,
|
|
stoppedAt: new Date(),
|
|
status: 'waiting',
|
|
waitTill: executionWithoutData?.waitTill ?? undefined,
|
|
storedAt: this.storageConfig.modeTag,
|
|
}
|
|
: {
|
|
data: createRunExecutionData({
|
|
resultData: {
|
|
error: new WorkflowOperationError(
|
|
`Execution ${executionId} succeeded, but its result could not be read`,
|
|
),
|
|
runData: {},
|
|
},
|
|
}),
|
|
finished: false,
|
|
mode: executionMode,
|
|
startedAt,
|
|
stoppedAt: new Date(),
|
|
status: 'error',
|
|
storedAt: this.storageConfig.modeTag,
|
|
};
|
|
|
|
const runData: IRun = storedRunData ?? unreadableRunData;
|
|
|
|
this.activeExecutions.resolveExecutionResponsePromise(executionId);
|
|
this.activeExecutions.finalizeExecution(executionId, runData);
|
|
|
|
return;
|
|
}
|
|
|
|
// The user cancelled during the grace window; the execution is already
|
|
// finalized as canceled elsewhere, so don't overwrite it with a failure.
|
|
if (status === 'canceled') return;
|
|
|
|
// A terminal status will not change, and a missing row cannot become one, so
|
|
// stop rechecking and fail the run now.
|
|
if (status === undefined || isTerminalExecutionStatus(status)) break;
|
|
|
|
if (recheck >= rechecks || Date.now() >= recheckUntil) break;
|
|
|
|
const jitter =
|
|
MAX_STALLED_COUNT_RECHECK_INTERVAL_MS *
|
|
MAX_STALLED_COUNT_RECHECK_JITTER_RATIO *
|
|
(2 * Math.random() - 1);
|
|
|
|
await sleep(MAX_STALLED_COUNT_RECHECK_INTERVAL_MS + jitter);
|
|
}
|
|
}
|
|
|
|
this.logger.error(`Problem with execution ${executionId}: ${error.message}. Aborting.`);
|
|
this.errorReporter.error(error, { executionId });
|
|
|
|
if (error instanceof MaxStalledCountError) {
|
|
const claimed = await this.executionCrashService.markAsCrashedWithoutCounting(
|
|
executionId,
|
|
'stall',
|
|
);
|
|
if (claimed.length === 0) {
|
|
this.activeExecutions.finalizeExecution(executionId);
|
|
return;
|
|
}
|
|
}
|
|
|
|
const fullRunData: IRun = {
|
|
data: createRunExecutionData({
|
|
resultData: {
|
|
error: {
|
|
...error,
|
|
name: error.constructor.name,
|
|
message: error.message,
|
|
stack: error.stack,
|
|
},
|
|
runData: {},
|
|
},
|
|
}),
|
|
finished: false,
|
|
mode: executionMode,
|
|
startedAt,
|
|
stoppedAt: new Date(),
|
|
status: error instanceof MaxStalledCountError ? 'crashed' : 'error',
|
|
storedAt,
|
|
};
|
|
|
|
// Remove from active execution with empty data. That will
|
|
// set the execution to failed.
|
|
this.activeExecutions.finalizeExecution(executionId, fullRunData);
|
|
|
|
await hooks?.runHook('workflowExecuteAfter', [fullRunData]);
|
|
}
|
|
|
|
/**
|
|
* Persist a failed execution record for an already-registered execution and
|
|
* finalize it, so a pre-flight failure surfaces as a normal failed run.
|
|
*/
|
|
private async failExecution(
|
|
data: IWorkflowExecutionDataProcess,
|
|
executionId: string,
|
|
error: ExecutionError & { node?: INode },
|
|
responsePromise?: IDeferredPromise<IExecuteResponsePromiseData>,
|
|
): Promise<void> {
|
|
const runData = this.failedRunFactory.generateFailedExecutionFromError(
|
|
data.executionMode,
|
|
error,
|
|
error.node,
|
|
);
|
|
const lifecycleHooks = getLifecycleHooksForRegularMain(data, executionId);
|
|
await lifecycleHooks.runHook('workflowExecuteBefore', [undefined, data.executionData]);
|
|
await lifecycleHooks.runHook('workflowExecuteAfter', [runData]);
|
|
responsePromise?.reject(error);
|
|
this.activeExecutions.finalizeExecution(executionId);
|
|
}
|
|
|
|
/**
|
|
* Returns the masking error, if any, having already emptied the trigger-item stack
|
|
* either way.
|
|
*/
|
|
async establishContextForPersistence(
|
|
data: IWorkflowExecutionDataProcess,
|
|
): Promise<(ExecutionError & { node?: INode }) | undefined> {
|
|
// Establish the execution context before persisting to the DB.
|
|
// activeExecutions.add() -> executionPersistence.create() writes
|
|
// data.executionData to the DB; any header masking or runtimeData
|
|
// population must happen before that write so the persisted record
|
|
// does not contain raw trigger-item data (e.g. Authorization headers).
|
|
// The runtimeData early-exit guard in establishExecutionContext keeps
|
|
// the subsequent worker-side call at workflow-execute.ts idempotent.
|
|
// Guard on the inner executionData: in queue mode with manual offload
|
|
// the outer IRunExecutionData is created with `executionData: null`
|
|
// so the trigger-item stack is undefined here; nothing to mask yet,
|
|
// the worker will establish context once it populates the stack.
|
|
let establishContextError: (ExecutionError & { node?: INode }) | undefined;
|
|
if (data.executionData?.executionData) {
|
|
// Deliberately lightweight: no pinData, no staticData loading,
|
|
// no additionalData. establishExecutionContext only needs the
|
|
// workflow's settings (for redactionPolicy) and node lookups.
|
|
// runMainProcess() builds its own fully-configured Workflow for
|
|
// actual execution.
|
|
const contextWorkflow = new Workflow({
|
|
id: data.workflowData.id,
|
|
name: data.workflowData.name,
|
|
nodes: data.workflowData.nodes,
|
|
connections: data.workflowData.connections,
|
|
active: data.workflowData.activeVersionId !== null,
|
|
nodeTypes: this.nodeTypes,
|
|
staticData: data.workflowData.staticData,
|
|
settings: data.workflowData.settings ?? {},
|
|
});
|
|
try {
|
|
await establishExecutionContext(
|
|
contextWorkflow,
|
|
data.executionData,
|
|
{ encryptedRunnerIdentity: data.encryptedRunnerIdentity },
|
|
data.executionMode,
|
|
);
|
|
} catch (error) {
|
|
// Masking may have failed partway through, so the trigger-item
|
|
// stack can still contain raw header data. Drop it before
|
|
// activeExecutions.add() persists the execution row.
|
|
data.executionData.executionData.nodeExecutionStack = [];
|
|
establishContextError = error as ExecutionError & { node?: INode };
|
|
}
|
|
}
|
|
|
|
return establishContextError;
|
|
}
|
|
|
|
/** Run the workflow
|
|
* @param realtime This is used in queue mode to change the priority of an execution, making sure they are picked up quicker.
|
|
*/
|
|
async run(
|
|
data: IWorkflowExecutionDataProcess,
|
|
loadStaticData?: boolean,
|
|
realtime?: boolean,
|
|
existingExecution?: ResumableExecution,
|
|
responsePromise?: IDeferredPromise<IExecuteResponsePromiseData>,
|
|
): Promise<string> {
|
|
// The engine v2 path owns the whole run: it keeps no control-plane
|
|
// execution row, so everything below here does not apply to it.
|
|
if (this.engineV2Dispatcher.routesToEngineV2(data, existingExecution)) {
|
|
return await this.engineV2Dispatcher.start(data);
|
|
}
|
|
|
|
const establishContextError = await this.establishContextForPersistence(data);
|
|
|
|
if (!establishContextError) {
|
|
try {
|
|
await this.credentialsPermissionChecker.check(
|
|
data.workflowData.id,
|
|
data.workflowData.nodes,
|
|
data.userId,
|
|
);
|
|
} catch (error) {
|
|
const executionId = await this.activeExecutions.add(data, existingExecution);
|
|
await this.failExecution(data, executionId, error, responsePromise);
|
|
return executionId;
|
|
}
|
|
}
|
|
|
|
let executionWorkflow: Workflow | undefined;
|
|
if (!existingExecution && !establishContextError) {
|
|
try {
|
|
executionWorkflow = await this.prepareNewExecution(data, loadStaticData);
|
|
} catch (error) {
|
|
throw PreExecuteBlockedError.unwrap(error);
|
|
}
|
|
}
|
|
|
|
// Register a new execution
|
|
const executionId = await this.activeExecutions.add(data, existingExecution);
|
|
|
|
if (establishContextError) {
|
|
await this.failExecution(data, executionId, establishContextError, responsePromise);
|
|
return executionId;
|
|
}
|
|
|
|
const { id: workflowId } = data.workflowData;
|
|
|
|
if (responsePromise) {
|
|
this.activeExecutions.attachResponsePromise(executionId, responsePromise);
|
|
}
|
|
// The v1 main process owns the HTTP response. Start this before the
|
|
// queue/local decision because a worker cannot write to it.
|
|
const heartbeat = this.setupV1StreamingHeartbeat(data);
|
|
|
|
// @TODO: Reduce to true branch once feature is stable
|
|
const shouldEnqueue =
|
|
process.env.OFFLOAD_MANUAL_EXECUTIONS_TO_WORKERS === 'true'
|
|
? this.executionsConfig.mode === 'queue'
|
|
: this.executionsConfig.mode === 'queue' && data.executionMode !== 'manual';
|
|
|
|
const shouldReloadStaticData = Boolean(existingExecution && loadStaticData);
|
|
|
|
try {
|
|
if (shouldEnqueue) {
|
|
await this.enqueueExecution(
|
|
executionId,
|
|
workflowId,
|
|
data,
|
|
loadStaticData,
|
|
realtime,
|
|
existingExecution?.executionId,
|
|
);
|
|
} else {
|
|
await this.runMainProcess(
|
|
executionId,
|
|
data,
|
|
shouldReloadStaticData,
|
|
existingExecution,
|
|
executionWorkflow,
|
|
);
|
|
}
|
|
} catch (error) {
|
|
// A failed start means the post-execute promise that normally clears the
|
|
// heartbeat never settles, so clear it here.
|
|
heartbeat?.stop();
|
|
throw error;
|
|
}
|
|
|
|
// only run these when not in queue mode or when the execution is manual,
|
|
// since these calls are now done by the worker directly
|
|
if (
|
|
this.executionsConfig.mode !== 'queue' ||
|
|
this.instanceSettings.instanceType === 'worker' ||
|
|
data.executionMode === 'manual' ||
|
|
data.executionMode === 'chat'
|
|
) {
|
|
const postExecutePromise = this.activeExecutions.getPostExecutePromise(executionId);
|
|
postExecutePromise.catch((error) => {
|
|
if (error instanceof ExecutionCancelledError) return;
|
|
this.errorReporter.error(error, {
|
|
extra: { executionId, workflowId },
|
|
});
|
|
this.logger.error('There was an error in the post-execution promise', {
|
|
error,
|
|
executionId,
|
|
workflowId,
|
|
workflowName: data.workflowData.name,
|
|
...(data.projectId && { projectId: data.projectId }),
|
|
...(data.projectName && { projectName: data.projectName }),
|
|
});
|
|
});
|
|
}
|
|
|
|
// Clean up the streaming heartbeat when the execution finishes
|
|
if (heartbeat) {
|
|
const postExecutePromise = this.activeExecutions.getPostExecutePromise(executionId);
|
|
void postExecutePromise.finally(() => {
|
|
heartbeat.stop();
|
|
});
|
|
}
|
|
|
|
return executionId;
|
|
}
|
|
|
|
private setupV1StreamingHeartbeat(
|
|
data: IWorkflowExecutionDataProcess,
|
|
): StreamingWebhookResponseHeartbeat | undefined {
|
|
if (data.streamingEnabled !== true || !data.httpResponse) return undefined;
|
|
return new StreamingWebhookResponseHeartbeat(data.httpResponse);
|
|
}
|
|
|
|
private resolvePinData(data: IWorkflowExecutionDataProcess): IPinData | undefined {
|
|
if (['manual', 'evaluation'].includes(data.executionMode)) {
|
|
return data.pinData ?? data.workflowData.pinData;
|
|
}
|
|
return undefined;
|
|
}
|
|
|
|
async prepareNewExecution(
|
|
data: IWorkflowExecutionDataProcess,
|
|
loadStaticData?: boolean,
|
|
): Promise<Workflow | undefined> {
|
|
if (loadStaticData === true && data.workflowData.id) {
|
|
data.workflowData.staticData = await this.workflowStaticDataService.getStaticDataById(
|
|
data.workflowData.id,
|
|
);
|
|
}
|
|
|
|
return await this.workflowPreExecute.run(
|
|
data.workflowData,
|
|
data.executionMode,
|
|
data.source,
|
|
this.resolvePinData(data),
|
|
);
|
|
}
|
|
|
|
/** Run the workflow in current process */
|
|
|
|
private async runMainProcess(
|
|
executionId: string,
|
|
data: IWorkflowExecutionDataProcess,
|
|
loadStaticData?: boolean,
|
|
existingExecution?: ResumableExecution,
|
|
executionWorkflow?: Workflow,
|
|
): Promise<void> {
|
|
const workflowId = data.workflowData.id;
|
|
if (loadStaticData === true && workflowId) {
|
|
data.workflowData.staticData =
|
|
await this.workflowStaticDataService.getStaticDataById(workflowId);
|
|
}
|
|
|
|
// Soft timeout to stop workflow execution after current running node
|
|
// Changes were made by adding the `workflowTimeout` to the `additionalData`
|
|
// So that the timeout will also work for executions with nested workflows.
|
|
let executionTimeout: NodeJS.Timeout;
|
|
|
|
const workflowSettings = data.workflowData.settings ?? {};
|
|
let workflowTimeout = workflowSettings.executionTimeout ?? this.executionsConfig.timeout; // initialize with default
|
|
if (workflowTimeout > 0) {
|
|
workflowTimeout = Math.min(workflowTimeout, this.executionsConfig.maxTimeout);
|
|
}
|
|
|
|
const pinData = this.resolvePinData(data);
|
|
const workflow =
|
|
executionWorkflow ??
|
|
new Workflow({
|
|
id: workflowId,
|
|
name: data.workflowData.name,
|
|
nodes: data.workflowData.nodes,
|
|
connections: data.workflowData.connections,
|
|
active: data.workflowData.activeVersionId !== null,
|
|
nodeTypes: this.nodeTypes,
|
|
staticData: data.workflowData.staticData,
|
|
settings: workflowSettings,
|
|
pinData,
|
|
});
|
|
|
|
const additionalData = await WorkflowExecuteAdditionalData.getBase({
|
|
userId: data.userId,
|
|
workflowId: workflow.id,
|
|
executionTimeoutTimestamp:
|
|
workflowTimeout <= 0 ? undefined : Date.now() + workflowTimeout * 1000,
|
|
workflowSettings,
|
|
});
|
|
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 },
|
|
);
|
|
let workflowExecution: PCancelable<IRun>;
|
|
await this.executionRepository.setRunning(executionId); // write
|
|
|
|
try {
|
|
const lifecycleHooks = getLifecycleHooksForRegularMain(data, executionId);
|
|
additionalData.hooks = lifecycleHooks;
|
|
|
|
lifecycleHooks.addHandler('sendResponse', (response) => {
|
|
this.activeExecutions.resolveResponsePromise(executionId, response);
|
|
});
|
|
|
|
if (data.streamingEnabled) {
|
|
lifecycleHooks.addHandler('sendChunk', (chunk) => {
|
|
data.httpResponse?.write(JSON.stringify(chunk) + '\n');
|
|
if (data.httpResponse) flushResponse(data.httpResponse);
|
|
});
|
|
}
|
|
|
|
additionalData.setExecutionStatus = WorkflowExecuteAdditionalData.setExecutionStatus.bind({
|
|
executionId,
|
|
});
|
|
|
|
additionalData.sendDataToUI = WorkflowExecuteAdditionalData.sendDataToUI.bind({
|
|
pushRef: data.pushRef,
|
|
});
|
|
|
|
if (data.configureAdditionalData) {
|
|
await data.configureAdditionalData(additionalData);
|
|
}
|
|
|
|
if (data.executionData !== undefined) {
|
|
this.logger.debug(`Execution ID ${executionId} had Execution data. Running with payload.`, {
|
|
executionId,
|
|
});
|
|
const workflowExecute = new WorkflowExecute(
|
|
additionalData,
|
|
data.executionMode,
|
|
data.executionData,
|
|
storedAt,
|
|
);
|
|
workflowExecution = workflowExecute.processRunExecutionData(workflow);
|
|
} else {
|
|
workflowExecution = this.manualExecutionService.runManually(
|
|
data,
|
|
workflow,
|
|
additionalData,
|
|
executionId,
|
|
pinData,
|
|
);
|
|
}
|
|
|
|
this.activeExecutions.attachWorkflowExecution(executionId, workflowExecution);
|
|
|
|
if (workflowTimeout > 0) {
|
|
let timeout = Math.min(workflowTimeout, this.executionsConfig.maxTimeout) * 1000; // as milliseconds
|
|
if (data.startedAt && data.startedAt instanceof Date) {
|
|
// If startedAt is set, we calculate the timeout based on the startedAt time
|
|
// This is useful for executions that were waiting in a waiting state
|
|
// and we want to ensure the timeout is relative to when the execution started.
|
|
const now = Date.now();
|
|
timeout = Math.max(timeout - (now - data.startedAt.getTime()), 0);
|
|
}
|
|
if (timeout === 0) {
|
|
this.activeExecutions.stopExecution(
|
|
executionId,
|
|
new TimeoutExecutionCancelledError(executionId),
|
|
);
|
|
} else {
|
|
executionTimeout = setTimeout(() => {
|
|
void this.activeExecutions.stopExecution(
|
|
executionId,
|
|
new TimeoutExecutionCancelledError(executionId),
|
|
);
|
|
}, timeout);
|
|
}
|
|
}
|
|
|
|
workflowExecution
|
|
.then((fullRunData) => {
|
|
clearTimeout(executionTimeout);
|
|
if (workflowExecution.isCanceled) {
|
|
// oxlint-disable-next-line typescript/no-deprecated
|
|
fullRunData.finished = false;
|
|
}
|
|
|
|
this.activeExecutions.resolveExecutionResponsePromise(executionId);
|
|
this.activeExecutions.finalizeExecution(executionId, fullRunData);
|
|
})
|
|
.catch(
|
|
async (error) =>
|
|
await this.processError(
|
|
error,
|
|
new Date(),
|
|
data.executionMode,
|
|
executionId,
|
|
additionalData.hooks,
|
|
storedAt,
|
|
),
|
|
);
|
|
} catch (error) {
|
|
await this.processError(
|
|
error,
|
|
new Date(),
|
|
data.executionMode,
|
|
executionId,
|
|
additionalData.hooks,
|
|
storedAt,
|
|
);
|
|
|
|
throw error;
|
|
}
|
|
}
|
|
|
|
private async enqueueExecution(
|
|
executionId: string,
|
|
workflowId: string,
|
|
data: IWorkflowExecutionDataProcess,
|
|
loadStaticData?: boolean,
|
|
realtime?: boolean,
|
|
restartExecutionId?: string,
|
|
): Promise<void> {
|
|
if (!this.scalingService) {
|
|
const { ScalingService } = await import('@/scaling/scaling.service.js');
|
|
this.scalingService = Container.get(ScalingService);
|
|
await this.scalingService.setupQueue();
|
|
}
|
|
|
|
if (!this.poolConfigService) {
|
|
const { PoolConfigService } = await import('@/scaling/pool-config.service.ee.js');
|
|
this.poolConfigService = Container.get(PoolConfigService);
|
|
}
|
|
|
|
// TODO: For realtime jobs should probably also not do retry or not retry if they are older than x seconds.
|
|
// Check if they get retried by default and how often.
|
|
let job: Job;
|
|
let lifecycleHooks: ExecutionLifecycleHooks;
|
|
try {
|
|
const { queueName, poolName } = await this.poolConfigService.resolvePoolForExecution(data);
|
|
|
|
const jobData: JobData = {
|
|
workflowId,
|
|
executionId,
|
|
loadStaticData: !!loadStaticData,
|
|
pushRef: data.pushRef,
|
|
streamingEnabled: data.streamingEnabled,
|
|
callerAwaitsOutcome: data.callerAwaitsOutcome,
|
|
restartExecutionId,
|
|
projectId: data.projectId,
|
|
projectName: data.projectName,
|
|
// Carry the manual-execution identity for private credential resolution on the worker.
|
|
encryptedRunnerIdentity: data.encryptedRunnerIdentity,
|
|
poolName,
|
|
// MCP-specific fields for queue mode support
|
|
isMcpExecution: data.isMcpExecution,
|
|
mcpType: data.mcpType,
|
|
mcpSessionId: data.mcpSessionId,
|
|
mcpMessageId: data.mcpMessageId,
|
|
mcpToolCall: data.mcpToolCall,
|
|
mcpToolInput: data.mcpToolInput,
|
|
};
|
|
|
|
job = await this.scalingService.addJob(jobData, { priority: realtime ? 50 : 100, queueName });
|
|
|
|
lifecycleHooks = getLifecycleHooksForScalingMain(data, executionId);
|
|
|
|
// Normally also workflow should be supplied here but as it only used for sending
|
|
// data to editor-UI is not needed.
|
|
await lifecycleHooks.runHook('workflowExecuteBefore', [undefined, data.executionData]);
|
|
} catch (error) {
|
|
// We use "getLifecycleHooksForScalingWorker" as "getLifecycleHooksForScalingMain" does not contain the
|
|
// "workflowExecuteAfter" which we require.
|
|
const lifecycleHooks = getLifecycleHooksForScalingWorker(data, executionId);
|
|
await this.processError(error, new Date(), data.executionMode, executionId, lifecycleHooks);
|
|
// Nobody will wait for this job, so drop any outcome the worker already reported
|
|
this.scalingService.popJobResult(executionId);
|
|
throw error;
|
|
}
|
|
|
|
const workflowExecution: PCancelable<IRun> = new PCancelable(
|
|
async (resolve, reject, onCancel) => {
|
|
onCancel.shouldReject = false;
|
|
onCancel(async () => {
|
|
await this.scalingService.stopJob(job);
|
|
|
|
// We use "getLifecycleHooksForScalingWorker" as "getLifecycleHooksForScalingMain" does not contain the
|
|
// "workflowExecuteAfter" which we require.
|
|
const lifecycleHooks = getLifecycleHooksForScalingWorker(data, executionId);
|
|
const error = new ManualExecutionCancelledError(executionId);
|
|
await this.processError(
|
|
error,
|
|
new Date(),
|
|
data.executionMode,
|
|
executionId,
|
|
lifecycleHooks,
|
|
);
|
|
|
|
reject(error);
|
|
});
|
|
|
|
try {
|
|
await this.scalingService.waitForJob(job);
|
|
} catch (error) {
|
|
if (
|
|
error instanceof Error &&
|
|
typeof error.message === 'string' &&
|
|
error.message.includes('job stalled more than maxStalledCount')
|
|
) {
|
|
error = new MaxStalledCountError(error);
|
|
this.eventService.emit('job-stalled', {
|
|
executionId: job.data.executionId,
|
|
workflowId: job.data.workflowId,
|
|
hostId: this.instanceSettings.hostId,
|
|
jobId: job.id.toString(),
|
|
});
|
|
}
|
|
|
|
// We use "getLifecycleHooksForScalingWorker" as "getLifecycleHooksForScalingMain" does not contain the
|
|
// "workflowExecuteAfter" which we require.
|
|
const lifecycleHooks = getLifecycleHooksForScalingWorker(data, executionId);
|
|
|
|
await this.processError(
|
|
error,
|
|
new Date(),
|
|
data.executionMode,
|
|
executionId,
|
|
lifecycleHooks,
|
|
);
|
|
|
|
this.scalingService.popJobResult(executionId);
|
|
|
|
return reject(error);
|
|
}
|
|
|
|
const jobResult = this.scalingService.popJobResult(executionId);
|
|
|
|
let runData: IRun;
|
|
|
|
if (
|
|
!jobResult ||
|
|
this.needsFullExecutionData(data.executionMode, executionId, data.forceFullExecutionData)
|
|
) {
|
|
let fullExecutionData: IExecutionResponse | undefined;
|
|
try {
|
|
fullExecutionData = await this.executionPersistence.findSingleExecution(executionId, {
|
|
includeData: true,
|
|
unflattenData: true,
|
|
});
|
|
} catch (error) {
|
|
// An async executor's throw would never settle this promise, and the
|
|
// active execution would keep the request alive until restart
|
|
await this.processError(
|
|
error,
|
|
new Date(),
|
|
data.executionMode,
|
|
executionId,
|
|
getLifecycleHooksForScalingWorker(data, executionId),
|
|
);
|
|
return reject(error);
|
|
}
|
|
|
|
if (!fullExecutionData) {
|
|
// Not a bug by itself: the worker deletes an execution that is not saved.
|
|
// Finalizing with a failed run makes the webhook respond with an error
|
|
// instead of a success without data.
|
|
this.logger.warn(`Execution ${executionId} ended but its record is gone`, {
|
|
executionId,
|
|
});
|
|
const error = new WorkflowOperationError(
|
|
`Could not find execution with id "${executionId}"`,
|
|
);
|
|
await this.processError(
|
|
error,
|
|
new Date(),
|
|
data.executionMode,
|
|
executionId,
|
|
getLifecycleHooksForScalingWorker(data, executionId),
|
|
);
|
|
return reject(error);
|
|
}
|
|
|
|
runData = {
|
|
// oxlint-disable-next-line typescript/no-deprecated
|
|
finished: fullExecutionData.finished,
|
|
mode: fullExecutionData.mode,
|
|
startedAt: fullExecutionData.startedAt,
|
|
stoppedAt: fullExecutionData.stoppedAt,
|
|
status: fullExecutionData.status,
|
|
waitTill: fullExecutionData.waitTill,
|
|
data: fullExecutionData.data,
|
|
jobId: job.id.toString(),
|
|
storedAt: fullExecutionData.storedAt,
|
|
};
|
|
} else {
|
|
runData = {
|
|
finished: jobResult.success,
|
|
mode: data.executionMode,
|
|
startedAt: jobResult.startedAt,
|
|
stoppedAt: jobResult.stoppedAt,
|
|
status: jobResult.status,
|
|
waitTill: jobResult.waitTill,
|
|
data: createRunExecutionData({
|
|
resultData: {
|
|
runData: {},
|
|
lastNodeExecuted: jobResult.lastNodeExecuted,
|
|
error: jobResult.error,
|
|
metadata: jobResult.metadata,
|
|
},
|
|
}),
|
|
jobId: job.id.toString(),
|
|
storedAt: this.storageConfig.modeTag,
|
|
};
|
|
}
|
|
|
|
this.activeExecutions.finalizeExecution(executionId, runData);
|
|
|
|
// Normally also static data should be supplied here but as it only used for sending
|
|
// data to editor-UI is not needed.
|
|
await lifecycleHooks.runHook('workflowExecuteAfter', [runData]);
|
|
|
|
resolve(runData);
|
|
},
|
|
);
|
|
|
|
workflowExecution.catch(() => {
|
|
// We `reject` this promise if the execution fails
|
|
// but the error is handled already by processError
|
|
// So we're just preventing crashes here.
|
|
});
|
|
|
|
this.activeExecutions.attachWorkflowExecution(executionId, workflowExecution, {
|
|
isQueueJob: true,
|
|
});
|
|
}
|
|
|
|
/**
|
|
* Whether main must retrieve full execution data from the DB on job completion.
|
|
*
|
|
* Full data is needed when:
|
|
* - `integrated` mode: parent workflow needs child execution output data
|
|
* - `lastNode` response mode: webhook response is built from the last node's output
|
|
* - `workflow.postExecute` hook: external hooks receive full execution data
|
|
*
|
|
* In all other cases we can skip the DB fetch and use the lightweight
|
|
* result summary sent by the worker via the job progress message.
|
|
*/
|
|
private needsFullExecutionData(
|
|
executionMode: WorkflowExecuteMode,
|
|
executionId: string,
|
|
forceFullExecutionData?: boolean,
|
|
): boolean {
|
|
if (forceFullExecutionData) return true;
|
|
|
|
return (
|
|
executionMode === 'integrated' ||
|
|
this.activeExecutions.getResponseMode(executionId) === 'lastNode' ||
|
|
this.externalHooks.hasHook('workflow.postExecute')
|
|
);
|
|
}
|
|
}
|