fix(core): Do not suspend executions that still owe a webhook response at worker shutdown

packages/@n8n/config/src/configs/scaling-mode.config.ts

@@ -187,7 +187,8 @@ export class ScalingModeConfig {
 	 * Whether a worker that receives a shutdown signal suspends in-flight
 	 * production executions at the next node boundary. Suspended executions are
 	 * persisted as 'waiting' and resumed on another worker, instead of running
-	 * to completion (or force-kill) on the exiting worker.
+	 * to completion (or force-kill) on the exiting worker. An execution that
+	 * still owes its webhook caller a response is not suspended.
 	 */
 	@Env('N8N_WORKER_SUSPEND_EXECUTIONS_ON_SHUTDOWN')
 	suspendExecutionsOnShutdown: boolean = false;

packages/cli/src/__tests__/workflow-runner.test.ts

@@ -1193,6 +1193,7 @@ describe('enqueueExecution', () => {
 			executionData: undefined,
 			pushRef: 'push-ref',
 			streamingEnabled: true,
+			webhookResponsePending: true,
 		});
 		const error = new Error('stop for test purposes');
 
@@ -1212,6 +1213,7 @@ describe('enqueueExecution', () => {
 				workflowId: 'workflow-xyz',
 				executionId: '1',
 				restartExecutionId,
+				webhookResponsePending: true,
 			}),
 			expect.any(Object),
 		);

packages/cli/src/scaling/__tests__/job-processor.service.test.ts

@@ -310,6 +310,7 @@ describe('JobProcessor', () => {
 					loadStaticData: false,
 					streamingEnabled: false,
 					isMcpExecution: false,
+					webhookResponsePending: false,
 				},
 			});
 
@@ -348,6 +349,7 @@ describe('JobProcessor', () => {
 					loadStaticData: false,
 					streamingEnabled: false,
 					isMcpExecution: false,
+					webhookResponsePending: false,
 				},
 			});
 
@@ -371,6 +373,75 @@ describe('JobProcessor', () => {
 			await processPromise;
 		});
 
