mirror of
https://github.com/n8n-io/n8n.git
synced 2026-09-01 15:47:41 +08:00
test(core): Add Schedule Trigger durable scheduler e2e (single- and multi-main) (no-changelog) (#34139)
Co-authored-by: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
Emilia
@@ -0,0 +1,85 @@
|
||||
import { expectScheduleTriggerFires } from './schedule-trigger-helpers';
|
||||
import {
|
||||
makeScheduleTriggerWorkflow,
|
||||
makeCronScheduleTriggerWorkflow,
|
||||
} from './schedule-trigger-workflow';
|
||||
import { test, expect } from '../../../fixtures/base';
|
||||
|
||||
const sleep = async (ms: number) => await new Promise((resolve) => setTimeout(resolve, ms));
|
||||
|
||||
// Durable scheduler path. Both flags are required: with only
|
||||
// `N8N_SCHEDULER_ENABLED` the job registrar early-returns and activation falls
|
||||
// back to the legacy in-memory timer. With both set the registrar intercepts and
|
||||
// the in-memory schedule is discarded.
|
||||
//
|
||||
// A successful trigger-mode execution does not by itself prove durable-vs-legacy
|
||||
// (both emit `mode:trigger`); the restart-continuity spec distinguishes them.
|
||||
test.use({
|
||||
capability: {
|
||||
env: {
|
||||
N8N_SCHEDULER_ENABLED: 'true',
|
||||
N8N_USE_WORKFLOW_PUBLICATION_SERVICE: 'true',
|
||||
N8N_SCHEDULER_SWEEP_INTERVAL: '1',
|
||||
N8N_SCHEDULER_EXECUTOR_INTERVAL: '1',
|
||||
},
|
||||
},
|
||||
});
|
||||
|
||||
test.describe(
|
||||
'Schedule Trigger (durable scheduler)',
|
||||
{
|
||||
annotation: [{ type: 'owner', description: 'Catalysts' }],
|
||||
},
|
||||
() => {
|
||||
test('should fire an activated Schedule Trigger through the durable scheduler', async ({
|
||||
api,
|
||||
}) => {
|
||||
await expectScheduleTriggerFires(api, makeScheduleTriggerWorkflow());
|
||||
});
|
||||
|
||||
test('should not fire once per sweep when the tick is slower than the sweep', async ({
|
||||
api,
|
||||
}) => {
|
||||
// Sweep and executor run every 1s but the schedule ticks every 2s. The
|
||||
// dedupe guards (row claim + guarded fire-time write) must collapse the
|
||||
// intervening sweeps so a single tick yields a single execution, not one
|
||||
// per second.
|
||||
const workflowId = await expectScheduleTriggerFires(api, makeScheduleTriggerWorkflow());
|
||||
|
||||
// Count the delta over a fixed window rather than the absolute total:
|
||||
// expectScheduleTriggerFires already polled for up to 60s, so ticks
|
||||
// accrued during detection must not count against the window's budget.
|
||||
const countBefore = (await api.workflows.getExecutions(workflowId, 100)).length;
|
||||
await sleep(10_000);
|
||||
const countAfter = (await api.workflows.getExecutions(workflowId, 100)).length;
|
||||
const fired = countAfter - countBefore;
|
||||
|
||||
// ~5 expected over 10s at a 2s tick. A per-sweep double-fire (once every
|
||||
// 1s) would land near ~10. Tolerant band absorbs scheduling jitter.
|
||||
expect(fired).toBeGreaterThanOrEqual(2);
|
||||
expect(fired).toBeLessThanOrEqual(8);
|
||||
});
|
||||
|
||||
test('should stop firing after the workflow is deactivated', async ({ api }) => {
|
||||
const workflowId = await expectScheduleTriggerFires(api, makeScheduleTriggerWorkflow());
|
||||
|
||||
await api.workflows.deactivate(workflowId);
|
||||
|
||||
// Let any in-flight tick settle, then snapshot and hold across several
|
||||
// intervals. Deactivation removes the scheduled job (no read path exists,
|
||||
// so this is proven indirectly by the count staying flat).
|
||||
await sleep(2_000);
|
||||
const countAfterDeactivate = (await api.workflows.getExecutions(workflowId, 50)).length;
|
||||
|
||||
await sleep(6_000);
|
||||
const countAtEnd = (await api.workflows.getExecutions(workflowId, 50)).length;
|
||||
|
||||
expect(countAtEnd).toBe(countAfterDeactivate);
|
||||
});
|
||||
|
||||
test('should fire a Schedule Trigger driven by a raw cron expression', async ({ api }) => {
|
||||
// Cron variant: exercises the `cronExpression` provisioning branch.
|
||||
await expectScheduleTriggerFires(api, makeCronScheduleTriggerWorkflow());
|
||||
});
|
||||
},
|
||||
);
|
||||
@@ -0,0 +1,27 @@
|
||||
import type { IWorkflowBase } from 'n8n-workflow';
|
||||
|
||||
import type { makeScheduleTriggerWorkflow } from './schedule-trigger-workflow';
|
||||
import { expect } from '../../../fixtures/base';
|
||||
import type { ApiHelpers } from '../../../services/api-helper';
|
||||
|
||||
type ScheduleTriggerWorkflow = ReturnType<typeof makeScheduleTriggerWorkflow>;
|
||||
|
||||
// Shared happy-path assertion: create and activate a Schedule Trigger workflow and
|
||||
// assert it produces a successful trigger-mode execution. Extracted because the
|
||||
// fire tests differ only in rule kind (seconds vs cron) and scheduler flags.
|
||||
// Returns the workflow id for callers that inspect further executions.
|
||||
export async function expectScheduleTriggerFires(
|
||||
api: ApiHelpers,
|
||||
wf: ScheduleTriggerWorkflow,
|
||||
): Promise<string> {
|
||||
const { workflowId, createdWorkflow } = await api.workflows.createWorkflowFromDefinition(
|
||||
wf.toJSON() as IWorkflowBase,
|
||||
);
|
||||
|
||||
await api.workflows.activate(workflowId, createdWorkflow.versionId!);
|
||||
|
||||
const execution = await api.workflows.waitForExecution(workflowId, 60_000, 'trigger');
|
||||
expect(execution.status).toBe('success');
|
||||
|
||||
return workflowId;
|
||||
}
|
||||
@@ -0,0 +1,20 @@
|
||||
import { expectScheduleTriggerFires } from './schedule-trigger-helpers';
|
||||
import { makeScheduleTriggerWorkflow } from './schedule-trigger-workflow';
|
||||
import { test } from '../../../fixtures/base';
|
||||
|
||||
// Flag-off parity control: with the durable scheduler disabled (default), the
|
||||
// same workflow must still fire via the legacy in-memory path. Guards against the
|
||||
// durable work regressing the common case.
|
||||
test.describe(
|
||||
'Schedule Trigger (legacy in-memory scheduler)',
|
||||
{
|
||||
annotation: [{ type: 'owner', description: 'Catalysts' }],
|
||||
},
|
||||
() => {
|
||||
test('should fire an activated Schedule Trigger through the legacy in-memory path', async ({
|
||||
api,
|
||||
}) => {
|
||||
await expectScheduleTriggerFires(api, makeScheduleTriggerWorkflow());
|
||||
});
|
||||
},
|
||||
);
|
||||
@@ -0,0 +1,113 @@
|
||||
import { expectScheduleTriggerFires } from './schedule-trigger-helpers';
|
||||
import { makeScheduleTriggerWorkflow } from './schedule-trigger-workflow';
|
||||
import { test, expect } from '../../../fixtures/base';
|
||||
|
||||
const sleep = async (ms: number) => await new Promise((resolve) => setTimeout(resolve, ms));
|
||||
|
||||
// Durable scheduler under a multi-main cluster. The scheduler has no leader: the
|
||||
// sweep, executor and reaper loops run on every main, and correctness comes from
|
||||
// the DB claim (`FOR UPDATE SKIP LOCKED` plus a `claimedBy`/`leaseEpoch` fence)
|
||||
// rather than electing one main to fire. These tests prove the two properties the
|
||||
// legacy in-memory timer cannot offer across a cluster: a tick fires exactly once
|
||||
// (not once per main), and firing survives losing a main (lease-expiry reclaim).
|
||||
//
|
||||
// Topology is inherited from the running project, not pinned here. Pinning
|
||||
// `mains: 2` in `test.use` would force a licensed 2-main stack under the sqlite
|
||||
// project too, which throws at startup without a license; instead the tests skip
|
||||
// via `mainUrls.length < 2` on single-main projects. Only the scheduler env is
|
||||
// added here, merged with the project's container config.
|
||||
test.use({
|
||||
capability: {
|
||||
env: {
|
||||
N8N_SCHEDULER_ENABLED: 'true',
|
||||
N8N_USE_WORKFLOW_PUBLICATION_SERVICE: 'true',
|
||||
N8N_SCHEDULER_SWEEP_INTERVAL: '1',
|
||||
N8N_SCHEDULER_EXECUTOR_INTERVAL: '1',
|
||||
},
|
||||
},
|
||||
});
|
||||
|
||||
test.describe(
|
||||
'Schedule Trigger multi-main (durable scheduler) @mode:multi-main',
|
||||
{
|
||||
annotation: [{ type: 'owner', description: 'Catalysts' }],
|
||||
},
|
||||
() => {
|
||||
test('should fire an activated Schedule Trigger exactly once across the cluster', async ({
|
||||
api,
|
||||
mainUrls,
|
||||
}) => {
|
||||
// Only meaningful with more than one main competing for the same tick.
|
||||
// eslint-disable-next-line playwright/no-skipped-test -- runtime topology guard, not a disabled test
|
||||
test.skip(mainUrls.length < 2, 'requires a multi-main cluster (2+ mains)');
|
||||
|
||||
const workflowId = await expectScheduleTriggerFires(api, makeScheduleTriggerWorkflow());
|
||||
|
||||
// Count the delta over a fixed window of roughly five 2s ticks. Both mains
|
||||
// run the sweep + executor every 1s, so a broken claim would let each main
|
||||
// fire the same tick, doubling the count towards ~10. A correct atomic
|
||||
// claim keeps it to one execution per tick (~5). Measuring the delta (not
|
||||
// the absolute total) keeps ticks accrued during the up-to-60s detection
|
||||
// in expectScheduleTriggerFires out of the window's budget.
|
||||
const countBefore = (await api.workflows.getExecutions(workflowId, 100)).length;
|
||||
await sleep(10_000);
|
||||
const countAfter = (await api.workflows.getExecutions(workflowId, 100)).length;
|
||||
const fired = countAfter - countBefore;
|
||||
|
||||
expect(fired).toBeGreaterThanOrEqual(2);
|
||||
expect(fired).toBeLessThanOrEqual(8);
|
||||
});
|
||||
|
||||
test('should keep firing after one main is stopped', async ({
|
||||
api,
|
||||
mainUrls,
|
||||
n8nContainer,
|
||||
createApiForMain,
|
||||
}) => {
|
||||
// eslint-disable-next-line playwright/no-skipped-test -- runtime topology guard, not a disabled test
|
||||
test.skip(mainUrls.length < 2, 'requires a multi-main cluster (2+ mains)');
|
||||
// Needs real containers to stop one; skipped against a local instance.
|
||||
// eslint-disable-next-line playwright/no-skipped-test -- container-only guard, not a disabled test
|
||||
test.skip(!n8nContainer, 'container-only: requires stoppable n8n containers');
|
||||
|
||||
const workflowId = await expectScheduleTriggerFires(api, makeScheduleTriggerWorkflow());
|
||||
|
||||
// Snapshot what fired before the stop. Continuity is only proven by an
|
||||
// execution whose id is not in this set appearing afterwards.
|
||||
const idsBeforeStop = new Set(
|
||||
(await api.workflows.getExecutions(workflowId, 50)).map((execution) => execution.id),
|
||||
);
|
||||
|
||||
// Stop main-1. There is no leader, so the survivor (main-2) is already
|
||||
// competing for the same ticks and its sweep/executor keep going; any tick
|
||||
// main-1 had claimed but not completed is reclaimed once its lease expires.
|
||||
const [stopped] = n8nContainer.findContainers(/-n8n-main-1$/);
|
||||
expect(stopped, 'main-1 container should be found').toBeDefined();
|
||||
await n8nContainer.stopContainer(/-n8n-main-1$/);
|
||||
|
||||
// Query the survivor directly rather than via the load balancer, which
|
||||
// keeps routing a share of requests to the stopped main until it drops it.
|
||||
const survivor = await createApiForMain(1);
|
||||
|
||||
// A brand-new trigger execution (not in the pre-stop set) fires without
|
||||
// re-activation and succeeds. Generous budget: covers, worst case, a
|
||||
// lease-expiry reclaim. Transient errors while the survivor settles are
|
||||
// swallowed so the poll keeps trying.
|
||||
await expect
|
||||
.poll(
|
||||
async () => {
|
||||
try {
|
||||
const fresh = (await survivor.workflows.getExecutions(workflowId, 50)).find(
|
||||
(execution) => !idsBeforeStop.has(execution.id) && execution.mode === 'trigger',
|
||||
);
|
||||
return fresh?.status ?? null;
|
||||
} catch {
|
||||
return null;
|
||||
}
|
||||
},
|
||||
{ timeout: 90_000, intervals: [2_000] },
|
||||
)
|
||||
.toBe('success');
|
||||
});
|
||||
},
|
||||
);
|
||||
@@ -0,0 +1,86 @@
|
||||
import { expectScheduleTriggerFires } from './schedule-trigger-helpers';
|
||||
import { makeScheduleTriggerWorkflow } from './schedule-trigger-workflow';
|
||||
import { test, expect } from '../../../fixtures/base';
|
||||
|
||||
// The durable job is DB state, written on activation and independent of process
|
||||
// lifetime. After a main restart the sweep reclaims it and keeps firing with no
|
||||
// re-activation. This is the property the legacy in-memory timer cannot offer,
|
||||
// and it is only observable in container mode (needs a real process restart).
|
||||
test.use({
|
||||
capability: {
|
||||
env: {
|
||||
N8N_SCHEDULER_ENABLED: 'true',
|
||||
N8N_USE_WORKFLOW_PUBLICATION_SERVICE: 'true',
|
||||
N8N_SCHEDULER_SWEEP_INTERVAL: '1',
|
||||
N8N_SCHEDULER_EXECUTOR_INTERVAL: '1',
|
||||
},
|
||||
},
|
||||
});
|
||||
|
||||
test.describe(
|
||||
'Schedule Trigger restart continuity (durable scheduler)',
|
||||
{
|
||||
annotation: [{ type: 'owner', description: 'Catalysts' }],
|
||||
},
|
||||
() => {
|
||||
test('should keep firing after a main restart without re-activation', async ({
|
||||
api,
|
||||
mainUrls,
|
||||
n8nContainer,
|
||||
}) => {
|
||||
// Single-main only. Under a cluster a surviving main keeps ticking while
|
||||
// one restarts, so the fresh execution would appear regardless of whether
|
||||
// restart recovery works, making the test pass vacuously. Cluster
|
||||
// crash-continuity is covered by the multi-main spec instead.
|
||||
// eslint-disable-next-line playwright/no-skipped-test -- runtime topology guard, not a disabled test
|
||||
test.skip(mainUrls.length >= 2, 'single-main only: cluster continuity is covered elsewhere');
|
||||
// Needs a real container to restart; skipped when running against a
|
||||
// pre-started local instance (n8nContainer is null there).
|
||||
// eslint-disable-next-line playwright/no-skipped-test -- runtime guard, not a disabled test
|
||||
test.skip(!n8nContainer, 'container-only: requires a restartable n8n container');
|
||||
|
||||
const workflowId = await expectScheduleTriggerFires(api, makeScheduleTriggerWorkflow());
|
||||
|
||||
// Snapshot the executions that exist BEFORE the restart. Continuity is
|
||||
// only proven by an execution whose id is not in this set firing after
|
||||
// the restart; `waitForExecution`'s recency fallback would otherwise
|
||||
// re-match a pre-restart execution when recovery takes under 5s.
|
||||
const idsBeforeRestart = new Set(
|
||||
(await api.workflows.getExecutions(workflowId, 50)).map((execution) => execution.id),
|
||||
);
|
||||
|
||||
// Restart the main container in place (same writable layer + DB).
|
||||
const [main] = n8nContainer.findContainers(/-n8n(-main-\d+)?$/);
|
||||
expect(main, 'main n8n container should be found').toBeDefined();
|
||||
await main.restart();
|
||||
|
||||
// Wait until the API is serving again after the restart.
|
||||
await expect
|
||||
.poll(
|
||||
async () => {
|
||||
try {
|
||||
return await api.isHealthy('readiness');
|
||||
} catch {
|
||||
return false;
|
||||
}
|
||||
},
|
||||
{ timeout: 60_000, intervals: [1_000] },
|
||||
)
|
||||
.toBe(true);
|
||||
|
||||
// A brand-new trigger execution (not in the pre-restart set) appears
|
||||
// without re-activating the workflow, and it succeeds.
|
||||
await expect
|
||||
.poll(
|
||||
async () => {
|
||||
const fresh = (await api.workflows.getExecutions(workflowId, 50)).find(
|
||||
(execution) => !idsBeforeRestart.has(execution.id) && execution.mode === 'trigger',
|
||||
);
|
||||
return fresh?.status ?? null;
|
||||
},
|
||||
{ timeout: 60_000, intervals: [1_000] },
|
||||
)
|
||||
.toBe('success');
|
||||
});
|
||||
},
|
||||
);
|
||||
@@ -0,0 +1,56 @@
|
||||
import type { IWorkflowBase } from 'n8n-workflow';
|
||||
|
||||
import { makeScheduleTriggerWorkflow } from './schedule-trigger-workflow';
|
||||
import { test, expect } from '../../../fixtures/base';
|
||||
|
||||
// Instance timezone set to a non-UTC zone. A workflow that leaves its timezone on
|
||||
// the `'DEFAULT'` sentinel must have it resolved to the instance timezone when the
|
||||
// durable path builds the trigger item; the bug was `'DEFAULT'` leaking straight
|
||||
// into `moment.tz`, which silently resolves an unknown zone to UTC.
|
||||
const INSTANCE_TIMEZONE = 'America/New_York';
|
||||
|
||||
test.use({
|
||||
capability: {
|
||||
env: {
|
||||
N8N_SCHEDULER_ENABLED: 'true',
|
||||
N8N_USE_WORKFLOW_PUBLICATION_SERVICE: 'true',
|
||||
N8N_SCHEDULER_SWEEP_INTERVAL: '1',
|
||||
N8N_SCHEDULER_EXECUTOR_INTERVAL: '1',
|
||||
GENERIC_TIMEZONE: INSTANCE_TIMEZONE,
|
||||
},
|
||||
},
|
||||
});
|
||||
|
||||
test.describe(
|
||||
'Schedule Trigger timezone (durable scheduler)',
|
||||
{
|
||||
annotation: [{ type: 'owner', description: 'Catalysts' }],
|
||||
},
|
||||
() => {
|
||||
test("should resolve the workflow's DEFAULT timezone to the instance timezone", async ({
|
||||
api,
|
||||
}) => {
|
||||
const wf = makeScheduleTriggerWorkflow();
|
||||
const { workflowId, createdWorkflow } = await api.workflows.createWorkflowFromDefinition({
|
||||
...(wf.toJSON() as IWorkflowBase),
|
||||
settings: { timezone: 'DEFAULT' },
|
||||
});
|
||||
|
||||
await api.workflows.activate(workflowId, createdWorkflow.versionId!);
|
||||
|
||||
const execution = await api.workflows.waitForExecution(workflowId, 60_000, 'trigger');
|
||||
expect(execution.status).toBe('success');
|
||||
|
||||
const full = await api.workflows.getExecution(execution.id, { redactExecutionData: false });
|
||||
|
||||
// The emitted item's Timezone field is `<zone> (UTC<offset>)`. Resolved
|
||||
// correctly it names the instance zone; the pre-fix leak resolved the
|
||||
// `'DEFAULT'` sentinel to UTC. America/New_York is never UTC+00:00 in any
|
||||
// season, so asserting the zone name and a non-UTC offset is DST-proof
|
||||
// without hardcoding -04:00/-05:00.
|
||||
expect(full.data).toContain(INSTANCE_TIMEZONE);
|
||||
expect(full.data).not.toContain('DEFAULT (UTC');
|
||||
expect(full.data).not.toContain('(UTC+00:00)');
|
||||
});
|
||||
},
|
||||
);
|
||||
@@ -0,0 +1,56 @@
|
||||
import { workflow, trigger, node } from '@n8n/workflow-sdk';
|
||||
import { nanoid } from 'nanoid';
|
||||
|
||||
// Builds a Schedule Trigger -> NoOp workflow firing every couple of seconds, so a
|
||||
// trigger-mode execution appears quickly and the observation windows stay short.
|
||||
export const makeScheduleTriggerWorkflow = (secondsInterval = 2) => {
|
||||
const scheduleTrigger = trigger({
|
||||
type: 'n8n-nodes-base.scheduleTrigger',
|
||||
version: 1.3,
|
||||
config: {
|
||||
name: 'Schedule Trigger',
|
||||
parameters: {
|
||||
rule: {
|
||||
interval: [{ field: 'seconds', secondsInterval }],
|
||||
},
|
||||
},
|
||||
},
|
||||
});
|
||||
|
||||
const noOp = node({
|
||||
type: 'n8n-nodes-base.noOp',
|
||||
version: 1,
|
||||
config: {
|
||||
name: 'NoOp',
|
||||
},
|
||||
});
|
||||
|
||||
return workflow(nanoid(), `Schedule Trigger Test ${nanoid()}`).add(scheduleTrigger.to(noOp));
|
||||
};
|
||||
|
||||
// Same shape but driven by a raw cron expression (the `cronExpression` field
|
||||
// takes a separate provisioning branch from the fixed-interval fields).
|
||||
export const makeCronScheduleTriggerWorkflow = (expression = '*/2 * * * * *') => {
|
||||
const scheduleTrigger = trigger({
|
||||
type: 'n8n-nodes-base.scheduleTrigger',
|
||||
version: 1.3,
|
||||
config: {
|
||||
name: 'Schedule Trigger',
|
||||
parameters: {
|
||||
rule: {
|
||||
interval: [{ field: 'cronExpression', expression }],
|
||||
},
|
||||
},
|
||||
},
|
||||
});
|
||||
|
||||
const noOp = node({
|
||||
type: 'n8n-nodes-base.noOp',
|
||||
version: 1,
|
||||
config: {
|
||||
name: 'NoOp',
|
||||
},
|
||||
});
|
||||
|
||||
return workflow(nanoid(), `Schedule Trigger Cron Test ${nanoid()}`).add(scheduleTrigger.to(noOp));
|
||||
};
|
||||
Reference in New Issue
Block a user