mirror of
https://github.com/n8n-io/n8n.git
synced 2026-10-24 02:09:26 +08:00
test(core): Cover leased trigger policy, lease kind and holder (no-changelog)
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Emilia
co-authored by
Claude Opus 5.5
Claude Opus 5.5
parent
0977ae57ed
commit
9e8a3be954
@@ -0,0 +1,42 @@
|
||||
import type { GlobalConfig } from '@n8n/config';
|
||||
import { mock } from 'vitest-mock-extended';
|
||||
|
||||
import type { ClusterLeases } from '@/scheduling/leases/cluster-leases';
|
||||
|
||||
import { LeasedTriggerPolicy } from '../leased-trigger-policy';
|
||||
|
||||
const policy = (nodeTypes: string, leasesAvailable = true) =>
|
||||
new LeasedTriggerPolicy(
|
||||
mock<GlobalConfig>({ workflows: { leasedTriggerNodeTypes: nodeTypes } }),
|
||||
mock<ClusterLeases>({ isAvailable: () => leasesAvailable }),
|
||||
);
|
||||
|
||||
describe('LeasedTriggerPolicy', () => {
|
||||
it('leases the listed node types', () => {
|
||||
const leased = policy(' n8n-nodes-base.kafkaTrigger , n8n-nodes-base.mqttTrigger,');
|
||||
|
||||
expect(leased.isEnabled()).toBe(true);
|
||||
expect(leased.isLeased({ type: 'n8n-nodes-base.kafkaTrigger' })).toBe(true);
|
||||
expect(leased.isLeased({ type: 'n8n-nodes-base.mqttTrigger' })).toBe(true);
|
||||
expect(leased.isLeased({ type: 'n8n-nodes-base.rabbitmqTrigger' })).toBe(false);
|
||||
});
|
||||
|
||||
it('leases nothing when no node type is listed', () => {
|
||||
expect(policy('').isEnabled()).toBe(false);
|
||||
});
|
||||
|
||||
it('leases nothing when cluster leases are unavailable', () => {
|
||||
const leased = policy('n8n-nodes-base.kafkaTrigger', false);
|
||||
|
||||
expect(leased.isEnabled()).toBe(false);
|
||||
expect(leased.isLeased({ type: 'n8n-nodes-base.kafkaTrigger' })).toBe(false);
|
||||
});
|
||||
|
||||
it('keeps only the nodes the leader activates', () => {
|
||||
const nodes = [{ type: 'n8n-nodes-base.kafkaTrigger' }, { type: 'n8n-nodes-base.webhook' }];
|
||||
|
||||
expect(policy('n8n-nodes-base.kafkaTrigger').withoutLeased(nodes)).toEqual([
|
||||
{ type: 'n8n-nodes-base.webhook' },
|
||||
]);
|
||||
});
|
||||
});
|
||||
+149
@@ -0,0 +1,149 @@
|
||||
import type { WorkflowEntity, WorkflowRepository } from '@n8n/db';
|
||||
import type { Lease } from '@n8n/leases';
|
||||
import { LeaseEndedError } from '@n8n/leases';
|
||||
import type { IWorkflowBase, IWorkflowExecuteAdditionalData } from 'n8n-workflow';
|
||||
import { WorkflowExpression } from 'n8n-workflow';
|
||||
import { mock } from 'vitest-mock-extended';
|
||||
|
||||
import * as WorkflowExecuteAdditionalData from '@/workflow-execute-additional-data';
|
||||
import type {
|
||||
NonWebhookTriggerRegistrar,
|
||||
NonWebhookTriggerRegistrationContext,
|
||||
PreparedNonWebhookTriggerRegistration,
|
||||
} from '@/workflows/triggers/non-webhook-trigger-registrar';
|
||||
import type { TriggerExecutionContextFactory } from '@/workflows/triggers/trigger-execution-context.factory';
|
||||
import type { WorkflowStaticDataService } from '@/workflows/workflow-static-data.service';
|
||||
|
||||
import { createNodeTypes, logger, node } from '../../__tests__/trigger-test-utils';
|
||||
import type { LeasedTriggerRegistry } from '../leased-trigger-registry';
|
||||
import { WorkflowTriggerLeaseHolder } from '../workflow-trigger-lease-holder';
|
||||
|
||||
const trigger = node('kafka', 'trigger');
|
||||
|
||||
const setup = ({ published = true } = {}) => {
|
||||
vi.spyOn(WorkflowExpression.prototype, 'acquireIsolate').mockResolvedValue(true);
|
||||
vi.spyOn(WorkflowExpression.prototype, 'releaseIsolate').mockResolvedValue();
|
||||
vi.spyOn(WorkflowExecuteAdditionalData, 'getBase').mockResolvedValue(
|
||||
mock<IWorkflowExecuteAdditionalData>(),
|
||||
);
|
||||
const registry = mock<LeasedTriggerRegistry>();
|
||||
const workflows = mock<WorkflowRepository>();
|
||||
workflows.findOneBy.mockResolvedValue(
|
||||
mock<WorkflowEntity>({
|
||||
id: 'wf-1',
|
||||
name: 'Orders',
|
||||
activeVersionId: published ? 'v2' : null,
|
||||
settings: {},
|
||||
staticData: {},
|
||||
}),
|
||||
);
|
||||
const executionContexts = mock<TriggerExecutionContextFactory>();
|
||||
executionContexts.loadPublishedWorkflowData.mockResolvedValue(
|
||||
mock<IWorkflowBase>({ nodes: [trigger], connections: {} }),
|
||||
);
|
||||
const registrar = mock<NonWebhookTriggerRegistrar>();
|
||||
let onTriggerFailure: NonWebhookTriggerRegistrationContext['onTriggerFailure'] = () => {};
|
||||
registrar.createRegistrationContext.mockImplementation((_workflow, context) => {
|
||||
onTriggerFailure = context.onTriggerFailure;
|
||||
return mock<PreparedNonWebhookTriggerRegistration>();
|
||||
});
|
||||
const staticData = mock<WorkflowStaticDataService>();
|
||||
const holder = new WorkflowTriggerLeaseHolder(
|
||||
logger,
|
||||
registry,
|
||||
workflows,
|
||||
createNodeTypes(),
|
||||
registrar,
|
||||
executionContexts,
|
||||
staticData,
|
||||
);
|
||||
return {
|
||||
registry,
|
||||
staticData,
|
||||
holder,
|
||||
failAtRuntime: (error: Error) => onTriggerFailure({ error } as never),
|
||||
};
|
||||
};
|
||||
|
||||
const lease = (signal: AbortSignal) => {
|
||||
const ready = vi.fn(async () => {});
|
||||
const held: Lease = {
|
||||
kind: 'workflow-trigger',
|
||||
key: 'wf-1:kafka',
|
||||
data: { workflowId: 'wf-1', nodeId: 'kafka', registration: {} },
|
||||
fencingToken: 3,
|
||||
signal,
|
||||
ready,
|
||||
};
|
||||
return { held, ready };
|
||||
};
|
||||
|
||||
describe('WorkflowTriggerLeaseHolder', () => {
|
||||
afterEach(() => {
|
||||
vi.restoreAllMocks();
|
||||
});
|
||||
|
||||
it('arms the one trigger node, reports ready, and disarms when the lease ends', async () => {
|
||||
const { registry, staticData, holder } = setup();
|
||||
const controller = new AbortController();
|
||||
const { held, ready } = lease(controller.signal);
|
||||
|
||||
const holding = holder.hold(held);
|
||||
await vi.waitFor(() => expect(ready).toHaveBeenCalledOnce());
|
||||
|
||||
expect(registry.addTriggers).toHaveBeenCalledWith(
|
||||
'wf-1',
|
||||
expect.objectContaining({ id: 'wf-1' }),
|
||||
['kafka'],
|
||||
expect.anything(),
|
||||
'trigger',
|
||||
'activate',
|
||||
expect.anything(),
|
||||
expect.anything(),
|
||||
);
|
||||
expect(registry.removeTriggers).not.toHaveBeenCalled();
|
||||
|
||||
controller.abort(new LeaseEndedError('lost'));
|
||||
await holding;
|
||||
|
||||
expect(registry.removeTriggers).toHaveBeenCalledWith('wf-1', new Set(['kafka']));
|
||||
expect(staticData.saveStaticData).toHaveBeenCalledOnce();
|
||||
});
|
||||
|
||||
it('disarms and throws the runtime error when the trigger fails', async () => {
|
||||
const { registry, holder, failAtRuntime } = setup();
|
||||
const { held, ready } = lease(new AbortController().signal);
|
||||
|
||||
const holding = holder.hold(held);
|
||||
await vi.waitFor(() => expect(ready).toHaveBeenCalledOnce());
|
||||
failAtRuntime(new Error('broker connection closed'));
|
||||
|
||||
await expect(holding).rejects.toThrow('broker connection closed');
|
||||
expect(registry.removeTriggers).toHaveBeenCalledWith('wf-1', new Set(['kafka']));
|
||||
});
|
||||
|
||||
it('fails the start for a workflow that is not published', async () => {
|
||||
const { registry, holder } = setup({ published: false });
|
||||
const { held, ready } = lease(new AbortController().signal);
|
||||
|
||||
await expect(holder.hold(held)).rejects.toThrow('not published');
|
||||
expect(registry.addTriggers).not.toHaveBeenCalled();
|
||||
expect(ready).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('fails the start when arming throws, without reporting ready', async () => {
|
||||
const { registry, holder } = setup();
|
||||
registry.addTriggers.mockRejectedValue(new Error('invalid credentials'));
|
||||
const { held, ready } = lease(new AbortController().signal);
|
||||
|
||||
await expect(holder.hold(held)).rejects.toThrow('invalid credentials');
|
||||
expect(ready).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('rejects lease data without a workflow and node id', async () => {
|
||||
const { holder } = setup();
|
||||
const { held } = lease(new AbortController().signal);
|
||||
|
||||
await expect(holder.hold({ ...held, data: {} })).rejects.toThrow('invalid lease data');
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,115 @@
|
||||
import type { WorkflowPublishedVersionRepository } from '@n8n/db';
|
||||
import type { LeaseKind } from '@n8n/leases';
|
||||
import { mock } from 'vitest-mock-extended';
|
||||
|
||||
import type { ClusterLeases } from '@/scheduling/leases/cluster-leases';
|
||||
import { triggerRegistration } from '@/workflows/publication/trigger-diff';
|
||||
|
||||
import { createNodeTypes, node } from '../../__tests__/trigger-test-utils';
|
||||
import type { LeasedTriggerPolicy } from '../leased-trigger-policy';
|
||||
import type { WorkflowTriggerLeaseHolder } from '../workflow-trigger-lease-holder';
|
||||
import { WorkflowTriggerLeases } from '../workflow-trigger-leases';
|
||||
|
||||
const leasedTrigger = node('kafka', 'trigger');
|
||||
const pollTrigger = node('poller', 'poll');
|
||||
const disabledTrigger = node('off', 'trigger', { disabled: true });
|
||||
|
||||
const setup = ({ enabled = true } = {}) => {
|
||||
const policy = mock<LeasedTriggerPolicy>();
|
||||
policy.isEnabled.mockReturnValue(enabled);
|
||||
policy.isLeased.mockImplementation((n) => n.type === 'trigger');
|
||||
const clusterLeases = mock<ClusterLeases>();
|
||||
const publishedVersions = mock<WorkflowPublishedVersionRepository>();
|
||||
const holder = mock<WorkflowTriggerLeaseHolder>();
|
||||
const leases = new WorkflowTriggerLeases(
|
||||
policy,
|
||||
clusterLeases,
|
||||
publishedVersions,
|
||||
createNodeTypes(),
|
||||
holder,
|
||||
);
|
||||
return { clusterLeases, publishedVersions, holder, leases };
|
||||
};
|
||||
|
||||
const registeredKind = (clusterLeases: ReturnType<typeof setup>['clusterLeases']): LeaseKind =>
|
||||
clusterLeases.registerKind.mock.calls[0][0];
|
||||
|
||||
describe('WorkflowTriggerLeases', () => {
|
||||
it('registers the workflow-trigger kind with the lease holder', () => {
|
||||
const { clusterLeases, holder, leases } = setup();
|
||||
|
||||
leases.init();
|
||||
|
||||
expect(registeredKind(clusterLeases)).toMatchObject({
|
||||
name: 'workflow-trigger',
|
||||
handler: holder,
|
||||
});
|
||||
});
|
||||
|
||||
it('registers nothing when no trigger is leased', () => {
|
||||
const { clusterLeases, leases } = setup({ enabled: false });
|
||||
|
||||
leases.init();
|
||||
|
||||
expect(clusterLeases.registerKind).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('wants one lease for each enabled leased trigger of a published workflow', async () => {
|
||||
const { clusterLeases, publishedVersions, leases } = setup();
|
||||
publishedVersions.findPublishedNodes.mockResolvedValue([
|
||||
{ workflowId: 'wf-1', nodes: [leasedTrigger, pollTrigger, disabledTrigger] },
|
||||
{ workflowId: 'wf-2', nodes: [pollTrigger] },
|
||||
]);
|
||||
leases.init();
|
||||
|
||||
const desired = await registeredKind(clusterLeases).listDesired();
|
||||
|
||||
expect(publishedVersions.findPublishedNodes).toHaveBeenCalledWith(undefined);
|
||||
expect(desired).toEqual([
|
||||
{
|
||||
key: 'wf-1:kafka',
|
||||
data: {
|
||||
workflowId: 'wf-1',
|
||||
nodeId: 'kafka',
|
||||
registration: triggerRegistration(leasedTrigger),
|
||||
},
|
||||
},
|
||||
]);
|
||||
});
|
||||
|
||||
it('reads only the workflows of the asked keys and returns only those keys', async () => {
|
||||
const { clusterLeases, publishedVersions, leases } = setup();
|
||||
publishedVersions.findPublishedNodes.mockResolvedValue([
|
||||
{ workflowId: 'wf-1', nodes: [leasedTrigger, node('other', 'trigger')] },
|
||||
]);
|
||||
leases.init();
|
||||
|
||||
const desired = await registeredKind(clusterLeases).listDesired(['wf-1:kafka', 'wf-9:kafka']);
|
||||
|
||||
expect(publishedVersions.findPublishedNodes).toHaveBeenCalledWith(['wf-1', 'wf-9']);
|
||||
expect(desired.map((spec) => spec.key)).toEqual(['wf-1:kafka']);
|
||||
});
|
||||
|
||||
it('syncs the leases of the leased triggers in every given version', async () => {
|
||||
const { clusterLeases, leases } = setup();
|
||||
|
||||
await leases.changed('wf-1', [
|
||||
[leasedTrigger, pollTrigger],
|
||||
[node('renamed', 'trigger')],
|
||||
null,
|
||||
]);
|
||||
|
||||
expect(clusterLeases.changed).toHaveBeenCalledWith('workflow-trigger', [
|
||||
'wf-1:kafka',
|
||||
'wf-1:renamed',
|
||||
]);
|
||||
});
|
||||
|
||||
it('syncs nothing for versions without leased triggers', async () => {
|
||||
const { clusterLeases, leases } = setup();
|
||||
|
||||
await leases.changed('wf-1', [[pollTrigger], undefined]);
|
||||
|
||||
expect(clusterLeases.changed).not.toHaveBeenCalled();
|
||||
});
|
||||
});
|
||||
Reference in New Issue
Block a user