n8n-io/n8n
1
0
Fork 0
mirror of https://github.com/n8n-io/n8n.git synced 2026-10-12 02:19:43 +00:00
Code Issues Packages Projects Releases Wiki Activity

refactor(engine): Split execution response endpoints (#39261)

Browse Source
This commit is contained in:
Tomi Turtiainen
2026-09-22 11:44:49 +00:00
committed by GitHub
parent 43d7cc755e
commit 518a24dd64
25 changed files with 355 additions and 365 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/@n8n/engine/src/execution/__tests__/execution-start.integration.test.ts
+2 -3
View File
@@ -14,14 +14,13 @@ import {
import { generateId } from '../../database/generate-id';
import type { WorkflowGraph } from '../../graph';
import { noopLifecycleEventPublisher } from '../../lifecycle-events';
import { createConsoleLogger } from '../../logging';
import {
InMemoryWorkQueue,
type OrchestrationMessage,
type StepMessage,
type WorkQueue,
} from '../../queue';
import { ExecutionResponseChannel, noopResponseTransport } from '../../response-channel';
import { noopExecutionResponseSender } from '../../response-channel';
import { ExecutionStartHandler } from '../execution-start-handler';
import { OrchestrationWorker } from '../orchestration-worker';
import { StartExecutionService } from '../start-execution.service';
@@ -76,7 +75,7 @@ describe('execution start (integration)', () => {
stepQueue,
orchestrationQueue,
noopLifecycleEventPublisher,
new ExecutionResponseChannel(noopResponseTransport, createConsoleLogger()),
noopExecutionResponseSender,
),
);
worker.start();
packages/@n8n/engine/src/execution/__tests__/step-execution.integration.test.ts
+2 -3
View File
@@ -17,9 +17,8 @@ import type { IStepExecutor, StepExecutionRequest } from '../../dependencies';
import type { WorkflowGraph } from '../../graph';
import { noopLifecycleEventPublisher } from '../../lifecycle-events';
import type { LifecycleEventCallback, LifecycleEvent } from '../../lifecycle-events';
import { createConsoleLogger } from '../../logging';
import { InMemoryWorkQueue, type OrchestrationMessage } from '../../queue';
import { ExecutionResponseChannel, noopResponseTransport } from '../../response-channel';
import { noopExecutionResponseSender } from '../../response-channel';
import { createEngineRuntime } from '../../runtime';
import type { TriggerOutputs } from '../execution.types';
import type { StartExecutionResult } from '../start-execution.service';
@@ -82,7 +81,7 @@ describe('step execution (integration)', () => {
dataSource,
admittance: new AllowAllAdmittance(),
identityVerifier: new SharedSecretIdentityVerifier(secret),
responseChannel: new ExecutionResponseChannel(noopResponseTransport, createConsoleLogger()),
responseSender: noopExecutionResponseSender,
// also how the test reaches the stores the runtime owns
externalDependencies: ({ executionStore }) => {
const finishExecution = executionStore.finishExecution.bind(executionStore);
packages/@n8n/engine/src/execution/__tests__/step-settled-handler.test.ts
+13 -13
View File
@@ -3,7 +3,7 @@ import { describe, expect, it, vi } from 'vitest';
import type { WorkflowGraph } from '../../graph';
import type { LifecycleEventPublisher } from '../../lifecycle-events';
import type { OrchestrationMessage, StepMessage, WorkQueue } from '../../queue';
import type { ExecutionResponseChannel } from '../../response-channel';
import type { ExecutionResponseSender } from '../../response-channel';
import type { ExecutionRecord, ExecutionStore } from '../execution-store';
import { stepKeyId, type StepKey, type StepStatus } from '../execution.types';
import { StepSettledHandler } from '../step-settled-handler';
@@ -128,9 +128,9 @@ function makeLifecycleEventPublisher(): LifecycleEventPublisher {
return { publish: vi.fn(), stop: vi.fn() };
}
/** A channel fake; tests that care assert on `publish`. */
function makeResponseChannel() {
return { publish: vi.fn() } as unknown as ExecutionResponseChannel;
/** A response sender fake; tests that care assert on `send`. */
function makeResponseSender() {
return { send: vi.fn() } as unknown as ExecutionResponseSender;
}
function makeHandler(
@@ -140,7 +140,7 @@ function makeHandler(
stepQueue = makeStepQueue(),
orchestrationQueue = makeOrchestrationQueue(),
lifecycleEventPublisher = makeLifecycleEventPublisher(),
responseChannel = makeResponseChannel(),
responseSender = makeResponseSender(),
} = {},
) {
return {
@@ -150,13 +150,13 @@ function makeHandler(
stepQueue,
orchestrationQueue,
lifecycleEventPublisher,
responseChannel,
responseSender,
),
executionStore,
stepQueue,
orchestrationQueue,
lifecycleEventPublisher,
responseChannel,
responseSender,
};
}
@@ -488,7 +488,7 @@ describe('StepSettledHandler lifecycle events', () => {
{ id: 'step-m', nodeId: 'm' },
{ countSettledSteps: vi.fn().mockResolvedValue(5) },
);
const { handler, lifecycleEventPublisher, responseChannel } = makeHandler(stepStore);
const { handler, lifecycleEventPublisher, responseSender } = makeHandler(stepStore);
await handler.handle({ ...event, stepId: 'step-m' });
@@ -496,7 +496,7 @@ describe('StepSettledHandler lifecycle events', () => {
type: 'execution:completed',
...finished,
});
expect(responseChannel.publish).toHaveBeenCalledExactlyOnceWith({
expect(responseSender.send).toHaveBeenCalledExactlyOnceWith({
type: 'ended',
executionId: 'exec-1',
workflowId: 'wf-1',
@@ -516,7 +516,7 @@ describe('StepSettledHandler lifecycle events', () => {
status: 'failed',
error: { name: 'NodeOperationError', message: 'it broke', stack: 'at trace' },
});
const { handler, lifecycleEventPublisher, responseChannel } = makeHandler(stepStore);
const { handler, lifecycleEventPublisher, responseSender } = makeHandler(stepStore);
await handler.handle(event);
@@ -524,7 +524,7 @@ describe('StepSettledHandler lifecycle events', () => {
type: 'execution:failed',
...finished,
});
expect(responseChannel.publish).toHaveBeenCalledExactlyOnceWith({
expect(responseSender.send).toHaveBeenCalledExactlyOnceWith({
type: 'ended',
executionId: 'exec-1',
workflowId: 'wf-1',
@@ -546,7 +546,7 @@ describe('StepSettledHandler lifecycle events', () => {
{},
{ finishExecution: vi.fn().mockResolvedValue(false) },
);
const { handler, lifecycleEventPublisher, responseChannel } = makeHandler(
const { handler, lifecycleEventPublisher, responseSender } = makeHandler(
makeStepStore({ status: 'failed' }),
{ executionStore },
);
@@ -554,7 +554,7 @@ describe('StepSettledHandler lifecycle events', () => {
await handler.handle(event);
expect(lifecycleEventPublisher.publish).not.toHaveBeenCalled();
expect(responseChannel.publish).not.toHaveBeenCalled();
expect(responseSender.send).not.toHaveBeenCalled();
});
it('announces nothing while any reachable node is unsettled', async () => {
packages/@n8n/engine/src/execution/step-settled-handler.ts
+3 -3
View File
@@ -7,7 +7,7 @@ import {
type GraphNode,
} from '../graph';
import type { LifecycleEventPublisher } from '../lifecycle-events';
import type { ExecutionResponseChannel } from '../response-channel';
import type { ExecutionResponseSender } from '../response-channel';
import type { OrchestrationMessage, StepMessage, StepSettledEvent, WorkQueue } from '../queue';
import { countExpectedSettledSteps } from './completion';
import type { ExecutionRecord, ExecutionStore } from './execution-store';
@@ -36,7 +36,7 @@ export class StepSettledHandler {
private readonly stepQueue: WorkQueue<StepMessage>,
private readonly orchestrationQueue: WorkQueue<OrchestrationMessage>,
private readonly lifecycleEventPublisher: LifecycleEventPublisher,
private readonly responseChannel: ExecutionResponseChannel,
private readonly responseSender: ExecutionResponseSender,
) {}
async handle(event: StepSettledEvent): Promise<void> {
@@ -227,7 +227,7 @@ export class StepSettledHandler {
);
}
this.responseChannel.publish({
this.responseSender.send({
type: 'ended',
executionId: execution.id,
workflowId: execution.workflowId,
packages/@n8n/engine/src/index.ts
+2 -9
View File
@@ -35,17 +35,10 @@ export type {
} from './lifecycle-events';
export {
ExecutionResponseChannel,
executionResponseSchema,
InMemoryResponseTransport,
noopResponseTransport,
} from './response-channel';
export type {
EndedMessage,
ExecutionResponse,
ResponseTransport,
Unsubscribe,
noopExecutionResponseSender,
} from './response-channel';
export type { EndedMessage, ExecutionResponse, ExecutionResponseSender } from './response-channel';
export { UnimplementedError } from './common';
export type { JsonObject, JsonValue } from './common';
packages/@n8n/engine/src/response-channel/__tests__/execution-response-channel.test.ts
-133
View File
@@ -1,133 +0,0 @@
import { describe, expect, it, vi } from 'vitest';
import type { EngineLogger } from '../../logging';
import { ExecutionResponseChannel } from '../execution-response-channel';
import type { ExecutionResponse } from '../execution-response.types';
import { InMemoryResponseTransport } from '../in-memory-transport';
import type { ResponseTransport, Unsubscribe } from '../response-transport';
const silentLogger = (): EngineLogger => ({
error: vi.fn(),
warn: vi.fn(),
info: vi.fn(),
debug: vi.fn(),
});
const ended = (outputs: unknown, executionId = 'exec-1'): ExecutionResponse => ({
type: 'ended',
executionId,
workflowId: 'wf-1',
status: 'completed',
lastStep: { nodeId: 'a', nodeName: 'A', status: 'completed', outputs: outputs as never },
});
/** A transport that keeps its frames, so a test can see what crossed. */
class RecordingTransport implements ResponseTransport {
readonly frames: string[] = [];
private handler?: (frame: string) => void;
publish(_executionId: string, frame: string): void {
this.frames.push(frame);
this.handler?.(frame);
}
subscribe(_executionId: string, handler: (frame: string) => void): Unsubscribe {
this.handler = handler;
return () => {
this.handler = undefined;
};
}
async stop(): Promise<void> {}
}
describe('ExecutionResponseChannel', () => {
const newChannel = () =>
new ExecutionResponseChannel(new InMemoryResponseTransport(), silentLogger());
it('delivers a response to every subscriber of that execution', () => {
const channel = newChannel();
const first: ExecutionResponse[] = [];
const second: ExecutionResponse[] = [];
channel.subscribe('exec-1', (r) => first.push(r));
channel.subscribe('exec-1', (r) => second.push(r));
channel.publish(ended([[{ json: { a: 1 } }]]));
expect(first[0]).toMatchObject({
type: 'ended',
status: 'completed',
lastStep: { nodeName: 'A', outputs: [[{ json: { a: 1 } }]] },
});
expect(second).toEqual(first);
});
it('tells one execution nothing about another', () => {
const channel = newChannel();
const seen: ExecutionResponse[] = [];
channel.subscribe('exec-1', (r) => seen.push(r));
channel.publish(ended(null, 'exec-2'));
expect(seen).toEqual([]);
});
it('stops delivering once the subscriber unsubscribes', () => {
const channel = newChannel();
const seen: ExecutionResponse[] = [];
const unsubscribe = channel.subscribe('exec-1', (r) => seen.push(r));
unsubscribe();
channel.publish(ended(null));
expect(seen).toEqual([]);
});
it('strips a value no transport could carry, rather than passing it live', () => {
const channel = newChannel();
const seen: ExecutionResponse[] = [];
channel.subscribe('exec-1', (r) => seen.push(r));
// The channel always serializes, so a `Date` crosses as the ISO string
// `JSON.stringify` gives it, never as the live object a subscriber could
// otherwise get away with in-process.
channel.publish(ended([[{ at: new Date(0) }]]));
expect(seen[0]).toMatchObject({
lastStep: { outputs: [[{ at: '1970-01-01T00:00:00.000Z' }]] },
});
});
it('discards a frame that is not a response', () => {
const transport = new RecordingTransport();
const channel = new ExecutionResponseChannel(transport, silentLogger());
const seen: ExecutionResponse[] = [];
channel.subscribe('exec-1', (r) => seen.push(r));
transport.publish('exec-1', '{"type":"nonsense"}');
transport.publish('exec-1', 'not json at all');
expect(seen).toEqual([]);
});
it('isolates a subscriber that throws', () => {
const channel = newChannel();
channel.subscribe('exec-1', () => {
throw new Error('boom');
});
expect(() => channel.publish(ended(null))).not.toThrow();
});
it('stops the underlying transport', async () => {
const transport = new RecordingTransport();
const stop = vi.spyOn(transport, 'stop');
const channel = new ExecutionResponseChannel(transport, silentLogger());
await channel.stop();
expect(stop).toHaveBeenCalledTimes(1);
});
});
packages/@n8n/engine/src/response-channel/execution-response-channel.ts
-64
View File
@@ -1,64 +0,0 @@
import type { EngineLogger } from '../logging';
import { executionResponseSchema } from './execution-response.schema';
import type { ExecutionResponse } from './execution-response.types';
import type { ResponseTransport, Unsubscribe } from './response-transport';
/**
* Sends responses from an execution back to its caller.
*
* The engine publishes a response with an execution ID. A caller subscribes
* with the same ID and receives only responses for that execution.
*
* This class serializes outgoing responses and validates incoming responses.
* The transport delivers them in-process or across process boundaries.
*/
export class ExecutionResponseChannel {
constructor(
private readonly transport: ResponseTransport,
private readonly logger: EngineLogger,
) {}
publish(response: ExecutionResponse): void {
this.transport.publish(response.executionId, JSON.stringify(response));
}
/** Listens to one execution. Call the returned function once the run is answered. */
subscribe(executionId: string, handler: (response: ExecutionResponse) => void): Unsubscribe {
return this.transport.subscribe(executionId, (frame) => {
const response = this.fromFrame(frame);
if (response === undefined) return;
try {
handler(response);
} catch (error) {
// One bad subscriber must not take the transport down with it.
this.logger.error('a response subscriber threw', {
executionId: response.executionId,
type: response.type,
error,
});
}
});
}
async stop(): Promise<void> {
await this.transport.stop();
}
private fromFrame(frame: string): ExecutionResponse | undefined {
try {
const parsed = executionResponseSchema.safeParse(JSON.parse(frame));
if (!parsed.success) {
this.logger.error('discarding a malformed response', {
details: parsed.error.flatten(),
});
return undefined;
}
return parsed.data;
} catch (error) {
this.logger.error('discarding an unreadable response frame', { error });
return undefined;
}
}
}
packages/@n8n/engine/src/response-channel/execution-response-sender.ts
+13
View File
@@ -0,0 +1,13 @@
import type { ExecutionResponse } from './execution-response.types';
/** Sends responses from an execution back to its caller. */
export interface ExecutionResponseSender {
send(response: ExecutionResponse): void;
stop(): Promise<void>;
}
/** Response sender for a host that discards execution responses. */
export const noopExecutionResponseSender: ExecutionResponseSender = Object.freeze({
send: () => {},
stop: async () => {},
});
packages/@n8n/engine/src/response-channel/execution-response.schema.ts
+2 -3
View File
@@ -6,9 +6,8 @@ import { jsonValueSchema } from '../common';
import { SETTLED_STEP_STATUSES } from '../execution/execution.types';
/**
* The one definition of a response's shape. A transport can cross a process
* boundary, so a frame that arrives is parsed against this before a subscriber
* ever sees it.
* The one definition of a response's shape. A frame can cross a process
* boundary, so it is parsed against this before a handler sees it.
*/
export const executionResponseSchema = z.discriminatedUnion('type', [
z.object({
packages/@n8n/engine/src/response-channel/in-memory-transport.ts
-33
View File
@@ -1,33 +0,0 @@
import { EventEmitter } from 'node:events';
import type { ResponseTransport, Unsubscribe } from './response-transport';
/**
* In-process `ResponseTransport`, for a deployment where both planes share a
* process.
*
* One emitter event per execution, which is the same addressing a networked
* transport gives each run.
*
* It carries frames rather than objects, exactly as a networked transport does.
* The `JSON.stringify` it pays for is the point: an integrated deployment that
* passed live objects would accept a response a split deployment mangles, and
* the integrated one is the one under test.
*/
export class InMemoryResponseTransport implements ResponseTransport {
private readonly executions = new EventEmitter();
publish(executionId: string, frame: string): void {
this.executions.emit(executionId, frame);
}
subscribe(executionId: string, handler: (frame: string) => void): Unsubscribe {
this.executions.on(executionId, handler);
return () => this.executions.off(executionId, handler);
}
async stop(): Promise<void> {
this.executions.removeAllListeners();
}
}
packages/@n8n/engine/src/response-channel/index.ts
+2 -4
View File
@@ -1,6 +1,4 @@
export { ExecutionResponseChannel } from './execution-response-channel';
export { noopExecutionResponseSender } from './execution-response-sender';
export type { ExecutionResponseSender } from './execution-response-sender';
export { executionResponseSchema } from './execution-response.schema';
export type { EndedMessage, ExecutionResponse } from './execution-response.types';
export { InMemoryResponseTransport } from './in-memory-transport';
export { noopResponseTransport } from './response-transport';
export type { ResponseTransport, Unsubscribe } from './response-transport';
packages/@n8n/engine/src/response-channel/response-transport.ts
-26
View File
@@ -1,26 +0,0 @@
/** Ends one subscription. Calling it twice is safe. */
export type Unsubscribe = () => void;
/**
* Delivers serialized responses to subscribers for a specific execution.
*
* A subscriber receives only frames published for its execution ID. The
* transport treats each frame as an opaque string. `ExecutionResponseChannel`
* handles serialization and validation.
*
* Frames are not buffered. Only active subscribers receive them, and each
* frame is delivered at most once.
*/
export interface ResponseTransport {
/** Never throws and never blocks: a step must not wait on the response path. */
publish(executionId: string, frame: string): void;
subscribe(executionId: string, handler: (frame: string) => void): Unsubscribe;
stop(): Promise<void>;
}
/** Transport for a host that is not listening. */
export const noopResponseTransport: ResponseTransport = Object.freeze({
publish: () => {},
subscribe: () => () => {},
stop: async () => {},
});
packages/@n8n/engine/src/runtime/__tests__/create-engine-runtime.test.ts
+6 -7
View File
@@ -6,8 +6,7 @@ import { AllowAllAdmittance } from '../../admittance';
import { mintIdentityToken, SharedSecretIdentityVerifier } from '../../auth';
import type { EngineStores } from '../../database';
import { BatchingLifecycleEventPublisher } from '../../lifecycle-events';
import { createConsoleLogger } from '../../logging';
import { ExecutionResponseChannel, noopResponseTransport } from '../../response-channel';
import { noopExecutionResponseSender } from '../../response-channel';
import { createEngineRuntime } from '../create-engine-runtime';
/** Enough of a `DataSource` for the stores: they only hold on to a repository. */
@@ -16,14 +15,14 @@ const fakeDataSource = () => ({ getRepository: vi.fn(() => ({})) }) as unknown a
const secret = 'a'.repeat(32);
const identityVerifier = new SharedSecretIdentityVerifier(secret);
const token = mintIdentityToken(secret, { cpId: 'cp-1', tenantId: 'tenant-1' });
const responseChannel = new ExecutionResponseChannel(noopResponseTransport, createConsoleLogger());
const responseSender = noopExecutionResponseSender;
const runtime = () =>
createEngineRuntime({
dataSource: fakeDataSource(),
admittance: new AllowAllAdmittance(),
identityVerifier,
responseChannel,
responseSender,
});
describe('createEngineRuntime', () => {
@@ -68,7 +67,7 @@ describe('createEngineRuntime', () => {
dataSource: fakeDataSource(),
admittance: new AllowAllAdmittance(),
identityVerifier,
responseChannel,
responseSender,
externalDependencies: (given) => {
stores = given;
return {};
@@ -87,7 +86,7 @@ describe('createEngineRuntime', () => {
dataSource: fakeDataSource(),
admittance: new AllowAllAdmittance(),
identityVerifier,
responseChannel,
responseSender,
externalDependencies: build,
});
@@ -102,7 +101,7 @@ describe('createEngineRuntime', () => {
dataSource: fakeDataSource(),
admittance: new AllowAllAdmittance(),
identityVerifier,
responseChannel,
responseSender,
externalDependencies: () => ({
lifecycleEventCallback: vi.fn().mockResolvedValue(undefined),
}),
packages/@n8n/engine/src/runtime/create-engine-runtime.ts
+6 -7
View File
@@ -20,7 +20,7 @@ import type { LifecycleEventPublisher } from '../lifecycle-events';
import { createConsoleLogger, type EngineLogger } from '../logging';
import { InMemoryWorkQueue } from '../queue';
import type { OrchestrationMessage, StepMessage } from '../queue';
import type { ExecutionResponseChannel } from '../response-channel';
import type { ExecutionResponseSender } from '../response-channel';
import { createEngineServer } from '../server';
export interface EngineRuntimeOptions {
@@ -32,11 +32,10 @@ export interface EngineRuntimeOptions {
/** Where the engine writes its own messages. Defaults to the console. */
logger?: EngineLogger;
/**
* Where an execution's responses go. The host owns it, because whoever waits
* for a response subscribes to the same channel. No default: a host that
* listens to nothing says so with `noopResponseTransport`.
* Where an execution sends responses. The host owns it. No default: a host
* that discards responses says so with `noopExecutionResponseSender`.
*/
responseChannel: ExecutionResponseChannel;
responseSender: ExecutionResponseSender;
/**
* Builds the capabilities the engine does not own. It receives the engine's
* stores, because a `v1-node` executor reads step data through them and the
@@ -72,7 +71,7 @@ export function createEngineRuntime({
admittance,
identityVerifier,
logger = createConsoleLogger(),
responseChannel,
responseSender,
externalDependencies,
}: EngineRuntimeOptions): EngineRuntime {
const orchestrationQueue = new InMemoryWorkQueue<OrchestrationMessage>(logger);
@@ -99,7 +98,7 @@ export function createEngineRuntime({
stepQueue,
orchestrationQueue,
lifecycleEventPublisher,
responseChannel,
responseSender,
),
);
const stepWorker = new StepWorker(
packages/@n8n/engine/src/serve.ts
+2 -2
View File
@@ -5,7 +5,7 @@ import { AllowAllAdmittance } from './admittance';
import { SharedSecretIdentityVerifier } from './auth';
import { createDataSource } from './database';
import { createConsoleLogger } from './logging';
import { ExecutionResponseChannel, noopResponseTransport } from './response-channel';
import { noopExecutionResponseSender } from './response-channel';
import { createEngineRuntime } from './runtime';
const logger = createConsoleLogger();
@@ -34,7 +34,7 @@ async function main(): Promise<void> {
logger,
// Nothing in this process waits for a response: the caller reads the run
// over the API instead.
responseChannel: new ExecutionResponseChannel(noopResponseTransport, logger),
responseSender: noopExecutionResponseSender,
});
runtime.start();
packages/@n8n/engine/src/server/__tests__/workflow-executions.integration.test.ts
+2 -3
View File
@@ -11,9 +11,8 @@ import { createDataSource, createStores, WorkflowExecution } from '../../databas
import { generateId } from '../../database/generate-id';
import { ExecutionQueryService, StartExecutionService } from '../../execution';
import type { WorkflowGraph } from '../../graph';
import { createConsoleLogger } from '../../logging';
import type { OrchestrationMessage, WorkQueue } from '../../queue';
import { ExecutionResponseChannel, noopResponseTransport } from '../../response-channel';
import { noopExecutionResponseSender } from '../../response-channel';
import { createEngineRuntime } from '../../runtime';
import { startEngineServer } from '../../testing/start-engine-server';
import type { SearchExecutionsResponse } from '../api.types';
@@ -488,7 +487,7 @@ describe('GET /api/workflow-executions/:id (integration)', () => {
dataSource,
admittance: new AllowAllAdmittance(),
identityVerifier: new SharedSecretIdentityVerifier(secret),
responseChannel: new ExecutionResponseChannel(noopResponseTransport, createConsoleLogger()),
responseSender: noopExecutionResponseSender,
externalDependencies: ({ executionStore }) => {
const finishExecution = executionStore.finishExecution.bind(executionStore);
vi.spyOn(executionStore, 'finishExecution').mockImplementation(async (id, status) => {
packages/@n8n/node-engine-compatibility/src/__tests__/acceptance-fixtures.ts
+2 -10
View File
@@ -1,6 +1,5 @@
import type {
createDataSource,
EngineLogger,
ExecutionMode,
StartExecutionResult,
TriggerOutputs,
@@ -9,8 +8,7 @@ import type {
import {
AllowAllAdmittance,
createEngineRuntime,
ExecutionResponseChannel,
noopResponseTransport,
noopExecutionResponseSender,
mintIdentityToken,
SharedSecretIdentityVerifier,
WorkflowExecution,
@@ -58,12 +56,6 @@ export const realNodeTypes: INodeTypes = {
export const converter = new V1WorkflowConverter();
const authSecret = 'a'.repeat(32);
const silentLogger: EngineLogger = {
error: () => {},
warn: () => {},
info: () => {},
debug: () => {},
};
const caller = { cpId: 'cp-1', tenantId: 'tenant-1' };
export type Assignment = { name: string; value: string | number; type: string };
@@ -332,7 +324,7 @@ export function makeRunWorkflow(getDataSource: () => EngineDataSource) {
admittance: new AllowAllAdmittance(),
identityVerifier: new SharedSecretIdentityVerifier(authSecret),
// Nothing here waits for a response; the fixture reads the run over the API.
responseChannel: new ExecutionResponseChannel(noopResponseTransport, silentLogger),
responseSender: noopExecutionResponseSender,
// also how the test reaches the stores the runtime owns
externalDependencies: ({ executionStore, stepStore }) => {
const finishExecution = executionStore.finishExecution.bind(executionStore);
packages/cli/src/modules/engine-v2/__tests__/engine-v2.runtime.test.ts
+22 -22
View File
@@ -1,6 +1,6 @@
import { mockLogger } from '@n8n/backend-test-utils';
import type { EngineConfig } from '@n8n/config';
import type { ExecutionResponseChannel } from '@n8n/engine';
import type { ExecutionResponseSender } from '@n8n/engine';
import { mock } from 'vitest-mock-extended';
import type { CredentialTypes } from '@/credential-types';
@@ -103,8 +103,8 @@ describe('EngineV2Runtime', () => {
const credentialsHelper = mock<CredentialsHelper>();
const credentialTypes = mock<CredentialTypes>();
// `@n8n/engine` is mocked below, so the channel is a stand-in too.
const responseChannel = () => mock<ExecutionResponseChannel>();
// `@n8n/engine` is mocked below, so the sender is a stand-in too.
const responseSender = () => mock<ExecutionResponseSender>();
const newRuntime = (databaseUrl = 'postgres://engine') =>
new EngineV2Runtime(
@@ -149,7 +149,7 @@ describe('EngineV2Runtime', () => {
describe('init', () => {
it('refuses to start without a data plane database', async () => {
await expect(newRuntime('').init(responseChannel())).rejects.toThrow(
await expect(newRuntime('').init(responseSender())).rejects.toThrow(
'N8N_ENGINE_DATABASE_URL',
);
@@ -157,7 +157,7 @@ describe('EngineV2Runtime', () => {
});
it('opens the data plane connection and migrates it before building the engine', async () => {
await newRuntime().init(responseChannel());
await newRuntime().init(responseSender());
expect(mocks.createDataSource).toHaveBeenCalledWith('postgres://engine');
expect(mocks.dataSource.initialize).toHaveBeenCalled();
@@ -168,13 +168,13 @@ describe('EngineV2Runtime', () => {
});
it('starts the engine', async () => {
await newRuntime().init(responseChannel());
await newRuntime().init(responseSender());
expect(mocks.engine.start).toHaveBeenCalled();
});
it('injects a lifecycle event callback that reports to the control plane', async () => {
await newRuntime().init(responseChannel());
await newRuntime().init(responseSender());
const events = [
{
@@ -198,7 +198,7 @@ describe('EngineV2Runtime', () => {
});
it('injects the v1 step executor so v1-node steps can run', async () => {
await newRuntime().init(responseChannel());
await newRuntime().init(responseSender());
const stores = { executionStore: {}, stepStore: {} };
@@ -215,7 +215,7 @@ describe('EngineV2Runtime', () => {
});
it('builds the v1 additional data for the workflow and user of the step', async () => {
await newRuntime().init(responseChannel());
await newRuntime().init(responseSender());
await additionalDataFactory()(stepContext);
@@ -227,7 +227,7 @@ describe('EngineV2Runtime', () => {
});
it('tags the v1 additional data with the engine execution id', async () => {
await newRuntime().init(responseChannel());
await newRuntime().init(responseSender());
const additionalData = await additionalDataFactory()(stepContext);
@@ -235,7 +235,7 @@ describe('EngineV2Runtime', () => {
});
it('gives the step a credentials helper that asks the control plane', async () => {
await newRuntime().init(responseChannel());
await newRuntime().init(responseSender());
const additionalData = await additionalDataFactory()(stepContext);
@@ -243,7 +243,7 @@ describe('EngineV2Runtime', () => {
});
it('binds the credentials helper to the execution of the step', async () => {
await newRuntime().init(responseChannel());
await newRuntime().init(responseSender());
vi.mocked(credentialsClient.resolve).mockResolvedValue({});
const { credentialsHelper: helper } = (await additionalDataFactory()(stepContext)) as {
@@ -267,7 +267,7 @@ describe('EngineV2Runtime', () => {
});
it('serves the engine API on the configured address', async () => {
await newRuntime().init(responseChannel());
await newRuntime().init(responseSender());
expect(mocks.engine.app.listen).toHaveBeenCalledWith(3000, '0.0.0.0');
});
@@ -275,7 +275,7 @@ describe('EngineV2Runtime', () => {
it('closes the connection when migrations fail', async () => {
mocks.dataSource.runMigrations.mockRejectedValueOnce(new Error('migration failed'));
await expect(newRuntime().init(responseChannel())).rejects.toThrow('migration failed');
await expect(newRuntime().init(responseSender())).rejects.toThrow('migration failed');
expect(mocks.dataSource.destroy).toHaveBeenCalled();
expect(mocks.engine.stop).not.toHaveBeenCalled();
@@ -284,7 +284,7 @@ describe('EngineV2Runtime', () => {
it('stops the engine and closes the connection when the server fails to listen', async () => {
mocks.listen.error = new Error('listen failed');
await expect(newRuntime().init(responseChannel())).rejects.toThrow('listen failed');
await expect(newRuntime().init(responseSender())).rejects.toThrow('listen failed');
expect(mocks.server.close).not.toHaveBeenCalled();
expect(mocks.engine.stop).toHaveBeenCalled();
@@ -295,14 +295,14 @@ describe('EngineV2Runtime', () => {
mocks.listen.error = new Error('listen failed');
mocks.engine.stop.mockRejectedValueOnce(new Error('stop failed'));
await expect(newRuntime().init(responseChannel())).rejects.toThrow('listen failed');
await expect(newRuntime().init(responseSender())).rejects.toThrow('listen failed');
});
});
describe('shutdown', () => {
it('closes the server, stops the engine and closes the connection', async () => {
const runtime = newRuntime();
await runtime.init(responseChannel());
await runtime.init(responseSender());
await runtime.shutdown();
@@ -313,7 +313,7 @@ describe('EngineV2Runtime', () => {
it('stops the engine and closes the connection when the server fails to close', async () => {
const runtime = newRuntime();
await runtime.init(responseChannel());
await runtime.init(responseSender());
mocks.server.close.mockImplementationOnce((done) => done(new Error('close failed')));
const error = await runtime
@@ -328,7 +328,7 @@ describe('EngineV2Runtime', () => {
it('closes the connection when the engine fails to stop', async () => {
const runtime = newRuntime();
await runtime.init(responseChannel());
await runtime.init(responseSender());
mocks.engine.stop.mockRejectedValueOnce(new Error('stop failed'));
await expect(runtime.shutdown()).rejects.toThrow(AggregateError);
@@ -338,7 +338,7 @@ describe('EngineV2Runtime', () => {
it('aborts the credential requests of running steps', async () => {
const runtime = newRuntime();
await runtime.init(responseChannel());
await runtime.init(responseSender());
vi.mocked(credentialsClient.resolve).mockResolvedValue({});
const { credentialsHelper: helper } = (await additionalDataFactory()(stepContext)) as {
credentialsHelper: RemoteCredentialsHelper;
@@ -365,7 +365,7 @@ describe('EngineV2Runtime', () => {
it('releases a resource again when its first release failed', async () => {
const runtime = newRuntime();
await runtime.init(responseChannel());
await runtime.init(responseSender());
mocks.server.close.mockImplementationOnce((done) => done(new Error('close failed')));
await expect(runtime.shutdown()).rejects.toThrow(AggregateError);
@@ -380,7 +380,7 @@ describe('EngineV2Runtime', () => {
it('is safe to call twice', async () => {
const runtime = newRuntime();
await runtime.init(responseChannel());
await runtime.init(responseSender());
await runtime.shutdown();
await expect(runtime.shutdown()).resolves.toBeUndefined();
packages/cli/src/modules/engine-v2/__tests__/in-memory-execution-response.test.ts
+127
View File
@@ -0,0 +1,127 @@
import { mockLogger } from '@n8n/backend-test-utils';
import type { ExecutionResponse } from '@n8n/engine';
import { InMemoryExecutionResponseChannel } from '../response-channel/in-memory-execution-response-channel';
import { InMemoryExecutionResponseReceiver } from '../response-channel/in-memory-execution-response-receiver';
import { InMemoryExecutionResponseSender } from '../response-channel/in-memory-execution-response-sender';
const ended = (executionId = 'exec-1', outputs: unknown = null): ExecutionResponse => ({
type: 'ended',
executionId,
workflowId: 'wf-1',
status: 'completed',
lastStep: { nodeId: 'a', nodeName: 'A', status: 'completed', outputs: outputs as never },
});
describe('in-memory execution responses', () => {
it('serializes a response before publishing it', () => {
const channel = new InMemoryExecutionResponseChannel();
const publish = vi.spyOn(channel, 'publish');
const sender = new InMemoryExecutionResponseSender(channel);
sender.send(ended('exec-1', [[{ at: new Date(0) }]]));
expect(publish).toHaveBeenCalledExactlyOnceWith(
'exec-1',
JSON.stringify(ended('exec-1', [[{ at: '1970-01-01T00:00:00.000Z' }]])),
);
});
it('validates a response before delivering it', () => {
const channel = new InMemoryExecutionResponseChannel();
const receiver = new InMemoryExecutionResponseReceiver(channel, mockLogger());
const seen: ExecutionResponse[] = [];
receiver.receive('exec-1', (response) => seen.push(response));
channel.publish('exec-1', JSON.stringify(ended()));
expect(seen).toEqual([ended()]);
});
it('discards malformed and unreadable responses', () => {
const channel = new InMemoryExecutionResponseChannel();
const receiver = new InMemoryExecutionResponseReceiver(channel, mockLogger());
const seen: ExecutionResponse[] = [];
receiver.receive('exec-1', (response) => seen.push(response));
channel.publish('exec-1', '{"type":"nonsense"}');
channel.publish('exec-1', 'not json at all');
expect(seen).toEqual([]);
});
it('isolates a handler that throws', () => {
const channel = new InMemoryExecutionResponseChannel();
const receiver = new InMemoryExecutionResponseReceiver(channel, mockLogger());
receiver.receive('exec-1', () => {
throw new Error('boom');
});
expect(() => channel.publish('exec-1', JSON.stringify(ended()))).not.toThrow();
});
it('delivers a response to every handler for that execution', () => {
const channel = new InMemoryExecutionResponseChannel();
const sender = new InMemoryExecutionResponseSender(channel);
const receiver = new InMemoryExecutionResponseReceiver(channel, mockLogger());
const first: ExecutionResponse[] = [];
const second: ExecutionResponse[] = [];
receiver.receive('exec-1', (response) => first.push(response));
receiver.receive('exec-1', (response) => second.push(response));
sender.send(ended());
expect(first).toEqual([ended()]);
expect(second).toEqual(first);
});
it('does not deliver a response to another execution', () => {
const channel = new InMemoryExecutionResponseChannel();
const sender = new InMemoryExecutionResponseSender(channel);
const receiver = new InMemoryExecutionResponseReceiver(channel, mockLogger());
const seen: ExecutionResponse[] = [];
receiver.receive('exec-1', (response) => seen.push(response));
sender.send(ended('exec-2'));
expect(seen).toEqual([]);
});
it('stops delivery after the handler is removed', () => {
const channel = new InMemoryExecutionResponseChannel();
const sender = new InMemoryExecutionResponseSender(channel);
const receiver = new InMemoryExecutionResponseReceiver(channel, mockLogger());
const seen: ExecutionResponse[] = [];
const unsubscribe = receiver.receive('exec-1', (response) => seen.push(response));
unsubscribe();
sender.send(ended());
expect(seen).toEqual([]);
});
it('stops publishing through the sender', async () => {
const channel = new InMemoryExecutionResponseChannel();
const sender = new InMemoryExecutionResponseSender(channel);
const seen: string[] = [];
channel.subscribe('exec-1', (frame) => seen.push(frame));
await sender.stop();
sender.send(ended());
expect(seen).toEqual([]);
});
it('stops delivery through the receiver', async () => {
const channel = new InMemoryExecutionResponseChannel();
const sender = new InMemoryExecutionResponseSender(channel);
const receiver = new InMemoryExecutionResponseReceiver(channel, mockLogger());
const seen: ExecutionResponse[] = [];
receiver.receive('exec-1', (response) => seen.push(response));
await receiver.stop();
sender.send(ended());
expect(seen).toEqual([]);
});
});
packages/cli/src/modules/engine-v2/engine-v2.module.ts
+26 -13
View File
@@ -3,7 +3,7 @@ import { EngineConfig, ExecutionsConfig } from '@n8n/config';
import type { ModuleInterface } from '@n8n/decorators';
import { BackendModule, OnShutdown } from '@n8n/decorators';
import { Container } from '@n8n/di';
import type { ExecutionResponseChannel } from '@n8n/engine';
import type { ExecutionResponseSender } from '@n8n/engine';
import { UserError } from 'n8n-workflow';
import { randomBytes } from 'node:crypto';
@@ -19,7 +19,9 @@ import { randomBytes } from 'node:crypto';
*/
@BackendModule({ name: 'engine-v2', instanceTypes: ['main'] })
export class EngineV2Module implements ModuleInterface {
private responseChannel?: ExecutionResponseChannel;
private responseSender?: ExecutionResponseSender;
private responseReceiver?: { stop(): Promise<void> };
async init() {
if (Container.get(ExecutionsConfig).mode === 'queue') {
@@ -37,19 +39,30 @@ export class EngineV2Module implements ModuleInterface {
const { EngineControlPlaneServer } = await import('./engine-control-plane-server.js');
await Container.get(EngineControlPlaneServer).start();
// One channel, both planes. Handed over before the engine starts: a short
// run answers before `startExecution` returns, and nobody replays.
const { ExecutionResponseChannel, InMemoryResponseTransport } = await import('@n8n/engine');
// In-memory for now: both planes share this process. A transport that
// crosses one arrives with CAT-4572.
const responseChannel = new ExecutionResponseChannel(
new InMemoryResponseTransport(),
// Create both endpoints before the engine starts. A short run can answer
// before `startExecution` returns, and responses are not replayed.
const { InMemoryExecutionResponseChannel } = await import(
'./response-channel/in-memory-execution-response-channel.js'
);
const { InMemoryExecutionResponseSender } = await import(
'./response-channel/in-memory-execution-response-sender.js'
);
const { InMemoryExecutionResponseReceiver } = await import(
'./response-channel/in-memory-execution-response-receiver.js'
);
// In-memory for now because both planes share this process. Redis endpoints
// can use the same response contracts when the planes run separately.
const responseChannel = new InMemoryExecutionResponseChannel();
const responseSender = new InMemoryExecutionResponseSender(responseChannel);
const responseReceiver = new InMemoryExecutionResponseReceiver(
responseChannel,
Container.get(Logger).scoped('engine-v2'),
);
this.responseChannel = responseChannel;
this.responseSender = responseSender;
this.responseReceiver = responseReceiver;
const { EngineV2Runtime } = await import('./engine-v2.runtime.js');
await Container.get(EngineV2Runtime).init(responseChannel);
await Container.get(EngineV2Runtime).init(responseSender);
const { EngineDataPlaneClient } = await import('./engine-data-plane-client.js');
const { EngineDataPlaneProxyService } = await import(
@@ -67,8 +80,8 @@ export class EngineV2Module implements ModuleInterface {
await Container.get(EngineV2Runtime).shutdown();
} finally {
// After the engine, so a final response still has somewhere to go.
// In a `finally`, so a failed runtime shutdown still releases the channel.
await this.responseChannel?.stop();
// A failed runtime shutdown must still release both endpoints.
await Promise.all([this.responseSender?.stop(), this.responseReceiver?.stop()]);
}
// After the engine, so its final flush still has somewhere to land.
packages/cli/src/modules/engine-v2/engine-v2.runtime.ts
+5 -7
View File
@@ -1,7 +1,7 @@
import { Logger } from '@n8n/backend-common';
import { EngineConfig } from '@n8n/config';
import { Service } from '@n8n/di';
import type { EngineRuntime, ExecutionResponseChannel } from '@n8n/engine';
import type { EngineRuntime, ExecutionResponseSender } from '@n8n/engine';
import {
AllowAllAdmittance,
createDataSource,
@@ -59,11 +59,11 @@ export class EngineV2Runtime {
this.logger = this.logger.scoped('engine-v2');
}
async init(responseChannel: ExecutionResponseChannel): Promise<void> {
async init(responseSender: ExecutionResponseSender): Promise<void> {
try {
await this.initDb();
this.initEngine(responseChannel);
this.initEngine(responseSender);
await this.initServer();
} catch (error) {
@@ -92,7 +92,7 @@ export class EngineV2Runtime {
await this.dataSource.runMigrations();
}
private initEngine(responseChannel: ExecutionResponseChannel): void {
private initEngine(responseSender: ExecutionResponseSender): void {
assert(this.dataSource, 'Engine 2.0 cannot start without a data source');
const stopping = new AbortController();
@@ -105,9 +105,7 @@ export class EngineV2Runtime {
admittance: new AllowAllAdmittance(),
identityVerifier: new SharedSecretIdentityVerifier(this.engineConfig.authSecret),
logger: this.logger,
// Whoever waits for a response subscribes to the same channel, so the
// host owns it and hands it in.
responseChannel,
responseSender,
externalDependencies: ({ executionStore, stepStore }) => ({
lifecycleEventCallback: async (events, signal) =>
await this.controlPlaneClient.sendLifecycleEvents(events, signal),
packages/cli/src/modules/engine-v2/response-channel/execution-response-receiver.ts
+12
View File
@@ -0,0 +1,12 @@
import type { ExecutionResponse } from '@n8n/engine';
export type UnsubscribeExecutionResponse = () => void;
/** Receives responses from executions and delivers them to their callers. */
export interface ExecutionResponseReceiver {
receive(
executionId: string,
handler: (response: ExecutionResponse) => void,
): UnsubscribeExecutionResponse;
stop(): Promise<void>;
}
packages/cli/src/modules/engine-v2/response-channel/in-memory-execution-response-channel.ts
+17
View File
@@ -0,0 +1,17 @@
import { EventEmitter } from 'node:events';
import type { UnsubscribeExecutionResponse } from './execution-response-receiver';
/** Carries serialized execution responses between endpoints in this process. */
export class InMemoryExecutionResponseChannel {
private readonly emitter = new EventEmitter();
publish(executionId: string, frame: string): void {
this.emitter.emit(executionId, frame);
}
subscribe(executionId: string, handler: (frame: string) => void): UnsubscribeExecutionResponse {
this.emitter.on(executionId, handler);
return () => this.emitter.off(executionId, handler);
}
}
packages/cli/src/modules/engine-v2/response-channel/in-memory-execution-response-receiver.ts
+70
View File
@@ -0,0 +1,70 @@
import type { Logger } from '@n8n/backend-common';
import { executionResponseSchema, type ExecutionResponse } from '@n8n/engine';
import type {
ExecutionResponseReceiver,
UnsubscribeExecutionResponse,
} from './execution-response-receiver';
import type { InMemoryExecutionResponseChannel } from './in-memory-execution-response-channel';
export class InMemoryExecutionResponseReceiver implements ExecutionResponseReceiver {
private readonly subscriptions = new Set<UnsubscribeExecutionResponse>();
private stopped = false;
constructor(
private readonly channel: InMemoryExecutionResponseChannel,
private readonly logger: Logger,
) {}
receive(
executionId: string,
handler: (response: ExecutionResponse) => void,
): UnsubscribeExecutionResponse {
if (this.stopped) return () => {};
const unsubscribeFromChannel = this.channel.subscribe(executionId, (frame) => {
const response = this.fromFrame(frame);
if (response === undefined) return;
try {
handler(response);
} catch (error) {
this.logger.error('A response handler threw', {
executionId: response.executionId,
type: response.type,
error,
});
}
});
const unsubscribe = () => {
unsubscribeFromChannel();
this.subscriptions.delete(unsubscribe);
};
this.subscriptions.add(unsubscribe);
return unsubscribe;
}
async stop(): Promise<void> {
this.stopped = true;
for (const unsubscribe of this.subscriptions) unsubscribe();
}
private fromFrame(frame: string): ExecutionResponse | undefined {
try {
const parsed = executionResponseSchema.safeParse(JSON.parse(frame));
if (!parsed.success) {
this.logger.error('Discarding a malformed response', {
details: parsed.error.flatten(),
});
return undefined;
}
return parsed.data;
} catch (error) {
this.logger.error('Discarding an unreadable response', { error });
return undefined;
}
}
}
packages/cli/src/modules/engine-v2/response-channel/in-memory-execution-response-sender.ts
+19
View File
@@ -0,0 +1,19 @@
import type { ExecutionResponse, ExecutionResponseSender } from '@n8n/engine';
import type { InMemoryExecutionResponseChannel } from './in-memory-execution-response-channel';
export class InMemoryExecutionResponseSender implements ExecutionResponseSender {
private stopped = false;
constructor(private readonly channel: InMemoryExecutionResponseChannel) {}
send(response: ExecutionResponse): void {
if (this.stopped) return;
this.channel.publish(response.executionId, JSON.stringify(response));
}
async stop(): Promise<void> {
this.stopped = true;
}
}