From a73df56745dc8f0be23dc6bb89e39f7eecef3b2a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Mike=20Repe=C4=87?= Date: Mon, 5 Oct 2026 12:02:08 +0200 Subject: [PATCH] fix(core): Do not suspend executions that still owe a webhook response at worker shutdown --- .../config/src/configs/scaling-mode.config.ts | 3 +- .../cli/src/__tests__/workflow-runner.test.ts | 2 + .../__tests__/job-processor.service.test.ts | 71 +++++++++++++++++++ packages/cli/src/scaling/job-processor.ts | 18 +++-- packages/cli/src/scaling/scaling.types.ts | 9 ++- .../__tests__/webhook-helpers.test.ts | 7 ++ packages/cli/src/webhooks/webhook-helpers.ts | 1 + packages/cli/src/workflow-runner.ts | 1 + packages/workflow/src/interfaces.ts | 6 ++ 9 files changed, 111 insertions(+), 7 deletions(-) diff --git a/packages/@n8n/config/src/configs/scaling-mode.config.ts b/packages/@n8n/config/src/configs/scaling-mode.config.ts index 6397b3756f0..8bcf968117c 100644 --- a/packages/@n8n/config/src/configs/scaling-mode.config.ts +++ b/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; diff --git a/packages/cli/src/__tests__/workflow-runner.test.ts b/packages/cli/src/__tests__/workflow-runner.test.ts index 4d590935660..892a6b090bc 100644 --- a/packages/cli/src/__tests__/workflow-runner.test.ts +++ b/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), ); diff --git a/packages/cli/src/scaling/__tests__/job-processor.service.test.ts b/packages/cli/src/scaling/__tests__/job-processor.service.test.ts index 6f21446afdc..a087145ae83 100644 --- a/packages/cli/src/scaling/__tests__/job-processor.service.test.ts +++ b/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) => { + const executionPersistence = mock(); + executionPersistence.findSingleExecution.mockResolvedValue( + mock({ + mode: 'webhook', + workflowData: { nodes: [], staticData: {} }, + data: mock(), + }), + ); + const additionalData = mock(); + vi.spyOn(WorkflowExecuteAdditionalData, 'getBase').mockResolvedValue(additionalData); + const jobProcessor = createJobProcessor(executionPersistence); + + let resolveRun!: (run: IRun) => void; + processRunExecutionDataMock.mockReturnValue(new Promise((r) => (resolveRun = r))); + + const job = mock({ + 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.findSingleExecution.mockResolvedValue( diff --git a/packages/cli/src/scaling/job-processor.ts b/packages/cli/src/scaling/job-processor.ts index 9871e3fb916..9fa140a5777 100644 --- a/packages/cli/src/scaling/job-processor.ts +++ b/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 => { // 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 => { @@ -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); } diff --git a/packages/cli/src/scaling/scaling.types.ts b/packages/cli/src/scaling/scaling.types.ts index 94704937dcc..0faba2c3b4f 100644 --- a/packages/cli/src/scaling/scaling.types.ts +++ b/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; - /** 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 = { diff --git a/packages/cli/src/webhooks/__tests__/webhook-helpers.test.ts b/packages/cli/src/webhooks/__tests__/webhook-helpers.test.ts index 4750856d37f..cafa2cd8ba8 100644 --- a/packages/cli/src/webhooks/__tests__/webhook-helpers.test.ts +++ b/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', diff --git a/packages/cli/src/webhooks/webhook-helpers.ts b/packages/cli/src/webhooks/webhook-helpers.ts index d3f8a6efe5a..c24f44e09d7 100644 --- a/packages/cli/src/webhooks/webhook-helpers.ts +++ b/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. diff --git a/packages/cli/src/workflow-runner.ts b/packages/cli/src/workflow-runner.ts index 709009dcd85..a572ae73cee 100644 --- a/packages/cli/src/workflow-runner.ts +++ b/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, diff --git a/packages/workflow/src/interfaces.ts b/packages/workflow/src/interfaces.ts index fc1354699ec..d5697440d69 100644 --- a/packages/workflow/src/interfaces.ts +++ b/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