mirror of
https://github.com/n8n-io/n8n.git
synced 2026-10-12 02:19:43 +00:00
fix(engine): Preserve host execution mode (#39253)
This commit is contained in:
@@ -41,6 +41,7 @@ describe('workflow_execution table (integration)', () => {
|
||||
graph: { nodes: [], edges: [] },
|
||||
workflow: {},
|
||||
triggerOutputs: [{ foo: 'bar' }],
|
||||
callerContext: { hostMode: 'trigger' },
|
||||
finishedAt: null,
|
||||
});
|
||||
await repo.save(created);
|
||||
@@ -55,6 +56,7 @@ describe('workflow_execution table (integration)', () => {
|
||||
expect(found.status).toBe('running');
|
||||
expect(found.mode).toBe('production');
|
||||
expect(found.triggerOutputs).toEqual([{ foo: 'bar' }]);
|
||||
expect(found.callerContext).toEqual({ hostMode: 'trigger' });
|
||||
expect(found.finishedAt).toBeNull();
|
||||
expect(found.createdAt).toBeInstanceOf(Date);
|
||||
expect(found.updatedAt).toBeInstanceOf(Date);
|
||||
@@ -71,6 +73,7 @@ describe('workflow_execution table (integration)', () => {
|
||||
graph: { nodes: [], edges: [] },
|
||||
workflow: sampleWorkflow,
|
||||
triggerOutputs: [{ foo: 'bar' }],
|
||||
callerContext: { hostMode: 'manual' },
|
||||
finishedAt,
|
||||
});
|
||||
await repo.save(created);
|
||||
@@ -86,6 +89,7 @@ describe('workflow_execution table (integration)', () => {
|
||||
workflowId: 'wf-3',
|
||||
status: 'completed',
|
||||
mode: 'manual',
|
||||
hostMode: 'manual',
|
||||
graph: { nodes: [], edges: [] },
|
||||
workflow: sampleWorkflow,
|
||||
createdAt: expect.any(Date) as Date,
|
||||
@@ -104,6 +108,7 @@ describe('workflow_execution table (integration)', () => {
|
||||
graph: { nodes: [], edges: [] },
|
||||
workflow: sampleWorkflow,
|
||||
triggerOutputs: [{ foo: 'bar' }],
|
||||
callerContext: { hostMode: 'trigger' },
|
||||
finishedAt: null,
|
||||
});
|
||||
await repo.save(created);
|
||||
@@ -118,7 +123,7 @@ describe('workflow_execution table (integration)', () => {
|
||||
mode: 'production',
|
||||
graph: { nodes: [], edges: [] },
|
||||
triggerOutputs: [{ foo: 'bar' }],
|
||||
callerContext: {},
|
||||
callerContext: { hostMode: 'trigger' },
|
||||
});
|
||||
});
|
||||
|
||||
@@ -148,6 +153,7 @@ describe('workflow_execution table (integration)', () => {
|
||||
graph: { nodes: [], edges: [] },
|
||||
workflow: {},
|
||||
triggerOutputs: null,
|
||||
callerContext: { hostMode: 'trigger' },
|
||||
finishedAt: null,
|
||||
}),
|
||||
);
|
||||
@@ -160,6 +166,7 @@ describe('workflow_execution table (integration)', () => {
|
||||
graph: { nodes: [], edges: [] },
|
||||
workflow: {},
|
||||
triggerOutputs: null,
|
||||
callerContext: { hostMode: 'trigger' },
|
||||
finishedAt: new Date(),
|
||||
}),
|
||||
);
|
||||
|
||||
+1
@@ -47,6 +47,7 @@ describe('workflow_step_execution table (integration)', () => {
|
||||
graph: { nodes: [], edges: [] },
|
||||
workflow: {},
|
||||
triggerOutputs: null,
|
||||
callerContext: { hostMode: 'trigger' },
|
||||
finishedAt: null,
|
||||
});
|
||||
await repo.save(execution);
|
||||
|
||||
@@ -43,7 +43,7 @@ export class WorkflowExecution {
|
||||
triggerOutputs!: TriggerOutputs | null;
|
||||
|
||||
/** Caller-supplied, opaque to the engine. See `CallerContext`. */
|
||||
@Column('jsonb', { name: 'caller_context', default: () => "'{}'" })
|
||||
@Column('jsonb', { name: 'caller_context' })
|
||||
callerContext!: CallerContext;
|
||||
|
||||
@CreateDateColumn({ name: 'created_at', type: 'timestamptz', precision: 3 })
|
||||
|
||||
@@ -19,7 +19,6 @@ export class CreateWorkflowExecution1778529600000 implements MigrationInterface
|
||||
{
|
||||
name: 'caller_context',
|
||||
type: 'jsonb',
|
||||
default: "'{}'",
|
||||
comment:
|
||||
'Supplied by the caller at start (user, project, host mode). The engine stores and forwards it to step executors without reading it. Not the engine-internal per-request context.',
|
||||
},
|
||||
|
||||
@@ -15,6 +15,8 @@ import type {
|
||||
/** The execution row, with its steps aggregated into one column. */
|
||||
type ExecutionWithStepsRow = ExecutionView & { steps: StepView[] };
|
||||
|
||||
const HOST_MODE = "execution.caller_context->>'hostMode'";
|
||||
|
||||
/**
|
||||
* The cursor compares `(created_at, id)`, so it only walks that order. A
|
||||
* status-bucket sort puts newer rows behind the top bucket, and the next page
|
||||
@@ -50,6 +52,7 @@ export class TypeOrmExecutionViewStore implements ExecutionViewStore {
|
||||
.addSelect('execution.workflow_id', 'workflowId')
|
||||
.addSelect('execution.status', 'status')
|
||||
.addSelect('execution.mode', 'mode')
|
||||
.addSelect(HOST_MODE, 'hostMode')
|
||||
.addSelect('execution.created_at', 'createdAt')
|
||||
.addSelect('execution.updated_at', 'updatedAt')
|
||||
.addSelect('execution.finished_at', 'finishedAt');
|
||||
@@ -76,7 +79,9 @@ export class TypeOrmExecutionViewStore implements ExecutionViewStore {
|
||||
qb.andWhere('execution.workflow_id = ANY(:workflowIds)', { workflowIds: query.workflowIds });
|
||||
}
|
||||
if (query.status) qb.andWhere('execution.status = ANY(:statuses)', { statuses: query.status });
|
||||
if (query.mode) qb.andWhere('execution.mode = :mode', { mode: query.mode });
|
||||
if (query.hostMode) {
|
||||
qb.andWhere(`${HOST_MODE} = :hostMode`, { hostMode: query.hostMode });
|
||||
}
|
||||
if (query.createdAfter)
|
||||
qb.andWhere('execution.created_at >= :createdAfter', { createdAfter: query.createdAfter });
|
||||
if (query.createdBefore)
|
||||
@@ -133,6 +138,7 @@ export class TypeOrmExecutionViewStore implements ExecutionViewStore {
|
||||
.addSelect('execution.workflow_id', 'workflowId')
|
||||
.addSelect('execution.status', 'status')
|
||||
.addSelect('execution.mode', 'mode')
|
||||
.addSelect(HOST_MODE, 'hostMode')
|
||||
.addSelect('execution.graph', 'graph')
|
||||
.addSelect('execution.workflow', 'workflow')
|
||||
.addSelect('execution.created_at', 'createdAt')
|
||||
|
||||
@@ -63,7 +63,7 @@ function record(graph: WorkflowGraph, overrides: Partial<ExecutionRecord> = {}):
|
||||
graph,
|
||||
workflow: {},
|
||||
triggerOutputs: null,
|
||||
callerContext: {},
|
||||
callerContext: { hostMode: 'trigger' },
|
||||
...overrides,
|
||||
};
|
||||
}
|
||||
@@ -220,10 +220,12 @@ describe('ExecutionStartHandler lifecycle events', () => {
|
||||
edges: [],
|
||||
};
|
||||
|
||||
it('announces execution:started once it wins the claim', async () => {
|
||||
it('announces execution:started with the engine and host modes', async () => {
|
||||
const lifecycleEventPublisher = makeLifecycleEventPublisher();
|
||||
const executionStore = makeExecutionStore({
|
||||
loadExecution: vi.fn().mockResolvedValue(record(graph, { mode: 'manual' })),
|
||||
loadExecution: vi
|
||||
.fn()
|
||||
.mockResolvedValue(record(graph, { callerContext: { hostMode: 'webhook' } })),
|
||||
});
|
||||
const handler = makeHandler(
|
||||
executionStore,
|
||||
@@ -238,7 +240,8 @@ describe('ExecutionStartHandler lifecycle events', () => {
|
||||
type: 'execution:started',
|
||||
executionId: 'exec-1',
|
||||
workflowId: 'wf-1',
|
||||
mode: 'manual',
|
||||
mode: 'production',
|
||||
hostMode: 'webhook',
|
||||
at: expect.any(String) as string,
|
||||
});
|
||||
});
|
||||
|
||||
@@ -102,7 +102,7 @@ describe('execution start (integration)', () => {
|
||||
workflow: {},
|
||||
triggerOutputs: [[{ json: { hello: 'world' } }]],
|
||||
executionId: generateId(),
|
||||
callerContext: {},
|
||||
callerContext: { hostMode: 'trigger' },
|
||||
});
|
||||
await ready;
|
||||
|
||||
@@ -149,7 +149,7 @@ describe('execution start (integration)', () => {
|
||||
graph,
|
||||
workflow: {},
|
||||
triggerOutputs: null,
|
||||
callerContext: {},
|
||||
callerContext: { hostMode: 'trigger' },
|
||||
});
|
||||
|
||||
// Delivered twice, both awaited — the CAS is what makes the second a no-op.
|
||||
|
||||
@@ -48,7 +48,7 @@ describe('StartExecutionService', () => {
|
||||
workflow: sampleWorkflow,
|
||||
triggerOutputs: [[{ json: { hello: 'world' } }]],
|
||||
executionId: 'exec-id-1',
|
||||
callerContext: {},
|
||||
callerContext: { hostMode: 'trigger' },
|
||||
});
|
||||
|
||||
expect(result.executionId).toBe('exec-id-1');
|
||||
@@ -61,7 +61,7 @@ describe('StartExecutionService', () => {
|
||||
graph: sampleGraph,
|
||||
workflow: sampleWorkflow,
|
||||
triggerOutputs: [[{ json: { hello: 'world' } }]],
|
||||
callerContext: {},
|
||||
callerContext: { hostMode: 'trigger' },
|
||||
});
|
||||
expect(queue.publish).toHaveBeenCalledWith({
|
||||
type: 'execution:enqueued',
|
||||
@@ -101,7 +101,7 @@ describe('StartExecutionService', () => {
|
||||
graph: sampleGraph,
|
||||
workflow: sampleWorkflow,
|
||||
executionId: 'exec-id-1',
|
||||
callerContext: {},
|
||||
callerContext: { hostMode: 'trigger' },
|
||||
});
|
||||
|
||||
expect(store.createExecution).toHaveBeenCalledWith(
|
||||
@@ -121,7 +121,7 @@ describe('StartExecutionService', () => {
|
||||
graph: sampleGraph,
|
||||
workflow: sampleWorkflow,
|
||||
executionId: 'exec-id-1',
|
||||
callerContext: {},
|
||||
callerContext: { hostMode: 'trigger' },
|
||||
});
|
||||
|
||||
expect(validateGraph).toHaveBeenCalledExactlyOnceWith(sampleGraph);
|
||||
@@ -145,7 +145,7 @@ describe('StartExecutionService', () => {
|
||||
graph: sampleGraph,
|
||||
workflow: sampleWorkflow,
|
||||
executionId: 'exec-id-1',
|
||||
callerContext: {},
|
||||
callerContext: { hostMode: 'trigger' },
|
||||
}),
|
||||
).rejects.toBe(rejection);
|
||||
|
||||
@@ -167,7 +167,7 @@ describe('StartExecutionService', () => {
|
||||
graph: sampleGraph,
|
||||
workflow: sampleWorkflow,
|
||||
executionId: 'exec-id-1',
|
||||
callerContext: {},
|
||||
callerContext: { hostMode: 'trigger' },
|
||||
}),
|
||||
).rejects.toBeInstanceOf(AdmittanceRejectedError);
|
||||
|
||||
|
||||
@@ -110,7 +110,7 @@ describe('step execution (integration)', () => {
|
||||
workflow: {},
|
||||
triggerOutputs,
|
||||
executionId: generateId(),
|
||||
callerContext: {},
|
||||
callerContext: { hostMode: 'trigger' },
|
||||
})
|
||||
.expect(201);
|
||||
const { executionId } = response.body as StartExecutionResult;
|
||||
@@ -154,6 +154,7 @@ describe('step execution (integration)', () => {
|
||||
executionId,
|
||||
workflowId: 'wf-1',
|
||||
mode: 'production',
|
||||
hostMode: 'trigger',
|
||||
at: expect.any(String) as string,
|
||||
});
|
||||
// The ids are the ones a consumer would re-query the data plane with.
|
||||
@@ -184,7 +185,7 @@ describe('step execution (integration)', () => {
|
||||
workflowId: 'wf-1',
|
||||
mode: 'production',
|
||||
iteration: 0,
|
||||
callerContext: {},
|
||||
callerContext: { hostMode: 'trigger' },
|
||||
});
|
||||
});
|
||||
|
||||
@@ -401,7 +402,7 @@ describe('step execution (integration)', () => {
|
||||
graph,
|
||||
workflow: {},
|
||||
triggerOutputs: null,
|
||||
callerContext: {},
|
||||
callerContext: { hostMode: 'trigger' },
|
||||
});
|
||||
const created = await stepStore.createSteps(executionId, [
|
||||
// completed steps always carry outputs, as the start handler writes them
|
||||
|
||||
@@ -68,7 +68,7 @@ function makeExecutionStore(overrides: Partial<ExecutionRecord> = {}): Execution
|
||||
graph,
|
||||
workflow: {},
|
||||
triggerOutputs: null,
|
||||
callerContext: {},
|
||||
callerContext: { hostMode: 'trigger' },
|
||||
...overrides,
|
||||
};
|
||||
return {
|
||||
@@ -153,7 +153,7 @@ describe('StepReadyHandler', () => {
|
||||
workflowId: 'wf-1',
|
||||
mode: 'production',
|
||||
iteration: 0,
|
||||
callerContext: {},
|
||||
callerContext: { hostMode: 'trigger' },
|
||||
},
|
||||
});
|
||||
expect(stepStore.completeStep).toHaveBeenCalledWith('step-a', [[{ json: { ok: true } }]]);
|
||||
|
||||
@@ -57,7 +57,7 @@ function makeExecutionStore(
|
||||
graph,
|
||||
workflow: {},
|
||||
triggerOutputs: null,
|
||||
callerContext: {},
|
||||
callerContext: { hostMode: 'trigger' },
|
||||
...overrides,
|
||||
};
|
||||
return {
|
||||
|
||||
@@ -40,6 +40,7 @@ export class ExecutionStartHandler {
|
||||
executionId: execution.id,
|
||||
workflowId: execution.workflowId,
|
||||
mode: execution.mode,
|
||||
hostMode: execution.callerContext.hostMode,
|
||||
at: new Date().toISOString(),
|
||||
});
|
||||
|
||||
|
||||
@@ -18,6 +18,8 @@ export interface ExecutionListItemView {
|
||||
workflowId: string;
|
||||
status: ExecutionStatus;
|
||||
mode: ExecutionMode;
|
||||
/** The host's finer execution mode. */
|
||||
hostMode: string;
|
||||
createdAt: Date;
|
||||
updatedAt: Date;
|
||||
finishedAt: Date | null;
|
||||
@@ -26,7 +28,7 @@ export interface ExecutionListItemView {
|
||||
export interface ExecutionListQuery {
|
||||
workflowIds: readonly string[] | 'all';
|
||||
status?: ExecutionStatus[];
|
||||
mode?: string;
|
||||
hostMode?: string;
|
||||
createdAfter?: string;
|
||||
createdBefore?: string;
|
||||
before?: { createdAt: string; id: string };
|
||||
@@ -47,6 +49,8 @@ export interface ExecutionView {
|
||||
workflowId: string;
|
||||
status: ExecutionStatus;
|
||||
mode: ExecutionMode;
|
||||
/** The host's finer execution mode. */
|
||||
hostMode: string;
|
||||
/** The graph captured at start, immutable for the execution's lifetime. */
|
||||
graph: WorkflowGraph;
|
||||
/** The workflow captured at start, alongside the graph and just as immutable. */
|
||||
|
||||
@@ -25,7 +25,7 @@ export interface CallerContext {
|
||||
* The host's own execution mode, which is finer than `ExecutionMode`. Opaque
|
||||
* to the engine; a v1 host stores its `WorkflowExecuteMode` here.
|
||||
*/
|
||||
hostMode?: string;
|
||||
hostMode: string;
|
||||
}
|
||||
|
||||
/**
|
||||
|
||||
@@ -21,7 +21,8 @@ const stepEvent = {
|
||||
};
|
||||
|
||||
const every: LifecycleEvent[] = [
|
||||
{ type: 'execution:started', ...executionEvent, mode: 'manual' },
|
||||
{ type: 'execution:started', ...executionEvent, mode: 'production', hostMode: 'webhook' },
|
||||
{ type: 'execution:started', ...executionEvent, mode: 'manual', hostMode: 'manual' },
|
||||
{ type: 'execution:completed', ...executionEvent },
|
||||
{ type: 'execution:failed', ...executionEvent },
|
||||
{ type: 'step:started', ...stepEvent },
|
||||
@@ -30,14 +31,18 @@ const every: LifecycleEvent[] = [
|
||||
];
|
||||
|
||||
describe('lifecycleEventSchema', () => {
|
||||
// The union cannot drift, but a field the schema restates in its own
|
||||
// vocabulary can. These pin those, in both directions.
|
||||
it('keeps the mode enum in step with ExecutionMode', () => {
|
||||
it('keeps the engine mode in step with ExecutionMode', () => {
|
||||
expectTypeOf<
|
||||
Extract<LifecycleEvent, { type: 'execution:started' }>['mode']
|
||||
>().toEqualTypeOf<ExecutionMode>();
|
||||
});
|
||||
|
||||
it('carries the host execution mode separately', () => {
|
||||
expectTypeOf<
|
||||
Extract<LifecycleEvent, { type: 'execution:started' }>['hostMode']
|
||||
>().toEqualTypeOf<string>();
|
||||
});
|
||||
|
||||
it('keeps the completed outputs in step with StepSlots', () => {
|
||||
expectTypeOf<
|
||||
Extract<LifecycleEvent, { type: 'step:completed' }>['outputs']
|
||||
|
||||
@@ -31,6 +31,8 @@ export const lifecycleEventSchema = z.discriminatedUnion('type', [
|
||||
...executionFields,
|
||||
// Must match `ExecutionMode`.
|
||||
mode: z.enum(['production', 'manual']),
|
||||
/** The host's finer execution mode. */
|
||||
hostMode: z.string().min(1),
|
||||
}),
|
||||
z.object({ type: z.literal('execution:completed'), ...executionFields }),
|
||||
z.object({ type: z.literal('execution:failed'), ...executionFields }),
|
||||
|
||||
@@ -59,6 +59,7 @@ describe('POST /api/workflow-executions/search (integration)', () => {
|
||||
{},
|
||||
{ workflowIds: [] },
|
||||
{ workflowIds: [''] },
|
||||
{ workflowIds: 'all', mode: 'webhook' },
|
||||
{ workflowIds: 'all', tenantId: 'tenant' },
|
||||
{ workflowIds: 'all', unknown: true },
|
||||
{ workflowIds: 'all', limit: 101 },
|
||||
@@ -93,7 +94,16 @@ describe('POST /api/workflow-executions/search (integration)', () => {
|
||||
expect(result.nextCursor).toBeNull();
|
||||
expect(result.items.map((item) => item.id)).toEqual([id]);
|
||||
expect(Object.keys(result.items[0]).sort()).toEqual(
|
||||
['id', 'workflowId', 'status', 'mode', 'createdAt', 'updatedAt', 'finishedAt'].sort(),
|
||||
[
|
||||
'id',
|
||||
'workflowId',
|
||||
'status',
|
||||
'mode',
|
||||
'hostMode',
|
||||
'createdAt',
|
||||
'updatedAt',
|
||||
'finishedAt',
|
||||
].sort(),
|
||||
);
|
||||
const { executionViewStore } = createStores(dataSource);
|
||||
const rows = await executionViewStore.listExecutionViews({
|
||||
@@ -114,13 +124,27 @@ describe('POST /api/workflow-executions/search (integration)', () => {
|
||||
it('compares modes exactly and applies status filters', async () => {
|
||||
const workflowId = generateId();
|
||||
await start(workflowId);
|
||||
const production = await search({
|
||||
await request(url)
|
||||
.post('/api/workflow-executions')
|
||||
.set(authHeader())
|
||||
.send(startBody({ workflowId, callerContext: { hostMode: 'webhook' } }))
|
||||
.expect(201);
|
||||
|
||||
const trigger = await search({
|
||||
workflowIds: [workflowId],
|
||||
mode: 'production',
|
||||
hostMode: 'trigger',
|
||||
status: ['queued'],
|
||||
}).expect(200);
|
||||
expect((production.body as SearchExecutionsResponse).items).toHaveLength(1);
|
||||
for (const filter of [{ mode: 'webhook' }, { mode: 'manual' }, { status: ['completed'] }]) {
|
||||
expect((trigger.body as SearchExecutionsResponse).items).toHaveLength(1);
|
||||
|
||||
const webhook = await search({ workflowIds: [workflowId], hostMode: 'webhook' }).expect(200);
|
||||
expect((webhook.body as SearchExecutionsResponse).items).toHaveLength(1);
|
||||
expect((webhook.body as SearchExecutionsResponse).items[0]).toMatchObject({
|
||||
mode: 'production',
|
||||
hostMode: 'webhook',
|
||||
});
|
||||
|
||||
for (const filter of [{ hostMode: 'manual' }, { status: ['completed'] }]) {
|
||||
const response = await search({
|
||||
workflowIds: [workflowId],
|
||||
...filter,
|
||||
@@ -172,7 +196,7 @@ const startBody = (overrides: Record<string, unknown> = {}) => ({
|
||||
graph: sampleGraph,
|
||||
workflow: sampleWorkflow,
|
||||
executionId: generateId(),
|
||||
callerContext: {},
|
||||
callerContext: { hostMode: 'trigger' },
|
||||
...overrides,
|
||||
});
|
||||
|
||||
@@ -242,13 +266,16 @@ describe('POST /api/workflow-executions (integration)', () => {
|
||||
it.each(['not-a-uuid', '9f1b7d0e-2c4a-4f8b-9d3e-6a5c1b2d3e4f', undefined])(
|
||||
'rejects the execution id %p with 400',
|
||||
async (executionId) => {
|
||||
const response = await request(url).post('/api/workflow-executions').set(authHeader()).send({
|
||||
workflowId: 'wf-1',
|
||||
graph: sampleGraph,
|
||||
workflow: sampleWorkflow,
|
||||
executionId,
|
||||
callerContext: {},
|
||||
});
|
||||
const response = await request(url)
|
||||
.post('/api/workflow-executions')
|
||||
.set(authHeader())
|
||||
.send({
|
||||
workflowId: 'wf-1',
|
||||
graph: sampleGraph,
|
||||
workflow: sampleWorkflow,
|
||||
executionId,
|
||||
callerContext: { hostMode: 'trigger' },
|
||||
});
|
||||
|
||||
expect(response.status).toBe(400);
|
||||
expect((response.body as { error: string }).error).toBe('invalid_request');
|
||||
@@ -281,6 +308,16 @@ describe('POST /api/workflow-executions (integration)', () => {
|
||||
expect((response.body as { error: string }).error).toBe('invalid_request');
|
||||
});
|
||||
|
||||
it('rejects a caller context without a host mode with 400', async () => {
|
||||
const response = await request(url)
|
||||
.post('/api/workflow-executions')
|
||||
.set(authHeader())
|
||||
.send(startBody({ callerContext: { userId: 'user-1' } }));
|
||||
|
||||
expect(response.status).toBe(400);
|
||||
expect((response.body as { error: string }).error).toBe('invalid_request');
|
||||
});
|
||||
|
||||
it('rejects a caller context with a key it does not know with 400', async () => {
|
||||
const response = await request(url)
|
||||
.post('/api/workflow-executions')
|
||||
@@ -433,11 +470,16 @@ describe('GET /api/workflow-executions/:id (integration)', () => {
|
||||
const response = await request(url)
|
||||
.post('/api/workflow-executions')
|
||||
.set(authHeader())
|
||||
.send(startBody({ triggerOutputs: [[{ json: { hello: 'world' } }]] }));
|
||||
.send(
|
||||
startBody({
|
||||
triggerOutputs: [[{ json: { hello: 'world' } }]],
|
||||
callerContext: { hostMode: 'webhook' },
|
||||
}),
|
||||
);
|
||||
return (response.body as { executionId: string }).executionId;
|
||||
}
|
||||
|
||||
it('returns the persisted status, mode, workflow id, graph, workflow and ISO timestamps', async () => {
|
||||
it('returns the host mode with the persisted execution data', async () => {
|
||||
const executionId = await createExecution();
|
||||
|
||||
const response = await request(url)
|
||||
@@ -450,6 +492,7 @@ describe('GET /api/workflow-executions/:id (integration)', () => {
|
||||
workflowId: 'wf-1',
|
||||
status: 'queued',
|
||||
mode: 'production',
|
||||
hostMode: 'webhook',
|
||||
graph: sampleGraph,
|
||||
workflow: sampleWorkflow,
|
||||
finishedAt: null,
|
||||
|
||||
@@ -36,6 +36,8 @@ export interface ExecutionSnapshot {
|
||||
workflowId: string;
|
||||
status: ExecutionStatus;
|
||||
mode: ExecutionMode;
|
||||
/** The host's finer execution mode. */
|
||||
hostMode: string;
|
||||
/** The graph captured at start, immutable for the execution's lifetime. */
|
||||
graph: WorkflowGraph;
|
||||
/**
|
||||
|
||||
@@ -28,7 +28,7 @@ const SearchExecutionsBody = z
|
||||
.object({
|
||||
workflowIds: z.union([z.literal('all'), z.array(z.string().min(1)).min(1).max(10_000)]),
|
||||
status: z.array(ExecutionStatusSchema).min(1).optional(),
|
||||
mode: z.string().min(1).max(32).optional(),
|
||||
hostMode: z.string().min(1).max(32).optional(),
|
||||
createdAfter: datetimeStringWithOffset().optional(),
|
||||
createdBefore: datetimeStringWithOffset().optional(),
|
||||
before: z
|
||||
@@ -80,6 +80,7 @@ function toExecutionSnapshot(record: ExecutionView | ExecutionWithStepsView): Ex
|
||||
workflowId: record.workflowId,
|
||||
status: record.status,
|
||||
mode: record.mode,
|
||||
hostMode: record.hostMode,
|
||||
graph: record.graph,
|
||||
workflow: record.workflow,
|
||||
createdAt: record.createdAt.toISOString(),
|
||||
|
||||
@@ -56,7 +56,7 @@ const StartExecutionBody = z.object({
|
||||
.object({
|
||||
userId: z.string().min(1).optional(),
|
||||
projectId: z.string().min(1).optional(),
|
||||
hostMode: z.string().min(1).optional(),
|
||||
hostMode: z.string().min(1),
|
||||
})
|
||||
.strict(),
|
||||
/** The caller mints the id. v7 only, so ids stay time-ordered. */
|
||||
|
||||
+1
-1
@@ -21,7 +21,7 @@ const context = {
|
||||
workflowId: 'wf-1',
|
||||
mode: 'production',
|
||||
iteration: 0,
|
||||
callerContext: {},
|
||||
callerContext: { hostMode: 'trigger' },
|
||||
} as const;
|
||||
|
||||
describe('createEngineStepDataLoader', () => {
|
||||
|
||||
@@ -214,7 +214,7 @@ describe('toV1ExecuteMode', () => {
|
||||
workflowId: 'wf-1',
|
||||
mode: 'production',
|
||||
iteration: 0,
|
||||
callerContext: {},
|
||||
callerContext: { hostMode: 'trigger' },
|
||||
...overrides,
|
||||
});
|
||||
|
||||
@@ -222,10 +222,6 @@ describe('toV1ExecuteMode', () => {
|
||||
expect(toV1ExecuteMode(context({ callerContext: { hostMode: 'webhook' } }))).toBe('webhook');
|
||||
});
|
||||
|
||||
it('throws when the caller context has no host mode', () => {
|
||||
expect(() => toV1ExecuteMode(context({}))).toThrow('no v1 execution mode');
|
||||
});
|
||||
|
||||
it('throws when the host mode is not one v1 knows', () => {
|
||||
expect(() => toV1ExecuteMode(context({ callerContext: { hostMode: 'production' } }))).toThrow(
|
||||
'unknown v1 execution mode',
|
||||
|
||||
@@ -41,6 +41,7 @@ const snapshot = (overrides: Partial<ExecutionSnapshot> = {}): ExecutionSnapshot
|
||||
workflowId: WORKFLOW_ID,
|
||||
status: 'completed',
|
||||
mode: 'manual',
|
||||
hostMode: 'manual',
|
||||
graph: { nodes: [{ id: 'trigger-id', name: 'Trigger', type: 'trigger' }], edges: [] },
|
||||
workflow: workflowDocument(),
|
||||
createdAt: '2026-08-25T10:00:00.000Z',
|
||||
@@ -65,7 +66,9 @@ describe('EngineV2ExecutionReader', () => {
|
||||
'forwards the mode %s without conversion',
|
||||
async (mode) => {
|
||||
await reader.findMany({ ...query, mode }, ['wf-1']);
|
||||
expect(dataPlane.searchExecutions).toHaveBeenCalledWith(expect.objectContaining({ mode }));
|
||||
expect(dataPlane.searchExecutions).toHaveBeenCalledWith(
|
||||
expect.objectContaining({ hostMode: mode }),
|
||||
);
|
||||
},
|
||||
);
|
||||
|
||||
@@ -235,13 +238,18 @@ describe('EngineV2ExecutionReader', () => {
|
||||
expect(result?.finished).toBe(finished);
|
||||
});
|
||||
|
||||
it('should map a production run onto the v1 trigger mode', async () => {
|
||||
dataPlane.getExecution.mockResolvedValue(snapshot({ mode: 'production' }));
|
||||
it.each(['manual', 'webhook', 'trigger'] as const)(
|
||||
'should preserve the %s execution mode',
|
||||
async (mode) => {
|
||||
dataPlane.getExecution.mockResolvedValue(
|
||||
snapshot({ mode: mode === 'manual' ? 'manual' : 'production', hostMode: mode }),
|
||||
);
|
||||
|
||||
const result = await reader.findOne(EXECUTION_ID, [WORKFLOW_ID]);
|
||||
const result = await reader.findOne(EXECUTION_ID, [WORKFLOW_ID]);
|
||||
|
||||
expect(result?.mode).toBe('trigger');
|
||||
});
|
||||
expect(result?.mode).toBe(mode);
|
||||
},
|
||||
);
|
||||
|
||||
it('should leave stoppedAt unset while the execution is unfinished', async () => {
|
||||
dataPlane.getExecution.mockResolvedValue(snapshot({ status: 'running', finishedAt: null }));
|
||||
|
||||
@@ -59,6 +59,7 @@ describe('ExecutionListV2Service', () => {
|
||||
workflowId: 'wf-a',
|
||||
status,
|
||||
mode: 'manual',
|
||||
hostMode: 'manual',
|
||||
createdAt: time,
|
||||
updatedAt: time,
|
||||
finishedAt: time,
|
||||
|
||||
@@ -113,7 +113,7 @@ export class EngineV2ExecutionReader {
|
||||
return {
|
||||
workflowIds: scope,
|
||||
status,
|
||||
mode: query.mode,
|
||||
hostMode: query.mode,
|
||||
createdAfter: query.startedAfter ? new Date(query.startedAfter).toISOString() : undefined,
|
||||
createdBefore: query.startedBefore ? new Date(query.startedBefore).toISOString() : undefined,
|
||||
before: query.before ? { createdAt: query.before.timestamp, id: query.before.id } : undefined,
|
||||
@@ -128,7 +128,7 @@ export class EngineV2ExecutionReader {
|
||||
id: item.id,
|
||||
workflowId: item.workflowId,
|
||||
status: toV1Status(item.status),
|
||||
mode: toV1Mode(item.mode),
|
||||
mode: toV1Mode(item.hostMode),
|
||||
finished: item.status === 'completed',
|
||||
createdAt: new Date(item.createdAt),
|
||||
startedAt: new Date(item.createdAt),
|
||||
@@ -175,7 +175,7 @@ export class EngineV2ExecutionReader {
|
||||
return {
|
||||
id: snapshot.id,
|
||||
workflowId: snapshot.workflowId,
|
||||
mode: toV1Mode(snapshot.mode),
|
||||
mode: toV1Mode(snapshot.hostMode),
|
||||
status: toV1Status(snapshot.status),
|
||||
finished: snapshot.status === 'completed',
|
||||
createdAt: startedAt,
|
||||
|
||||
@@ -1,5 +1,10 @@
|
||||
import type { ExecutionMode, ExecutionStatus } from '@n8n/engine';
|
||||
import type { ExecutionStatus as ExecutionStatusV1, WorkflowExecuteMode } from 'n8n-workflow';
|
||||
import type { ExecutionStatus } from '@n8n/engine';
|
||||
import assert from 'node:assert';
|
||||
import {
|
||||
WorkflowExecuteModeList,
|
||||
type ExecutionStatus as ExecutionStatusV1,
|
||||
type WorkflowExecuteMode,
|
||||
} from 'n8n-workflow';
|
||||
|
||||
/** A status added later reads as `unknown` rather than being guessed at. */
|
||||
const V1_STATUS_BY_V2_STATUS = new Map<ExecutionStatus, ExecutionStatusV1>([
|
||||
@@ -10,18 +15,18 @@ const V1_STATUS_BY_V2_STATUS = new Map<ExecutionStatus, ExecutionStatusV1>([
|
||||
['cancelled', 'canceled'],
|
||||
]);
|
||||
|
||||
/** Anything not manual is a production run. */
|
||||
const V1_MODE_BY_V2_MODE = new Map<ExecutionMode, WorkflowExecuteMode>([
|
||||
['manual', 'manual'],
|
||||
['production', 'trigger'],
|
||||
]);
|
||||
const V1_MODE_BY_REPORTED_MODE = new Map<string, WorkflowExecuteMode>(
|
||||
WorkflowExecuteModeList.map((mode) => [mode, mode]),
|
||||
);
|
||||
|
||||
export function toV1Status(status: ExecutionStatus): ExecutionStatusV1 {
|
||||
return V1_STATUS_BY_V2_STATUS.get(status) ?? 'unknown';
|
||||
}
|
||||
|
||||
export function toV1Mode(mode: ExecutionMode): WorkflowExecuteMode {
|
||||
return V1_MODE_BY_V2_MODE.get(mode) ?? 'trigger';
|
||||
export function toV1Mode(hostMode: string): WorkflowExecuteMode {
|
||||
const v1Mode = V1_MODE_BY_REPORTED_MODE.get(hostMode);
|
||||
assert(v1Mode, `Unknown v1 mode ${hostMode}`);
|
||||
return v1Mode;
|
||||
}
|
||||
|
||||
/**
|
||||
|
||||
@@ -23,7 +23,7 @@ describe('EngineDataPlaneClient', () => {
|
||||
graph: { nodes: [], edges: [] },
|
||||
workflow: {},
|
||||
executionId: EXECUTION_ID,
|
||||
callerContext: {},
|
||||
callerContext: { hostMode: 'trigger' },
|
||||
};
|
||||
|
||||
let http: HttpRequestClient;
|
||||
@@ -63,7 +63,7 @@ describe('EngineDataPlaneClient', () => {
|
||||
});
|
||||
|
||||
it('posts search filters without following redirects', async () => {
|
||||
const body = { workflowIds: ['wf'], mode: 'webhook', includeTotal: true, limit: 20 };
|
||||
const body = { workflowIds: ['wf'], hostMode: 'webhook', includeTotal: true, limit: 20 };
|
||||
const result = { items: [], hasMore: false, total: 0 };
|
||||
respondWith(200, result);
|
||||
await expect(client.searchExecutions(body)).resolves.toEqual(result);
|
||||
|
||||
@@ -26,6 +26,7 @@ const executionStarted: LifecycleEvent = {
|
||||
executionId: EXECUTION_ID,
|
||||
workflowId: WORKFLOW_ID,
|
||||
mode: 'manual',
|
||||
hostMode: 'manual',
|
||||
at: '2026-08-25T10:00:00.000Z',
|
||||
};
|
||||
|
||||
|
||||
@@ -14,6 +14,7 @@ const events: LifecycleEvent[] = [
|
||||
executionId: 'exec-1',
|
||||
workflowId: 'wf-1',
|
||||
mode: 'manual',
|
||||
hostMode: 'manual',
|
||||
at: '2026-08-24T10:00:00.000Z',
|
||||
},
|
||||
{
|
||||
|
||||
@@ -51,7 +51,7 @@ describe('EngineV2Module', () => {
|
||||
graph: { nodes: [], edges: [] },
|
||||
workflow: {},
|
||||
executionId: '01a038ae-c4a8-7799-8a3e-e3c2ca055cfa',
|
||||
callerContext: {},
|
||||
callerContext: { hostMode: 'trigger' },
|
||||
};
|
||||
await expect(proxy.startExecution(request)).rejects.toThrow('N8N_ENABLED_MODULES');
|
||||
|
||||
|
||||
@@ -15,7 +15,7 @@ describe('EngineDataPlaneProxyService', () => {
|
||||
graph: { nodes: [], edges: [] },
|
||||
workflow: {},
|
||||
executionId,
|
||||
callerContext: {},
|
||||
callerContext: { hostMode: 'trigger' },
|
||||
};
|
||||
|
||||
let proxy: EngineDataPlaneProxyService;
|
||||
|
||||
@@ -111,8 +111,8 @@ export class EngineV2Dispatcher {
|
||||
// workflow that ran even after the live one is edited.
|
||||
workflow: toWorkflowDocument(workflowData),
|
||||
triggerOutputs: this.toTriggerOutputs(trigger.outputs, toStepOutputs),
|
||||
// Only manual, webhook and trigger route here, so anything that is not a
|
||||
// manual run is a production run.
|
||||
// The engine keeps only a coarse manual/production distinction. The exact
|
||||
// host mode is carried in callerContext for reads and lifecycle events.
|
||||
mode: data.executionMode === 'manual' ? 'manual' : 'production',
|
||||
// The step executor needs the v1 mode and the caller to resolve credentials.
|
||||
callerContext: {
|
||||
|
||||
@@ -283,6 +283,7 @@ describe('GET /executions/:id', () => {
|
||||
workflowId,
|
||||
status: 'completed',
|
||||
mode: 'manual',
|
||||
hostMode: 'manual',
|
||||
graph: { nodes: [{ id: 'trigger-id', name: 'Trigger', type: 'trigger' }], edges: [] },
|
||||
workflow: ranWorkflow(workflowId),
|
||||
createdAt: '2026-08-25T10:00:00.000Z',
|
||||
|
||||
@@ -47,6 +47,7 @@ const row = (workflowId: string): ExecutionListItem => ({
|
||||
workflowId,
|
||||
status: 'completed',
|
||||
mode: 'manual',
|
||||
hostMode: 'manual',
|
||||
createdAt: time,
|
||||
updatedAt: time,
|
||||
finishedAt: time,
|
||||
|
||||
Reference in New Issue
Block a user