+		const startWebhookJob = async (jobData: Partial<Job['data']>) => {
+			const executionPersistence = mock<ExecutionPersistence>();
+			executionPersistence.findSingleExecution.mockResolvedValue(
+				mock<IExecutionResponse>({
+					mode: 'webhook',
+					workflowData: { nodes: [], staticData: {} },
+					data: mock<IRunExecutionData>(),
+				}),
+			);
+			const additionalData = mock<IWorkflowExecuteAdditionalData>();
+			vi.spyOn(WorkflowExecuteAdditionalData, 'getBase').mockResolvedValue(additionalData);
+			const jobProcessor = createJobProcessor(executionPersistence);
+
+			let resolveRun!: (run: IRun) => void;
+			processRunExecutionDataMock.mockReturnValue(new Promise<IRun>((r) => (resolveRun = r)));
+
+			const job = mock<Job>({
+				id: 'job-1',
+				data: {
+					executionId: 'exec-1',
+					loadStaticData: false,
+					streamingEnabled: false,
+					isMcpExecution: false,
+					...jobData,
+				},
+			});
+			const processPromise = jobProcessor.processJob(job);
+			await vi.waitFor(() => expect(jobProcessor.getRunningJobIds()).toEqual(['job-1']));
+
+			const finish = async () => {
+				resolveRun(successRun());
+				await processPromise;
+			};
+			return { jobProcessor, hooks: additionalData.hooks!, finish };
+		};
+
+		it('should not suspend a webhook job while its response is still pending', async () => {
+			const { jobProcessor, finish } = await startWebhookJob({ webhookResponsePending: true });
+
+			jobProcessor.suspendRunningJobs();
+			expect(workflowExecuteSuspendMock).not.toHaveBeenCalled();
+			expect(logger.info).not.toHaveBeenCalledWith(expect.stringContaining('Requested suspension'));
+
+			await finish();
+		});
+
+		it('should treat a webhook job without the flag as owing a response', async () => {
+			const { jobProcessor, finish } = await startWebhookJob({ webhookResponsePending: undefined });
+
+			jobProcessor.suspendRunningJobs();
+			expect(workflowExecuteSuspendMock).not.toHaveBeenCalled();
+
+			await finish();
+		});
+
+		it('should suspend a webhook job once it has relayed its response', async () => {
+			const { jobProcessor, hooks, finish } = await startWebhookJob({
+				webhookResponsePending: true,
+			});
+
+			jobProcessor.suspendRunningJobs();
+			expect(workflowExecuteSuspendMock).not.toHaveBeenCalled();
+
+			await hooks.runHook('sendResponse', [{ body: {}, headers: {}, statusCode: 200 }]);
+			expect(workflowExecuteSuspendMock).toHaveBeenCalledTimes(1);
+
+			await finish();
+		});
+
 		it('should not attach a suspend handle to a non-suspendable job', async () => {
 			const executionPersistence = mock<ExecutionPersistence>();
 			executionPersistence.findSingleExecution.mockResolvedValue(

packages/cli/src/scaling/job-processor.ts

@@ -235,6 +235,9 @@ export class JobProcessor {
 			});
 		}
 
+		// An older main sends no flag, so a webhook run is assumed to owe a response.
+		let webhookResponsePending = job.data.webhookResponsePending ?? execution.mode === 'webhook';
+
 		lifecycleHooks.addHandler('sendResponse', async (response): Promise<void> => {
 			// An MCP Service call takes its result from the execution's stored data, so a
 			// response relayed to main has no reader. Relaying one would also reach main
@@ -273,6 +276,9 @@ export class JobProcessor {
 			};
 
 			await job.progress(msg);
+
+			webhookResponsePending = false;
+			if (this.suspensionRequested) this.runningJobs[job.id]?.suspend?.();
 		});
 
 		lifecycleHooks.addHandler('sendChunk', async (chunk: StructuredChunk): Promise<void> => {
@@ -368,7 +374,11 @@ export class JobProcessor {
 
 		if (workflowExecute && this.isJobSuspendable(job, execution)) {
 			const suspendable = workflowExecute;
-			runningJob.suspend = () => suspendable.suspend();
+			runningJob.suspend = () => {
+				if (webhookResponsePending) return false;
+				suspendable.suspend();
+				return true;
+			};
 			// A job that was still in its preflight reads when shutdown asked the
 			// running jobs to suspend would otherwise run to completion unasked.
 			if (this.suspensionRequested) runningJob.suspend();
@@ -590,7 +600,8 @@ export class JobProcessor {
 	 * Whether a job may be suspended at worker shutdown. Only production
 	 * executions qualify: manual and evaluation runs are tied to a session,
 	 * streaming responses cannot migrate mid-stream, and MCP executions are
-	 * pinned to their session.
+	 * pinned to their session. A pending webhook response is checked at
+	 * suspend time, since the run may still send it.
 	 */
 	/** Set once at shutdown; jobs that register afterwards suspend right away. */
 	private suspensionRequested = false;
@@ -612,8 +623,7 @@ export class JobProcessor {
 		const executionIds: string[] = [];
 
 		for (const runningJob of Object.values(this.runningJobs)) {
-			if (!runningJob.suspend) continue;
-			runningJob.suspend();
+			if (!runningJob.suspend?.()) continue;
 			executionIds.push(runningJob.executionId);
 		}
 

packages/cli/src/scaling/scaling.types.ts

@@ -22,6 +22,8 @@ export type JobData = {
 	loadStaticData: boolean;
 	pushRef?: string;
 	streamingEnabled?: boolean;
+	/** True while the webhook caller still waits for a response the run produces. */
+	webhookResponsePending?: boolean;
 	restartExecutionId?: string;
 	projectId?: string;
 	projectName?: string;
@@ -147,8 +149,11 @@ export type AbortJobMessage = {
 
 export type RunningJob = RunningJobSummary & {
 	run: PCancelable<IRun>;
-	/** Present only when the job may be suspended at worker shutdown. */
-	suspend?: () => void;
+	/**
+	 * Present only when the job may be suspended at worker shutdown. Returns
+	 * false while the run still owes its webhook caller a response.
+	 */
+	suspend?: () => boolean;
 };
 
 export type QueueRecoveryContext = {

packages/cli/src/webhooks/__tests__/webhook-helpers.test.ts

@@ -1747,6 +1747,13 @@ describe('executeWebhook credential-status gate', () => {
 		expect(workflowRunner.run).toHaveBeenCalled();
 	});
 
+	it('marks an onReceived run as owing no webhook response', async () => {
+		await runGate({ authentication: 'none', gateResult: missingGateResult });
+
+		const [runData] = vi.mocked(workflowRunner.run).mock.calls[0];
+		expect(runData.webhookResponsePending).toBe(false);
+	});
+
 	it('does not gate webhooks that do not establish a triggering identity', async () => {
 		const { checkCredentialStatus, responseCallback } = await runGate({
 			authentication: 'none',

packages/cli/src/webhooks/webhook-helpers.ts

@@ -1184,6 +1184,7 @@ export async function executeWebhook(
 			projectName: project?.name,
 			userId: webhookData.userId,
 			encryptedRunnerIdentity: additionalData.encryptedRunnerIdentity,
+			webhookResponsePending: responseMode !== 'onReceived',
 			// v1 reads this from `executionData.startData`, which `prepareExecutionData`
 			// sets, so carrying it here changes nothing for v1. Engine v2 has no way to
 			// stop at a node, and its dispatcher refuses the run on this field.

packages/cli/src/workflow-runner.ts

@@ -763,6 +763,7 @@ export class WorkflowRunner {
 				loadStaticData: !!loadStaticData,
 				pushRef: data.pushRef,
 				streamingEnabled: data.streamingEnabled,
+				webhookResponsePending: data.webhookResponsePending,
 				restartExecutionId,
 				projectId: data.projectId,
 				projectName: data.projectName,

packages/workflow/src/interfaces.ts

@@ -3789,6 +3789,12 @@ export interface IWorkflowExecutionDataProcess {
 	agentRequest?: AiAgentRequest;
 	httpResponse?: express.Response; // Used for streaming responses
 	streamingEnabled?: boolean;
+	/**
+	 * True while the webhook caller still waits for a response that the run
+	 * itself produces (every response mode except `onReceived`). A worker must
+	 * not park such a run at shutdown before that response has gone out.
+	 */
+	webhookResponsePending?: boolean;
 	/**
 	 * Only engine v2 reads this. The data-plane execution id, set by a caller that
 	 * minted it before the run starts: to subscribe to the run's answer, or because