mirror of
https://github.com/n8n-io/n8n.git
synced 2026-10-29 19:10:43 +08:00
fix(core): Respond with an error when a webhook execution ends before the Respond to Webhook node runs (#37006)
Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
co-authored by
Claude Opus 5
parent
7785b08a97
commit
f9814b753f
@@ -23,6 +23,7 @@ import type { Mock } from 'vitest';
|
||||
import { captor, mock } from 'vitest-mock-extended';
|
||||
|
||||
import { ActiveExecutions } from '@/active-executions';
|
||||
import { EXECUTION_ENDED_WITHOUT_RESPONSE } from '@/webhooks/constants';
|
||||
import { ConcurrencyControlService } from '@/concurrency/concurrency-control.service';
|
||||
import type { EventService } from '@/events/event.service';
|
||||
import type { ExecutionPersistence } from '@/executions/execution-persistence';
|
||||
@@ -368,6 +369,37 @@ describe('ActiveExecutions', () => {
|
||||
expect(resumedExecution.responsePromise).toBe(responsePromise);
|
||||
});
|
||||
|
||||
describe('resolveExecutionResponsePromise', () => {
|
||||
test('Should settle the response promise with the no-response sentinel', async () => {
|
||||
const executionId = await activeExecutions.add(executionData);
|
||||
activeExecutions.attachResponsePromise(executionId, responsePromise);
|
||||
|
||||
activeExecutions.resolveExecutionResponsePromise(executionId);
|
||||
|
||||
// Identity, not equality: `webhook-helpers` recognises the sentinel by reference,
|
||||
// so any other empty object would be treated as a real, empty response.
|
||||
expect(vi.mocked(responsePromise.resolve).mock.calls[0][0]).toBe(
|
||||
EXECUTION_ENDED_WITHOUT_RESPONSE,
|
||||
);
|
||||
});
|
||||
|
||||
test('Should leave a waiting execution alone, so it can still respond on resume', async () => {
|
||||
const executionId = await activeExecutions.add(executionData);
|
||||
activeExecutions.attachResponsePromise(executionId, responsePromise);
|
||||
activeExecutions.setStatus(executionId, 'waiting');
|
||||
|
||||
activeExecutions.resolveExecutionResponsePromise(executionId);
|
||||
|
||||
expect(responsePromise.resolve).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
test('Should do nothing for an unknown execution', () => {
|
||||
expect(() =>
|
||||
activeExecutions.resolveExecutionResponsePromise(FAKE_EXECUTION_ID),
|
||||
).not.toThrow();
|
||||
});
|
||||
});
|
||||
|
||||
describe('finalizeExecution', () => {
|
||||
test('Should not remove a waiting execution', async () => {
|
||||
const executionId = await activeExecutions.add(executionData);
|
||||
|
||||
@@ -26,6 +26,7 @@ import type {
|
||||
ResumableExecution,
|
||||
} from '@/interfaces';
|
||||
import { isWorkflowIdValid } from '@/utils';
|
||||
import { EXECUTION_ENDED_WITHOUT_RESPONSE } from '@/webhooks/constants';
|
||||
|
||||
import { ConcurrencyCapacityReservation } from './concurrency/concurrency-capacity-reservation';
|
||||
import { ConcurrencyControlService } from './concurrency/concurrency-control.service';
|
||||
@@ -275,7 +276,7 @@ export class ActiveExecutions {
|
||||
const execution = this.getExecutionOrFail(executionId);
|
||||
|
||||
if (execution.status !== 'waiting' && execution?.responsePromise) {
|
||||
execution.responsePromise.resolve({});
|
||||
execution.responsePromise.resolve(EXECUTION_ENDED_WITHOUT_RESPONSE);
|
||||
this.logger.debug('Execution response promise cleaned', { executionId });
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1,4 +1,5 @@
|
||||
import { Logger } from '@n8n/backend-common';
|
||||
import { Container } from '@n8n/di';
|
||||
import { mockInstance } from '@n8n/backend-test-utils';
|
||||
import type express from 'express';
|
||||
import {
|
||||
@@ -28,6 +29,8 @@ import type {
|
||||
IWebhookData,
|
||||
IWorkflowExecuteAdditionalData,
|
||||
CredentialCheckResult,
|
||||
IRun,
|
||||
IExecuteResponsePromiseData,
|
||||
} from 'n8n-workflow';
|
||||
import {
|
||||
FORM_NODE_TYPE,
|
||||
@@ -54,6 +57,7 @@ import {
|
||||
executeWebhook,
|
||||
_privateGetWebhookErrorMessage,
|
||||
} from '../webhook-helpers';
|
||||
import { EXECUTION_ENDED_WITHOUT_RESPONSE } from '../constants';
|
||||
import type { IWebhookResponseCallbackData, WebhookRequest } from '../webhook.types';
|
||||
import type { Project } from '@n8n/db';
|
||||
import { ActiveExecutions } from '@/active-executions';
|
||||
@@ -525,6 +529,26 @@ describe('setupResponseNodePromise', () => {
|
||||
);
|
||||
expect(responseCallback).toHaveBeenCalledWith(error, {});
|
||||
});
|
||||
|
||||
// When an execution ends without the Respond to Webhook node having run,
|
||||
// `ActiveExecutions.resolveExecutionResponsePromise` settles this promise with a
|
||||
// sentinel. The post-execute handler answers in that case, so this one must not.
|
||||
test('should not respond when the execution ended without a response', async () => {
|
||||
setupResponseNodePromise(
|
||||
responsePromise,
|
||||
res,
|
||||
responseCallback,
|
||||
workflowStartNode,
|
||||
executionId,
|
||||
workflow,
|
||||
);
|
||||
|
||||
responsePromise.resolve(EXECUTION_ENDED_WITHOUT_RESPONSE as IN8nHttpFullResponse);
|
||||
await new Promise(process.nextTick);
|
||||
|
||||
expect(errorReporter.error).not.toHaveBeenCalled();
|
||||
expect(responseCallback).not.toHaveBeenCalled();
|
||||
});
|
||||
});
|
||||
|
||||
describe('handleHostedChatResponse', () => {
|
||||
@@ -1432,3 +1456,207 @@ describe('executeWebhook establishTriggerIdentity', () => {
|
||||
);
|
||||
});
|
||||
});
|
||||
|
||||
// Reproduction for CAT-4050 / GitHub issue #36175: in `responseNode` mode, when
|
||||
// a node fails before the Respond to Webhook node has run, the execution never
|
||||
// sends a response. These tests pin down what the HTTP caller receives instead.
|
||||
describe('executeWebhook in responseNode mode when the Respond node never runs', () => {
|
||||
// `mockInstance(Logger)` above already replaced the container binding.
|
||||
const logger = Container.get(Logger);
|
||||
/** An execution that failed at a node, as the reported agent branch does. */
|
||||
const erroredRun = mock<IRun>({
|
||||
mode: 'webhook',
|
||||
finished: false,
|
||||
status: 'error',
|
||||
data: {
|
||||
resultData: {
|
||||
error: new NodeOperationError(mock<INode>({ name: 'Agent' }), 'Model call failed'),
|
||||
runData: {},
|
||||
lastNodeExecuted: 'Agent',
|
||||
},
|
||||
},
|
||||
});
|
||||
|
||||
/**
|
||||
* Drives `executeWebhook` up to the point where the workflow is running, then
|
||||
* hands back the deferred response promise that `executeWebhook` passed to
|
||||
* `WorkflowRunner`, plus the post-execute deferred and the response callback.
|
||||
* The caller decides in which order the two settle.
|
||||
*/
|
||||
const startWebhook = async () => {
|
||||
vi.spyOn(WorkflowExecuteAdditionalData, 'getBase').mockResolvedValue(
|
||||
mock<IWorkflowExecuteAdditionalData>(),
|
||||
);
|
||||
ownershipService.getWorkflowProjectCached.mockResolvedValue(
|
||||
mock<Project>({ id: 'project-1', name: 'Project 1' }),
|
||||
);
|
||||
webhookService.runWebhook.mockResolvedValue({ workflowData: [[{ json: {} }]] });
|
||||
workflowRunner.run.mockResolvedValue(EXECUTION_ID);
|
||||
|
||||
const postExecute = createDeferredPromise<IRun | undefined>();
|
||||
activeExecutions.getPostExecutePromise.mockReturnValue(postExecute.promise);
|
||||
|
||||
const workflow = mock<Workflow>({
|
||||
id: WORKFLOW_ID,
|
||||
name: 'Test Workflow',
|
||||
nodeTypes: {
|
||||
getByNameAndVersion: vi
|
||||
.fn()
|
||||
.mockReturnValue(mock<INodeType>({ description: { name: 'webhook' } })),
|
||||
},
|
||||
expression: {
|
||||
// Return the webhook description value, so `responseMode` below applies.
|
||||
getSimpleParameterValue: vi.fn(
|
||||
(...args: Parameters<Workflow['expression']['getSimpleParameterValue']>) =>
|
||||
args[1] /* paramValue */ ?? args[5] /* defaultValue */,
|
||||
),
|
||||
getComplexParameterValue: vi.fn(
|
||||
(...args: Parameters<Workflow['expression']['getComplexParameterValue']>) =>
|
||||
args[1] /* paramValue */,
|
||||
),
|
||||
},
|
||||
});
|
||||
|
||||
const webhookData = {
|
||||
webhookDescription: { name: 'default', responseMode: 'responseNode' },
|
||||
workflowId: WORKFLOW_ID,
|
||||
} as unknown as IWebhookData;
|
||||
|
||||
const responseCallback = vi.fn();
|
||||
|
||||
await executeWebhook(
|
||||
workflow,
|
||||
webhookData,
|
||||
mock<IWorkflowBase>({ id: WORKFLOW_ID, name: 'Test Workflow' }),
|
||||
mock<INode>({ name: 'Webhook', type: WEBHOOK_NODE_TYPE, typeVersion: 2, parameters: {} }),
|
||||
'webhook',
|
||||
undefined,
|
||||
undefined,
|
||||
undefined,
|
||||
mock<WebhookRequest>({ method: 'POST', contentType: undefined, headers: {} }),
|
||||
mock<express.Response>({ headersSent: false }),
|
||||
responseCallback,
|
||||
);
|
||||
|
||||
// Argument 5 of `WorkflowRunner.run` is the deferred response promise.
|
||||
const responsePromise = workflowRunner.run.mock
|
||||
.calls[0][4] as IDeferredPromise<IExecuteResponsePromiseData>;
|
||||
|
||||
return { responsePromise, postExecute, responseCallback };
|
||||
};
|
||||
|
||||
beforeEach(() => {
|
||||
vi.restoreAllMocks();
|
||||
vi.clearAllMocks();
|
||||
});
|
||||
|
||||
// `WorkflowRunner` calls `resolveExecutionResponsePromise` before
|
||||
// `finalizeExecution`, so the sentinel usually settles first. Either ordering must
|
||||
// produce the same answer.
|
||||
it('responds 500 when the response promise settles first', async () => {
|
||||
const { responsePromise, postExecute, responseCallback } = await startWebhook();
|
||||
|
||||
responsePromise.resolve(EXECUTION_ENDED_WITHOUT_RESPONSE);
|
||||
postExecute.resolve(erroredRun);
|
||||
await new Promise(process.nextTick);
|
||||
|
||||
expect(responseCallback.mock.calls[0]).toEqual([
|
||||
null,
|
||||
{ data: { message: 'Error in workflow' }, responseCode: 500 },
|
||||
]);
|
||||
expect(logger.warn).toHaveBeenCalledWith(
|
||||
'Webhook execution failed before a response was sent',
|
||||
{
|
||||
executionId: EXECUTION_ID,
|
||||
workflowId: WORKFLOW_ID,
|
||||
responseMode: 'responseNode',
|
||||
lastNodeExecuted: 'Agent',
|
||||
},
|
||||
);
|
||||
});
|
||||
|
||||
it('responds 500 when the post-execute promise settles first', async () => {
|
||||
const { responsePromise, postExecute, responseCallback } = await startWebhook();
|
||||
|
||||
postExecute.resolve(erroredRun);
|
||||
await new Promise(process.nextTick);
|
||||
responsePromise.resolve(EXECUTION_ENDED_WITHOUT_RESPONSE);
|
||||
await new Promise(process.nextTick);
|
||||
|
||||
expect(responseCallback.mock.calls[0]).toEqual([
|
||||
null,
|
||||
{ data: { message: 'Error in workflow' }, responseCode: 500 },
|
||||
]);
|
||||
});
|
||||
|
||||
it('responds only once when the Respond to Webhook node answered before the failure', async () => {
|
||||
const { responsePromise, postExecute, responseCallback } = await startWebhook();
|
||||
|
||||
responsePromise.resolve({ body: { ok: true }, headers: {}, statusCode: 200 });
|
||||
await new Promise(process.nextTick);
|
||||
postExecute.resolve(erroredRun);
|
||||
await new Promise(process.nextTick);
|
||||
|
||||
expect(responseCallback).toHaveBeenCalledTimes(1);
|
||||
expect(responseCallback.mock.calls[0]).toEqual([
|
||||
null,
|
||||
{ data: { ok: true }, headers: {}, responseCode: 200 },
|
||||
]);
|
||||
});
|
||||
|
||||
it('responds with an empty body when a successful execution never reached the node', async () => {
|
||||
const { responsePromise, postExecute, responseCallback } = await startWebhook();
|
||||
|
||||
const successfulRun = mock<IRun>({
|
||||
mode: 'webhook',
|
||||
finished: true,
|
||||
status: 'success',
|
||||
data: { resultData: { error: undefined, runData: {}, lastNodeExecuted: 'Agent' } },
|
||||
});
|
||||
|
||||
responsePromise.resolve(EXECUTION_ENDED_WITHOUT_RESPONSE);
|
||||
postExecute.resolve(successfulRun);
|
||||
await new Promise(process.nextTick);
|
||||
|
||||
expect(responseCallback.mock.calls[0]).toEqual([null, { data: undefined, responseCode: 200 }]);
|
||||
});
|
||||
|
||||
it('does not answer again while an offloaded binary response is still streaming', async () => {
|
||||
// The stream never arrives, so the binary branch has not answered yet when the
|
||||
// execution fails. The post-execute handler must not answer in its place.
|
||||
vi.mocked(Container.get(BinaryDataService).getAsStream).mockReturnValue(
|
||||
new Promise<Readable>(() => {}),
|
||||
);
|
||||
|
||||
const { responsePromise, postExecute, responseCallback } = await startWebhook();
|
||||
|
||||
responsePromise.resolve({
|
||||
body: { binaryData: { id: 'binary-1' } },
|
||||
headers: {},
|
||||
statusCode: 200,
|
||||
});
|
||||
await new Promise(process.nextTick);
|
||||
postExecute.resolve(erroredRun);
|
||||
await new Promise(process.nextTick);
|
||||
|
||||
expect(responseCallback).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('does not answer twice when the node responded and the execution then succeeded', async () => {
|
||||
const { responsePromise, postExecute, responseCallback } = await startWebhook();
|
||||
|
||||
const successfulRun = mock<IRun>({
|
||||
mode: 'webhook',
|
||||
finished: true,
|
||||
status: 'success',
|
||||
data: { resultData: { error: undefined, runData: {}, lastNodeExecuted: 'Agent' } },
|
||||
});
|
||||
|
||||
responsePromise.resolve({ body: { ok: true }, headers: {}, statusCode: 200 });
|
||||
await new Promise(process.nextTick);
|
||||
postExecute.resolve(successfulRun);
|
||||
await new Promise(process.nextTick);
|
||||
|
||||
expect(responseCallback).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
});
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
import type { IExecuteResponsePromiseData } from 'n8n-workflow';
|
||||
import { CHAT_TRIGGER_NODE_TYPE, FORM_NODE_TYPE, FORM_TRIGGER_NODE_TYPE } from 'n8n-workflow';
|
||||
|
||||
export const WEBHOOK_CONFLICT_MESSAGE = 'There is a conflict with one of the webhooks.';
|
||||
@@ -7,3 +8,11 @@ export const authAllowlistedNodes = new Set([
|
||||
FORM_TRIGGER_NODE_TYPE,
|
||||
FORM_NODE_TYPE,
|
||||
]);
|
||||
|
||||
/**
|
||||
* Resolved into an execution's response promise when the execution ends without
|
||||
* the Respond to Webhook node having produced a response. Consumers compare by
|
||||
* identity and must not answer the HTTP request with it: an empty object here
|
||||
* means "no response was produced", not "the workflow responded with no body".
|
||||
*/
|
||||
export const EXECUTION_ENDED_WITHOUT_RESPONSE: IExecuteResponsePromiseData = Object.freeze({});
|
||||
|
||||
@@ -76,6 +76,7 @@ import { OwnershipService } from '@/services/ownership.service';
|
||||
import { ProtectedResourceRegistry } from '@/services/protected-resource.registry';
|
||||
import { WorkflowStatisticsService } from '@/services/workflow-statistics.service';
|
||||
import { WaitTracker } from '@/wait-tracker';
|
||||
import { EXECUTION_ENDED_WITHOUT_RESPONSE } from '@/webhooks/constants';
|
||||
import { WebhookExecutionContext } from '@/webhooks/webhook-execution-context';
|
||||
import { createMultiFormDataParser } from '@/webhooks/webhook-form-data';
|
||||
import { extractWebhookLastNodeResponse } from '@/webhooks/webhook-last-node-response-extractor';
|
||||
@@ -324,6 +325,13 @@ export function setupResponseNodePromise(
|
||||
): void {
|
||||
void responsePromise.promise
|
||||
.then(async (response: IN8nHttpFullResponse) => {
|
||||
if (response === EXECUTION_ENDED_WITHOUT_RESPONSE) {
|
||||
// The execution ended without the Respond to Webhook node running. The
|
||||
// post-execute handler answers instead, because only it knows whether the
|
||||
// execution failed.
|
||||
return;
|
||||
}
|
||||
|
||||
const binaryData = (response.body as IDataObject)?.binaryData as IBinaryData;
|
||||
if (binaryData?.id) {
|
||||
if (response.statusCode) {
|
||||
@@ -921,6 +929,17 @@ export async function executeWebhook(
|
||||
let responsePromise: IDeferredPromise<IN8nHttpFullResponse> | undefined;
|
||||
if (responseMode === 'responseNode') {
|
||||
responsePromise = createDeferredPromise<IN8nHttpFullResponse>();
|
||||
// Mark the request as answered as soon as the node produces a response, before
|
||||
// `setupResponseNodePromise` starts writing it. Streaming offloaded binary data
|
||||
// takes time, and a node failing during that wait must not answer a second time.
|
||||
void responsePromise.promise.then(
|
||||
(response) => {
|
||||
if (response !== EXECUTION_ENDED_WITHOUT_RESPONSE) didSendResponse = true;
|
||||
},
|
||||
() => {
|
||||
didSendResponse = true;
|
||||
},
|
||||
);
|
||||
setupResponseNodePromise(
|
||||
responsePromise,
|
||||
res,
|
||||
@@ -1064,6 +1083,14 @@ export async function executeWebhook(
|
||||
const lastNodeTaskData = WorkflowHelpers.getLastExecutedNodeData(runData);
|
||||
if (runData.data.resultData.error || lastNodeTaskData?.error !== undefined) {
|
||||
if (!didSendResponse) {
|
||||
// The node that failed is not named in the response, so log it here:
|
||||
// this is the only channel where naming it is safe.
|
||||
Container.get(Logger).warn('Webhook execution failed before a response was sent', {
|
||||
executionId,
|
||||
workflowId: workflowData.id,
|
||||
responseMode,
|
||||
lastNodeExecuted: runData.data.resultData.lastNodeExecuted,
|
||||
});
|
||||
responseCallback(null, {
|
||||
data: {
|
||||
message: 'Error in workflow',
|
||||
@@ -1078,6 +1105,12 @@ export async function executeWebhook(
|
||||
// in `responseNode` mode `responseCallback` is called by `responsePromise`
|
||||
if (responseMode === 'responseNode' && responsePromise) {
|
||||
await Promise.allSettled([responsePromise.promise]);
|
||||
if (!didSendResponse) {
|
||||
// The execution succeeded but never reached the Respond to Webhook node,
|
||||
// so answer here rather than leaving the caller waiting.
|
||||
responseCallback(null, { data: undefined, responseCode });
|
||||
didSendResponse = true;
|
||||
}
|
||||
return undefined;
|
||||
}
|
||||
|
||||
|
||||
@@ -0,0 +1,218 @@
|
||||
// Documents branch execution order around the Respond to Webhook node, for
|
||||
// CAT-4050 / GitHub issue #36175.
|
||||
//
|
||||
// These tests describe behaviour that already exists and pass on master. They pin
|
||||
// the ordering rule that produces the reported problem. They are not a regression
|
||||
// guard for the response handling fixed in `packages/cli`, which has its own tests.
|
||||
//
|
||||
// A webhook trigger in `responseMode: responseNode` fans out to two children: the
|
||||
// work branch, and a shared Respond to Webhook node. With `executionOrder: v1` the
|
||||
// engine runs the top-most child first, so the y coordinate of the Respond node
|
||||
// relative to the work branch decides whether the caller is acknowledged
|
||||
// immediately or only after the work finishes.
|
||||
//
|
||||
// Respond above the work node Respond below the work node
|
||||
// (y = 0 vs y = 500) (y = 1000 vs y = 500)
|
||||
//
|
||||
// Trigger Trigger
|
||||
// / \ / \
|
||||
// Respond Work Work Respond
|
||||
// (1st) (2nd) (1st) (2nd)
|
||||
//
|
||||
// When the work node runs first and fails, the Respond node never executes, so the
|
||||
// execution ends without producing a response. What the HTTP caller receives in
|
||||
// that case is decided in `packages/cli`, not here.
|
||||
|
||||
import { createDeferredPromise } from '@n8n/utils/promise/deferred-promise';
|
||||
import type { IExecuteFunctions, INodeTypeData, INodeTypeDescription, IRun } from 'n8n-workflow';
|
||||
import { NodeConnectionTypes, NodeOperationError, Workflow } from 'n8n-workflow';
|
||||
|
||||
import * as Helpers from '@test/helpers';
|
||||
|
||||
import { WorkflowExecute } from '../workflow-execute';
|
||||
|
||||
const RESPOND_NODE = 'Respond to Webhook';
|
||||
const WORK_NODE = 'Agent';
|
||||
const TRIGGER_NODE = 'Webhook';
|
||||
|
||||
/** Records the order in which nodes ran, so a test can assert on it. */
|
||||
let executed: string[] = [];
|
||||
/** Set by a test to make the work node fail, as a flaky model call would. */
|
||||
let workNodeFails = false;
|
||||
|
||||
const passThrough: INodeTypeDescription = {
|
||||
displayName: 'Test Node',
|
||||
name: 'testNode',
|
||||
group: ['transform'],
|
||||
version: 1,
|
||||
description: '',
|
||||
defaults: { name: 'Test Node' },
|
||||
inputs: [NodeConnectionTypes.Main],
|
||||
outputs: [NodeConnectionTypes.Main],
|
||||
properties: [],
|
||||
};
|
||||
|
||||
const nodeTypeData: INodeTypeData = {
|
||||
testTrigger: {
|
||||
sourcePath: '',
|
||||
type: {
|
||||
description: { ...passThrough, name: 'trigger', inputs: [] },
|
||||
async execute(this: IExecuteFunctions) {
|
||||
executed.push(this.getNode().name);
|
||||
return [[{ json: { body: { jobId: 'job-1' } } }]];
|
||||
},
|
||||
},
|
||||
},
|
||||
testWork: {
|
||||
sourcePath: '',
|
||||
type: {
|
||||
description: { ...passThrough, name: 'work' },
|
||||
async execute(this: IExecuteFunctions) {
|
||||
executed.push(this.getNode().name);
|
||||
if (workNodeFails) {
|
||||
throw new NodeOperationError(this.getNode(), 'Model call failed');
|
||||
}
|
||||
return [this.getInputData()];
|
||||
},
|
||||
},
|
||||
},
|
||||
testRespond: {
|
||||
sourcePath: '',
|
||||
type: {
|
||||
description: { ...passThrough, name: 'respond' },
|
||||
async execute(this: IExecuteFunctions) {
|
||||
executed.push(this.getNode().name);
|
||||
return [this.getInputData()];
|
||||
},
|
||||
},
|
||||
},
|
||||
};
|
||||
|
||||
/**
|
||||
* Builds the two-child fan-out. Only the Respond node's y coordinate differs
|
||||
* between the two cases; every other property is identical.
|
||||
*/
|
||||
function buildWorkflow(respondY: number) {
|
||||
const nodeTypes = Helpers.NodeTypes(nodeTypeData);
|
||||
|
||||
return new Workflow({
|
||||
id: 'repro',
|
||||
active: false,
|
||||
nodeTypes,
|
||||
settings: { executionOrder: 'v1' },
|
||||
nodes: [
|
||||
{
|
||||
id: '1',
|
||||
name: TRIGGER_NODE,
|
||||
type: 'testTrigger',
|
||||
typeVersion: 1,
|
||||
position: [0, 500],
|
||||
parameters: {},
|
||||
},
|
||||
{
|
||||
id: '2',
|
||||
name: WORK_NODE,
|
||||
type: 'testWork',
|
||||
typeVersion: 1,
|
||||
position: [300, 500],
|
||||
parameters: {},
|
||||
},
|
||||
{
|
||||
id: '3',
|
||||
name: RESPOND_NODE,
|
||||
type: 'testRespond',
|
||||
typeVersion: 1,
|
||||
position: [300, respondY],
|
||||
parameters: {},
|
||||
},
|
||||
],
|
||||
connections: {
|
||||
[TRIGGER_NODE]: {
|
||||
main: [
|
||||
[
|
||||
{ node: WORK_NODE, type: NodeConnectionTypes.Main, index: 0 },
|
||||
{ node: RESPOND_NODE, type: NodeConnectionTypes.Main, index: 0 },
|
||||
],
|
||||
],
|
||||
},
|
||||
},
|
||||
});
|
||||
}
|
||||
|
||||
async function runWorkflow(respondY: number) {
|
||||
const workflow = buildWorkflow(respondY);
|
||||
const waitPromise = createDeferredPromise<IRun>();
|
||||
const additionalData = Helpers.WorkflowExecuteAdditionalData(waitPromise);
|
||||
const workflowExecute = new WorkflowExecute(additionalData, 'webhook');
|
||||
|
||||
await workflowExecute.run({ workflow, startNode: workflow.getNode(TRIGGER_NODE)! });
|
||||
|
||||
return await waitPromise.promise;
|
||||
}
|
||||
|
||||
beforeEach(() => {
|
||||
executed = [];
|
||||
workNodeFails = false;
|
||||
});
|
||||
|
||||
describe('webhook responseNode branch ordering', () => {
|
||||
// The connection list is identical in both cases; only the canvas y differs.
|
||||
test('runs the Respond node first when it sits above the work node', async () => {
|
||||
await runWorkflow(0);
|
||||
|
||||
expect(executed).toEqual([TRIGGER_NODE, RESPOND_NODE, WORK_NODE]);
|
||||
});
|
||||
|
||||
test('runs the work node first when the Respond node sits below it', async () => {
|
||||
await runWorkflow(1000);
|
||||
|
||||
expect(executed).toEqual([TRIGGER_NODE, WORK_NODE, RESPOND_NODE]);
|
||||
});
|
||||
|
||||
test('skips the Respond node when the work node runs first and fails', async () => {
|
||||
workNodeFails = true;
|
||||
|
||||
const result = await runWorkflow(1000);
|
||||
|
||||
expect(executed).toEqual([TRIGGER_NODE, WORK_NODE]);
|
||||
expect(result.data.resultData.error).toBeDefined();
|
||||
// No response was sent, so the caller receives the node error.
|
||||
expect(result.data.resultData.runData[RESPOND_NODE]).toBeUndefined();
|
||||
});
|
||||
|
||||
test('still runs the Respond node when it precedes a failing work node', async () => {
|
||||
workNodeFails = true;
|
||||
|
||||
const result = await runWorkflow(0);
|
||||
|
||||
expect(executed).toEqual([TRIGGER_NODE, RESPOND_NODE, WORK_NODE]);
|
||||
expect(result.data.resultData.error).toBeDefined();
|
||||
// The caller already received its response, so the failure is invisible.
|
||||
expect(result.data.resultData.runData[RESPOND_NODE]).toBeDefined();
|
||||
});
|
||||
});
|
||||
|
||||
describe('the reported workflow layout', () => {
|
||||
// Canvas y coordinates taken from the attached export. The single shared
|
||||
// Respond to Webhook node is at y = 736.
|
||||
const SHARED_RESPOND_Y = 736;
|
||||
const branches = [
|
||||
{ trigger: 'Webhook Gemini', agentY: -240, respondsFirst: false },
|
||||
{ trigger: 'Webhook Chat GPT', agentY: 368, respondsFirst: false },
|
||||
{ trigger: 'Claude Webhook', agentY: 992, respondsFirst: true },
|
||||
{ trigger: 'Free Models Webhook', agentY: 1712, respondsFirst: true },
|
||||
];
|
||||
|
||||
test.each(branches)(
|
||||
'$trigger acknowledges before its agent: $respondsFirst',
|
||||
async ({ agentY, respondsFirst }) => {
|
||||
// Reuse the two-node fan-out with the agent fixed at y = 500 by
|
||||
// expressing the reported layout relative to the shared Respond node.
|
||||
const respondY = 500 + (SHARED_RESPOND_Y - agentY);
|
||||
|
||||
await runWorkflow(respondY);
|
||||
|
||||
expect(executed[1]).toBe(respondsFirst ? RESPOND_NODE : WORK_NODE);
|
||||
},
|
||||
);
|
||||
});
|
||||
Reference in New Issue
Block a user