feat(core): Coalesce a Schedule Trigger node's catch-up runs after downtime (#35524)

Co-authored-by: Claude Opus 5 <[email protected]>
Co-authored-by: Lorent Lempereur <[email protected]>

docs/generated/postgres-schema/public.scheduled_job.md

@@ -14,7 +14,7 @@
 | lastFiredAt | timestamp(3) with time zone |  | true |  |  | Last time an occurrence was materialized; used to recompute nextRunAt. |
 | maxAttempts | integer | 1 | false |  |  | Retry ceiling copied onto each occurrence this job materializes. |
 | misfireGraceSeconds | integer | 60 | false |  |  | How late an occurrence may be before the misfire policy applies to it; an ordinary restart stays inside it. |
-| misfirePolicy | varchar(16) | 'coalesce'::character varying | false |  |  | What to do with occurrences that came due while nothing ran them: 'coalesce' records a single catch-up run, 'skip' records none. |
+| misfirePolicy | varchar(16) | 'coalesce'::character varying | false |  |  | What to do with occurrences that came due while nothing ran them: 'coalesce' records a single late run per job, 'coalesce_owner' a single one across every job the same owner scheduled, 'skip' records none. |
 | name | varchar(255) |  | false |  |  | Human-readable job name. A well-known scheduler key for system jobs; generated for workflow trigger jobs. |
 | nextRunAt | timestamp(3) with time zone |  | true |  |  | Next time an occurrence is due; the scheduler sweep reads this to find work. NULL once disabled or a one-off has fired. |
 | nodeId | varchar(36) |  | true |  |  | Trigger node within the workflow that owns this job; NULL for non-trigger jobs. |
@@ -35,7 +35,7 @@
 | CHK_scheduled_job_interval_seconds | CHECK | CHECK ((((kind)::text <> 'interval'::text) OR ("intervalSeconds" IS NOT NULL))) |
 | CHK_scheduled_job_kind | CHECK | CHECK (((kind)::text = ANY ((ARRAY['cron'::character varying, 'interval'::character varying, 'one_off'::character varying, 'recurring_cron'::character varying])::text[]))) |
 | CHK_scheduled_job_misfireGraceSeconds | CHECK | CHECK (("misfireGraceSeconds" > 0)) |
-| CHK_scheduled_job_misfirePolicy | CHECK | CHECK ((("misfirePolicy")::text = ANY ((ARRAY['coalesce'::character varying, 'skip'::character varying])::text[]))) |
+| CHK_scheduled_job_misfirePolicy | CHECK | CHECK ((("misfirePolicy")::text = ANY ((ARRAY['coalesce'::character varying, 'skip'::character varying, 'coalesce_owner'::character varying])::text[]))) |
 | CHK_scheduled_job_recurrence_size | CHECK | CHECK (("recurrenceSize" >= 2)) |
 | CHK_scheduled_job_recurrence_unit | CHECK | CHECK ((("recurrenceUnit")::text = ANY ((ARRAY['hours'::character varying, 'days'::character varying, 'weeks'::character varying, 'months'::character varying])::text[]))) |
 | CHK_scheduled_job_recurring_cron | CHECK | CHECK ((((kind)::text <> 'recurring_cron'::text) OR (("cronExpression" IS NOT NULL) AND ("recurrenceUnit" IS NOT NULL) AND ("recurrenceSize" IS NOT NULL)))) |

docs/generated/sqlite-schema/scheduled_job.md

@@ -6,7 +6,7 @@
 <summary><strong>Table Definition</strong></summary>
 
 ```sql
-CREATE TABLE "scheduled_job" ("id" integer PRIMARY KEY NOT NULL, "name" varchar(255) NOT NULL, "workflowId" varchar(36), "nodeId" varchar(36), "taskType" varchar(128) NOT NULL, "payload" text NOT NULL DEFAULT ('{}'), "kind" varchar(16) NOT NULL, "cronExpression" varchar(255), "timezone" varchar(64), "intervalSeconds" integer, "fireAt" datetime(3), "enabled" boolean NOT NULL DEFAULT (true), "nextRunAt" datetime(3), "lastFiredAt" datetime(3), "maxAttempts" integer NOT NULL DEFAULT (1), "createdAt" datetime(3) NOT NULL DEFAULT (STRFTIME('%Y-%m-%d %H:%M:%f', 'NOW')), "updatedAt" datetime(3) NOT NULL DEFAULT (STRFTIME('%Y-%m-%d %H:%M:%f', 'NOW')), "recurrenceUnit" varchar(16), "recurrenceSize" int, "misfirePolicy" varchar(16) NOT NULL DEFAULT 'coalesce' CONSTRAINT "CHK_scheduled_job_misfirePolicy" CHECK ("misfirePolicy" IN ('coalesce', 'skip')), "misfireGraceSeconds" int NOT NULL DEFAULT 60 CONSTRAINT "CHK_scheduled_job_misfireGraceSeconds" CHECK ("misfireGraceSeconds" > 0), CONSTRAINT "CHK_scheduled_job_recurrence_unit" CHECK (("recurrenceUnit" IN ('hours', 'days', 'weeks', 'months'))), CONSTRAINT "CHK_scheduled_job_recurrence_size" CHECK (("recurrenceSize" >= 2)), CONSTRAINT "CHK_scheduled_job_cron_expression" CHECK (("kind" <> 'cron' OR "cronExpression" IS NOT NULL)), CONSTRAINT "CHK_scheduled_job_interval_seconds" CHECK (("kind" <> 'interval' OR "intervalSeconds" IS NOT NULL)), CONSTRAINT "CHK_scheduled_job_fire_at" CHECK (("kind" <> 'one_off' OR "fireAt" IS NOT NULL)), CONSTRAINT "CHK_scheduled_job_kind" CHECK ("kind" IN ('cron', 'interval', 'one_off', 'recurring_cron')), CONSTRAINT "CHK_scheduled_job_recurring_cron" CHECK ("kind" <> 'recurring_cron' OR ("cronExpression" IS NOT NULL AND "recurrenceUnit" IS NOT NULL AND "recurrenceSize" IS NOT NULL)), CONSTRAINT "FK_scheduled_job_workflowId" FOREIGN KEY ("workflowId") REFERENCES "workflow_published_version" ("workflowId") ON DELETE CASCADE ON UPDATE NO ACTION)
+CREATE TABLE "scheduled_job" ("id" integer PRIMARY KEY NOT NULL, "name" varchar(255) NOT NULL, "workflowId" varchar(36), "nodeId" varchar(36), "taskType" varchar(128) NOT NULL, "payload" text NOT NULL DEFAULT ('{}'), "kind" varchar(16) NOT NULL, "cronExpression" varchar(255), "timezone" varchar(64), "intervalSeconds" integer, "fireAt" datetime(3), "enabled" boolean NOT NULL DEFAULT (true), "nextRunAt" datetime(3), "lastFiredAt" datetime(3), "maxAttempts" integer NOT NULL DEFAULT (1), "createdAt" datetime(3) NOT NULL DEFAULT (STRFTIME('%Y-%m-%d %H:%M:%f', 'NOW')), "updatedAt" datetime(3) NOT NULL DEFAULT (STRFTIME('%Y-%m-%d %H:%M:%f', 'NOW')), "recurrenceUnit" varchar(16), "recurrenceSize" int, "misfirePolicy" varchar(16) NOT NULL DEFAULT ('coalesce'), "misfireGraceSeconds" int NOT NULL DEFAULT (60), CONSTRAINT "CHK_scheduled_job_misfireGraceSeconds" CHECK (("misfireGraceSeconds" > 0)), CONSTRAINT "CHK_scheduled_job_recurrence_unit" CHECK ((("recurrenceUnit" IN ('hours', 'days', 'weeks', 'months')))), CONSTRAINT "CHK_scheduled_job_recurrence_size" CHECK ((("recurrenceSize" >= 2))), CONSTRAINT "CHK_scheduled_job_cron_expression" CHECK ((("kind" <> 'cron' OR "cronExpression" IS NOT NULL))), CONSTRAINT "CHK_scheduled_job_interval_seconds" CHECK ((("kind" <> 'interval' OR "intervalSeconds" IS NOT NULL))), CONSTRAINT "CHK_scheduled_job_fire_at" CHECK ((("kind" <> 'one_off' OR "fireAt" IS NOT NULL))), CONSTRAINT "CHK_scheduled_job_kind" CHECK (("kind" IN ('cron', 'interval', 'one_off', 'recurring_cron'))), CONSTRAINT "CHK_scheduled_job_recurring_cron" CHECK (("kind" <> 'recurring_cron' OR ("cronExpression" IS NOT NULL AND "recurrenceUnit" IS NOT NULL AND "recurrenceSize" IS NOT NULL))), CONSTRAINT "CHK_scheduled_job_misfirePolicy" CHECK ("misfirePolicy" IN ('coalesce', 'skip', 'coalesce_owner')), CONSTRAINT "FK_scheduled_job_workflowId" FOREIGN KEY ("workflowId") REFERENCES "workflow_published_version" ("workflowId") ON DELETE CASCADE ON UPDATE NO ACTION)
 ```
 
 </details>
@@ -41,15 +41,15 @@ CREATE TABLE "scheduled_job" ("id" integer PRIMARY KEY NOT NULL, "name" varchar(
 
 | Name | Type | Definition |
 | ---- | ---- | ---------- |
-| - | CHECK | CHECK ("misfirePolicy" IN ('coalesce', 'skip')) |
-| - | CHECK | CHECK ("misfireGraceSeconds" > 0) |
-| - | CHECK | CHECK (("recurrenceUnit" IN ('hours', 'days', 'weeks', 'months'))) |
-| - | CHECK | CHECK (("recurrenceSize" >= 2)) |
-| - | CHECK | CHECK (("kind" <> 'cron' OR "cronExpression" IS NOT NULL)) |
-| - | CHECK | CHECK (("kind" <> 'interval' OR "intervalSeconds" IS NOT NULL)) |
-| - | CHECK | CHECK (("kind" <> 'one_off' OR "fireAt" IS NOT NULL)) |
-| - | CHECK | CHECK ("kind" IN ('cron', 'interval', 'one_off', 'recurring_cron')) |
-| - | CHECK | CHECK ("kind" <> 'recurring_cron' OR ("cronExpression" IS NOT NULL AND "recurrenceUnit" IS NOT NULL AND "recurrenceSize" IS NOT NULL)) |
+| - | CHECK | CHECK (("misfireGraceSeconds" > 0)) |
+| - | CHECK | CHECK ((("recurrenceUnit" IN ('hours', 'days', 'weeks', 'months')))) |
+| - | CHECK | CHECK ((("recurrenceSize" >= 2))) |
+| - | CHECK | CHECK ((("kind" <> 'cron' OR "cronExpression" IS NOT NULL))) |
+| - | CHECK | CHECK ((("kind" <> 'interval' OR "intervalSeconds" IS NOT NULL))) |
+| - | CHECK | CHECK ((("kind" <> 'one_off' OR "fireAt" IS NOT NULL))) |
+| - | CHECK | CHECK (("kind" IN ('cron', 'interval', 'one_off', 'recurring_cron'))) |
+| - | CHECK | CHECK (("kind" <> 'recurring_cron' OR ("cronExpression" IS NOT NULL AND "recurrenceUnit" IS NOT NULL AND "recurrenceSize" IS NOT NULL))) |
+| - | CHECK | CHECK ("misfirePolicy" IN ('coalesce', 'skip', 'coalesce_owner')) |
 | - (Foreign key ID: 0) | FOREIGN KEY | FOREIGN KEY (workflowId) REFERENCES workflow_published_version (workflowId) ON UPDATE NO ACTION ON DELETE CASCADE MATCH NONE |
 | id | PRIMARY KEY | PRIMARY KEY (id) |
 

packages/@n8n/constants/src/scheduler.ts

@@ -49,17 +49,24 @@ export type RecurringCronUnit = (typeof RecurringCronUnit)[keyof typeof Recurrin
 export const RecurringCronUnitList = Object.values(RecurringCronUnit);
 
 /**
- * What a schedule does with occurrences that missed their grace window:
- * - `coalesce`: run the most recent one and drop the rest
- * - `skip`: drop all of them and resume from the next occurrence
+ * What happens to occurrences that missed their grace window:
+ * - `coalesce`: run the latest one, drop the rest
+ * - `coalesce_owner`: same as `coalesce`, but per owner. Only the job with
+ *   the latest missed occurrence runs late. The other jobs sharing that
+ *   owner drop theirs.
+ * - `skip`: drop them all, resume from the next occurrence
  *
- * Either way the schedule's clock advances past the backlog, so it never replays.
+ * In every case the clock moves past the backlog, so nothing replays.
  *
- * A one-off schedule has no next occurrence to resume from, so `coalesce` still runs
- * it, late, while `skip` discards it for good.
+ * One-off schedules have no next occurrence to fall back on:
+ * - `coalesce` still runs it, late
+ * - `skip` drops it for good
+ * - `coalesce_owner` runs it late only if it wins its owner group; losing
+ *   means it never runs at all
  */
 export const ScheduledJobMisfirePolicy = {
 	Coalesce: 'coalesce',
+	CoalesceOwner: 'coalesce_owner',
 	Skip: 'skip',
 } as const;
 

packages/@n8n/db/src/migrations/common/1786666615643-AddCoalesceOwnerMisfirePolicy.ts

@@ -0,0 +1,101 @@
+import type { MigrationContext, ReversibleMigration } from '../migration-types';
+
+const jobTable = 'scheduled_job';
+const policyColumn = 'misfirePolicy';
+const taskTypeColumn = 'taskType';
+
+// Pinned like the policy values below: this migration must keep targeting the
+// same rows even if the constant in `packages/cli` moves or changes.
+const SCHEDULE_TRIGGER_TASK_TYPE = 'workflow:schedule-trigger';
+
+// Pinned rather than read from `ScheduledJobMisfirePolicy`: a migration has to keep
+// meaning what it meant when it ran, so a later addition to the enum needs its own
+// migration and every instance converges on the same CHECK.
+const POLICY_VALUES_BEFORE = ['coalesce', 'skip'];
+const POLICY_VALUES = [...POLICY_VALUES_BEFORE, 'coalesce_owner'];
+
+const POLICY_COMMENT_BEFORE =
+	"What to do with occurrences that came due while nothing ran them: ''coalesce'' records a single catch-up run, ''skip'' records none.";
+const POLICY_COMMENT =
+	"What to do with occurrences that came due while nothing ran them: ''coalesce'' records a single late run per job, ''coalesce_owner'' a single one across every job the same owner scheduled, ''skip'' records none.";
+
+/**
+ * Two changes to `misfirePolicy`:
+ * - widens its CHECK to accept the new `coalesce_owner` value
+ * - moves existing Schedule Trigger jobs from `coalesce` to `skip`
+ *
+ * `skip` matches what n8n did before the durable scheduler: a missed run is
+ * dropped, never run late. It is also an old, widely understood value, so
+ * binaries from earlier releases handle the moved rows fine during a rolling
+ * upgrade.
+ *
+ * `down()` moves those rows back to `coalesce`, and folds any `coalesce_owner`
+ * rows to `coalesce` too (instead of deleting them), so a rollback leaves
+ * every live trigger scheduled.
+ */
+export class AddCoalesceOwnerMisfirePolicy1786666615643 implements ReversibleMigration {
+	async up(context: MigrationContext) {
+		await this.refreshTableMetadata(context);
+		await this.setPolicyCheck(context, POLICY_VALUES);
+		await this.moveScheduleTriggerPolicy(context, { from: 'coalesce', to: 'skip' });
+		if (context.isPostgres) {
+			await this.commentPolicyColumn(context, POLICY_COMMENT);
+		}
+	}
+
+	async down(context: MigrationContext) {
+		await this.refreshTableMetadata(context);
+		await this.foldOwnerPolicyBack(context);
+		await this.moveScheduleTriggerPolicy(context, { from: 'skip', to: 'coalesce' });
+		await this.setPolicyCheck(context, POLICY_VALUES_BEFORE);
+		if (context.isPostgres) {
+			await this.commentPolicyColumn(context, POLICY_COMMENT_BEFORE);
+		}
+	}
+
+	/**
+	 * `misfirePolicy`, `misfireGraceSeconds` and `missedAfter` were added with raw
+	 * `ALTER TABLE`, which TypeORM never observed, so the `Table` it caches per query
+	 * runner can be stale. SQLite rebuilds the table from that cache for the CHECK swap
+	 * and would silently drop those columns; every dialect refreshes rather than rely
+	 * on what the cache happens to hold.
+	 */
+	private async refreshTableMetadata({ queryRunner, tablePrefix }: MigrationContext) {
+		await queryRunner.getTable(`${tablePrefix}${jobTable}`);
+	}
+
+	private async setPolicyCheck({ schemaBuilder }: MigrationContext, values: string[]) {
+		await schemaBuilder.dropEnumCheck(jobTable, policyColumn, { recreatesOnSqlite: true });
+		await schemaBuilder.addEnumCheck(jobTable, policyColumn, values, {
+			recreatesOnSqlite: true,
+		});
+	}
+
+	private async foldOwnerPolicyBack({ escape, runQuery }: MigrationContext) {
+		const policy = escape.columnName(policyColumn);
+		await runQuery(
+			`UPDATE ${escape.tableName(jobTable)} SET ${policy} = 'coalesce' ` +
+				`WHERE ${policy} = 'coalesce_owner'`,
+		);
+	}
+
+	/** Only Schedule Trigger jobs: other task types pick their own policy. */
+	private async moveScheduleTriggerPolicy(
+		{ escape, runQuery }: MigrationContext,
+		{ from, to }: { from: string; to: string },
+	) {
+		const policy = escape.columnName(policyColumn);
+		const taskType = escape.columnName(taskTypeColumn);
+		await runQuery(
+			`UPDATE ${escape.tableName(jobTable)} SET ${policy} = '${to}' ` +
+				`WHERE ${policy} = '${from}' AND ${taskType} = '${SCHEDULE_TRIGGER_TASK_TYPE}'`,
+		);
+	}
+
+	/** Doubled quotes escape a literal apostrophe inside the SQL string. */
+	private async commentPolicyColumn({ escape, runQuery }: MigrationContext, comment: string) {
+		await runQuery(
+			`COMMENT ON COLUMN ${escape.tableName(jobTable)}.${escape.columnName(policyColumn)} IS '${comment}'`,
+		);
+	}
+}

packages/@n8n/db/src/migrations/sqlite/1786666615643-AddCoalesceOwnerMisfirePolicy.ts

@@ -0,0 +1,15 @@
+import { AddCoalesceOwnerMisfirePolicy1786666615643 as BaseMigration } from '../common/1786666615643-AddCoalesceOwnerMisfirePolicy';
+
+/**
+ * Swapping the `misfirePolicy` CHECK recreates the whole `scheduled_job` table on SQLite,
+ * on the way up as well as down. `scheduled_task` references it with ON DELETE CASCADE,
+ * so the recreate's DROP would cascade and wipe queued tasks.
+ *
+ * Only the rollback needs this flag: for `up`, TypeORM's own `beforeMigration` hook
+ * already issues `PRAGMA foreign_keys = OFF` before the migration transaction opens. On
+ * the rollback path it issues that pragma inside the transaction, where SQLite ignores
+ * it, so this flag keeps the drop local to `scheduled_job` on that path.
+ */
+export class AddCoalesceOwnerMisfirePolicy1786666615643 extends BaseMigration {
+	withFKsDisabled = true as const;
+}

packages/@n8n/scheduler/README.md

@@ -180,6 +180,42 @@ The filter looks only at the *previous* run, with no stored counter. So any serv
 in the cluster can compute the next run on its own, and a restart never loses the
 cadence.
 
+## Misfires and late runs
+
+A **misfire** is a run that came due while nothing was there to fire it, and is now
+past its grace window (`misfireGraceSeconds`); while still inside that window a late
+instant is simply queued and run late, no policy involved. Once the oldest overdue
+instant is past its window, the job's **misfire policy** decides what happens to the
+rest of the backlog:
+
+| Policy | What it does |
+|---|---|
+| `skip` | Records nothing for the backlog; the rule resumes from its next future instant. |
+| `coalesce` | Records one **late run** for that job, standing in for the whole backlog. |
+| `coalesce_owner` | Records one late run for the whole **owner** (see below), across every job that shares it. |
+
+A late run is a normal task, queued at the most recent missed instant and
+claimable immediately, so a backlog fires once rather than N times back to back. The
+job's clock advances past the backlog in the same transaction as the late run, so
+no two servers can disagree about what a backlog produced.
+
+### Why `coalesce_owner` exists
+
+`coalesce` operates per job, so a Schedule Trigger with several rules on one node
+still produces one late run per rule after downtime, N fires in a row.
+`coalesce_owner` closes that gap: jobs that share an **owner** (`ownerKey`, an opaque
+per-job key the scheduler only compares for equality; in n8n, the trigger node) group
+unconditionally, and only the one with the latest missed instant survives a planning
+pass (ties break on the lowest job id).
+
+Grouping only sees jobs claimed together in one pass, so it is best-effort, not a
+guarantee of one fire per owner: a sibling not yet due, or one whose backlog exceeds
+`maxPerJob`, still records its own late run on a later pass.
+
+Schedule and poll triggers both use `skip`: it matches what n8n did before the
+durable scheduler, where a missed run was never replayed. Running late is opt-in
+per job, which is where `coalesce` and `coalesce_owner` come in.
+
 ## Durable and distributed
 
 Two properties describe the whole design, and everything else follows from them.

packages/@n8n/scheduler/src/core/__tests__/factory.test.ts

@@ -348,6 +348,7 @@ describe('createScheduler materialize', () => {
 			maxAttempts: 3,
 			misfirePolicy: ScheduledJobMisfirePolicy.Coalesce,
 			misfireGraceSeconds: 60,
+			ownerKey: null,
 		};
 		const tx = mock<MaterializerTransaction>();
 		tx.retireSuperseded.mockResolvedValue(0);
@@ -393,6 +394,7 @@ describe('createScheduler materialize', () => {
 			maxAttempts: 3,
 			misfirePolicy: ScheduledJobMisfirePolicy.Coalesce,
 			misfireGraceSeconds: 60,
+			ownerKey: null,
 		};
 		const tx = mock<MaterializerTransaction>();
 		tx.retireSuperseded.mockResolvedValue(0);
@@ -1045,6 +1047,7 @@ describe('createScheduler tracing', () => {
 			maxAttempts: 3,
 			misfirePolicy: ScheduledJobMisfirePolicy.Coalesce,
 			misfireGraceSeconds: 60,
+			ownerKey: null,
 		};
 		const tx = mock<MaterializerTransaction>();
 		tx.retireSuperseded.mockResolvedValue(0);
@@ -1097,6 +1100,7 @@ describe('createScheduler tracing', () => {
 			maxAttempts: 3,
 			misfirePolicy: ScheduledJobMisfirePolicy.Coalesce,
 			misfireGraceSeconds: 60,
+			ownerKey: null,
 		};
 		const tx = mock<MaterializerTransaction>();
 		tx.retireSuperseded.mockResolvedValue(0);
@@ -1392,6 +1396,7 @@ describe('createScheduler metrics', () => {
 			maxAttempts: 3,
 			misfirePolicy: ScheduledJobMisfirePolicy.Coalesce,
 			misfireGraceSeconds: 60,
+			ownerKey: null,
 		};
 		const tx = mock<MaterializerTransaction>();
 		tx.retireSuperseded.mockResolvedValue(0);

packages/@n8n/scheduler/src/core/materializer/__tests__/coalesce-group.test.ts

@@ -0,0 +1,342 @@
+import { ScheduledJobMisfirePolicy } from '@n8n/constants';
+
+import type { ScheduledJob } from '../../types';
+import { applyCoalesceOwnerPolicy } from '../coalesce-group';
+import type { OccurrencePlan } from '../plan';
+import type { PlannedJob } from '../transaction';
+
+const NOW = new Date('2026-01-01T00:00:00.000Z');
+const secondsAfter = (base: Date, seconds: number) => new Date(base.getTime() + seconds * 1000);
+const secondsBefore = (base: Date, seconds: number) => secondsAfter(base, -seconds);
+
+const makeJob = (overrides: Partial<ScheduledJob> = {}): ScheduledJob => ({
+	id: 1,
+	taskType: 'test',
+	payload: {},
+	kind: 'interval',
+	cronExpression: null,
+	timezone: null,
+	intervalSeconds: 10,
+	fireAt: null,
+	recurrenceUnit: null,
+	recurrenceSize: null,
+	nextRunAt: NOW,
+	lastFiredAt: null,
+	maxAttempts: 1,
+	misfirePolicy: ScheduledJobMisfirePolicy.CoalesceOwner,
+	misfireGraceSeconds: 60,
+	ownerKey: 'owner-a',
+	...overrides,
+});
+
+const makePlan = (overrides: Partial<OccurrencePlan> = {}): OccurrencePlan => ({
+	occurrences: [],
+	skippedOccurrences: 0,
+	catchUpAt: null,
+	retireBefore: null,
+	nextRunAt: secondsAfter(NOW, 10),
+	lastFiredAt: null,
+	...overrides,
+});
+
+const makeMember = (
+	job: Partial<ScheduledJob>,
+	missedAt: Date,
+	plan: Partial<OccurrencePlan> = {},
+): PlannedJob => ({
+	job: makeJob(job),
+	plan: makePlan({
+		occurrences: [missedAt],
+		skippedOccurrences: 4,
+		catchUpAt: missedAt,
+		retireBefore: missedAt,
+		lastFiredAt: secondsBefore(NOW, 10),
+		...plan,
+	}),
+});
+
+const byId = (result: PlannedJob[], id: number): PlannedJob => {
+	const found = result.find(({ job }) => job.id === id);
+	if (found === undefined) throw new Error(`no planned job with id ${id}`);
+	return found;
+};
+
+describe('applyCoalesceOwnerPolicy', () => {
+	describe('a group of overdue siblings', () => {
+		const missedOne = secondsBefore(NOW, 30);
+		const missedTwo = secondsBefore(NOW, 20);
+		const missedThree = secondsBefore(NOW, 10);
+
+		const makeSiblings = (): PlannedJob[] => [
+			makeMember({ id: 1 }, missedOne),
+			makeMember({ id: 2 }, missedTwo),
+			makeMember({ id: 3 }, missedThree),
+		];
+
+		it("keeps the winner's late run and plan untouched", () => {
+			const result = applyCoalesceOwnerPolicy(makeSiblings());
+
+			const withLateRun = result.filter(({ plan }) => plan.catchUpAt !== null);
+			expect(withLateRun).toHaveLength(1);
+			const winner = withLateRun[0];
+			expect(winner.job.id).toBe(3);
+			expect(winner.plan.catchUpAt).toEqual(missedThree);
+			expect(winner.plan.occurrences).toEqual([missedThree]);
+			expect(winner.plan.retireBefore).toEqual(missedThree);
+		});
+
+		it("drops each loser's late run, leaving its clock, retirement instant and backlog count untouched", () => {
+			const result = applyCoalesceOwnerPolicy(makeSiblings());
+
+			const loserOne = byId(result, 1);
+			expect(loserOne.plan.catchUpAt).toBeNull();
+			expect(loserOne.plan.retireBefore).toEqual(missedOne);
+
+			const loserTwo = byId(result, 2);
+			expect(loserTwo.plan.catchUpAt).toBeNull();
+			expect(loserTwo.plan.retireBefore).toEqual(missedTwo);
+
+			for (const id of [1, 2, 3]) {
+				const member = byId(result, id);
+				expect(member.plan.nextRunAt).toEqual(secondsAfter(NOW, 10));
+				expect(member.plan.lastFiredAt).toEqual(secondsBefore(NOW, 10));
+				expect(member.plan.skippedOccurrences).toBe(4);
+			}
+		});
+	});
+
+	it('removes only the late run from a loser, keeping its future occurrences in order', () => {
+		const missedAt = secondsBefore(NOW, 30);
+		const laterMissed = secondsBefore(NOW, 10);
+		const futureOne = secondsAfter(NOW, 10);
+		const futureTwo = secondsAfter(NOW, 20);
+		const planned = [
+			makeMember({ id: 1 }, missedAt, {
+				occurrences: [missedAt, futureOne, futureTwo],
+				nextRunAt: secondsAfter(NOW, 30),
+			}),
+			makeMember({ id: 2 }, laterMissed),
+		];
+
+		const result = applyCoalesceOwnerPolicy(planned);
+
+		const loser = byId(result, 1);
+		expect(loser.plan.occurrences).toEqual([futureOne, futureTwo]);
+		expect(loser.plan.nextRunAt).toEqual(secondsAfter(NOW, 30));
+	});
+
+	it('leaves a lone member of an owner untouched', () => {
+		const missedAt = secondsBefore(NOW, 30);
+		const planned = [makeMember({ id: 1 }, missedAt)];
+
+		const result = applyCoalesceOwnerPolicy(planned);
+
+		expect(result).toHaveLength(1);
+		expect(result[0].plan.catchUpAt).toEqual(missedAt);
+		expect(result[0].plan.occurrences).toEqual([missedAt]);
+		expect(result[0].plan.retireBefore).toEqual(missedAt);
+	});
+
+	it('leaves ownerless members untouched even when they share a task type', () => {
+		const planned = [
+			makeMember({ id: 1, ownerKey: null }, secondsBefore(NOW, 30)),
+			makeMember({ id: 2, ownerKey: null }, secondsBefore(NOW, 20)),
+			makeMember({ id: 3, ownerKey: null }, secondsBefore(NOW, 10)),
+		];
+
+		const result = applyCoalesceOwnerPolicy(planned);
+
+		expect(result.filter(({ plan }) => plan.catchUpAt !== null)).toHaveLength(3);
+	});
+
+	it('leaves members on the per-job coalesce policy untouched despite a shared owner', () => {
+		const perJob = (id: number, missedAt: Date) =>
+			makeMember({ id, misfirePolicy: ScheduledJobMisfirePolicy.Coalesce }, missedAt);
+		const planned = [perJob(1, secondsBefore(NOW, 30)), perJob(2, secondsBefore(NOW, 20))];
+
+		const result = applyCoalesceOwnerPolicy(planned);
+
+		expect(result.filter(({ plan }) => plan.catchUpAt !== null)).toHaveLength(2);
+	});
+
+	it('leaves members on the skip policy untouched despite a shared owner', () => {
+		const ahead = secondsAfter(NOW, 10);
+		const skipped = (id: number): PlannedJob => ({
+			job: makeJob({ id, misfirePolicy: ScheduledJobMisfirePolicy.Skip }),
+			plan: makePlan({ occurrences: [ahead], skippedOccurrences: 4 }),
+		});
+		const planned = [skipped(1), skipped(2)];
+
+		const result = applyCoalesceOwnerPolicy(planned);
+
+		for (const id of [1, 2]) {
+			const member = byId(result, id);
+			expect(member.plan.occurrences).toEqual([ahead]);
+			expect(member.plan.retireBefore).toBeNull();
+		}
+	});
+
+	it('groups each owner in a batch separately', () => {
+		const planned = [
+			makeMember({ id: 1, ownerKey: 'owner-a' }, secondsBefore(NOW, 30)),
+			makeMember({ id: 2, ownerKey: 'owner-a' }, secondsBefore(NOW, 20)),
+			makeMember({ id: 3, ownerKey: 'owner-b' }, secondsBefore(NOW, 45)),
+			makeMember({ id: 4, ownerKey: 'owner-b' }, secondsBefore(NOW, 5)),
+		];
+
+		const result = applyCoalesceOwnerPolicy(planned);
+
+		expect(byId(result, 1).plan.catchUpAt).toBeNull();
+		expect(byId(result, 2).plan.catchUpAt).toEqual(secondsBefore(NOW, 20));
+		expect(byId(result, 3).plan.catchUpAt).toBeNull();
+		expect(byId(result, 4).plan.catchUpAt).toEqual(secondsBefore(NOW, 5));
+	});
+
+	it('breaks a tie on the missed instant by the lowest job id', () => {
+		const tied = secondsBefore(NOW, 10);
+		const planned = [makeMember({ id: 5 }, tied), makeMember({ id: 2 }, tied)];
+
+		const result = applyCoalesceOwnerPolicy(planned);
+
+		expect(byId(result, 2).plan.catchUpAt).toEqual(tied);
+		expect(byId(result, 5).plan.catchUpAt).toBeNull();
+	});
+
+	it('drops the late run of an exhausted member the same as any other loser', () => {
+		const exhaustedMissed = secondsBefore(NOW, 30);
+		const laterMissed = secondsBefore(NOW, 10);
+		const planned = [
+			makeMember({ id: 1 }, exhaustedMissed, { nextRunAt: null }),
+			makeMember({ id: 2 }, laterMissed),
+		];
+
+		const result = applyCoalesceOwnerPolicy(planned);
+
+		const exhausted = byId(result, 1);
+		expect(exhausted.plan.catchUpAt).toBeNull();
+		expect(exhausted.plan.occurrences).toEqual([]);
+		expect(exhausted.plan.nextRunAt).toBeNull();
+		expect(byId(result, 2).plan.catchUpAt).toEqual(laterMissed);
+	});
+
+	it.each([
+		['taskType', { taskType: 'trigger' }, { taskType: 'poll' }],
+		['misfireGraceSeconds', { misfireGraceSeconds: 60 }, { misfireGraceSeconds: 120 }],
+		['payload', { payload: { rule: 'one' } }, { payload: { rule: 'two' } }],
+	])('still groups siblings whose %s differs', (_label, first, second) => {
+		const planned = [
+			makeMember({ id: 1, ...first }, secondsBefore(NOW, 30)),
+			makeMember({ id: 2, ...second }, secondsBefore(NOW, 20)),
+		];
+
+		const result = applyCoalesceOwnerPolicy(planned);
+
+		expect(byId(result, 1).plan.catchUpAt).toBeNull();
+		expect(byId(result, 2).plan.catchUpAt).toEqual(secondsBefore(NOW, 20));
+	});
+
+	it('still groups siblings whose attempt limits differ', () => {
+		const planned = [
+			makeMember({ id: 1, maxAttempts: 1 }, secondsBefore(NOW, 30)),
+			makeMember({ id: 2, maxAttempts: 3 }, secondsBefore(NOW, 20)),
+		];
+
+		const result = applyCoalesceOwnerPolicy(planned);
+
+		expect(byId(result, 1).plan.catchUpAt).toBeNull();
+		expect(byId(result, 2).plan.catchUpAt).toEqual(secondsBefore(NOW, 20));
+	});
+
+	it('leaves a member with nothing missed alone and still groups the rest', () => {
+		const planned = [
+			makeMember({ id: 1 }, secondsBefore(NOW, 30), {
+				occurrences: [],
+				catchUpAt: null,
+				retireBefore: null,
+			}),
+			makeMember({ id: 2 }, secondsBefore(NOW, 20)),
+			makeMember({ id: 3 }, secondsBefore(NOW, 10)),
+		];
+
+		const result = applyCoalesceOwnerPolicy(planned);
+
+		const untouched = byId(result, 1);
+		expect(untouched.plan.catchUpAt).toBeNull();
+		expect(untouched.plan.occurrences).toEqual([]);
+		expect(untouched.plan.retireBefore).toBeNull();
+
+		expect(byId(result, 2).plan.catchUpAt).toBeNull();
+		expect(byId(result, 3).plan.catchUpAt).toEqual(secondsBefore(NOW, 10));
+	});
+
+	it('keeps a single late run even when every member of the owner is exhausted', () => {
+		const planned = [
+			makeMember({ id: 1 }, secondsBefore(NOW, 30), { nextRunAt: null }),
+			makeMember({ id: 2 }, secondsBefore(NOW, 20), { nextRunAt: null }),
+			makeMember({ id: 3 }, secondsBefore(NOW, 10), { nextRunAt: null }),
+		];
+
+		const result = applyCoalesceOwnerPolicy(planned);
+
+		const withLateRun = result.filter(({ plan }) => plan.catchUpAt !== null);
+		expect(withLateRun).toHaveLength(1);
+		expect(withLateRun[0].job.id).toBe(3);
+	});
+
+	it("lets an exhausted member hold the group's late run when its instant is the latest", () => {
+		const latest = secondsBefore(NOW, 10);
+		const planned = [
+			makeMember({ id: 1 }, secondsBefore(NOW, 30)),
+			makeMember({ id: 2 }, secondsBefore(NOW, 20)),
+			makeMember({ id: 3 }, latest, { nextRunAt: null }),
+		];
+
+		const result = applyCoalesceOwnerPolicy(planned);
+
+		const winner = byId(result, 3);
+		expect(winner.plan.catchUpAt).toEqual(latest);
+		expect(winner.plan.occurrences).toEqual([latest]);
+		expect(winner.plan.nextRunAt).toBeNull();
+		for (const id of [1, 2]) {
+			expect(byId(result, id).plan.catchUpAt).toBeNull();
+		}
+	});
+
+	it('groups only the owner-wide members when a sibling uses the per-job coalesce policy', () => {
+		const perJobMissed = secondsBefore(NOW, 30);
+		const planned = [
+			makeMember({ id: 1, misfirePolicy: ScheduledJobMisfirePolicy.Coalesce }, perJobMissed),
+			makeMember({ id: 2 }, secondsBefore(NOW, 20)),
+			makeMember({ id: 3 }, secondsBefore(NOW, 10)),
+		];
+
+		const result = applyCoalesceOwnerPolicy(planned);
+
+		const perJob = byId(result, 1);
+		expect(perJob.plan.catchUpAt).toEqual(perJobMissed);
+		expect(perJob.plan.occurrences).toEqual([perJobMissed]);
+
+		expect(byId(result, 2).plan.catchUpAt).toBeNull();
+		expect(byId(result, 3).plan.catchUpAt).toEqual(secondsBefore(NOW, 10));
+	});
+
+	it('groups a member whose occurrences were capped but still hold a late run', () => {
+		const cappedMissed = secondsBefore(NOW, 30);
+		const remaining = secondsBefore(NOW, 25);
+		const planned = [
+			makeMember({ id: 1 }, cappedMissed, {
+				occurrences: [cappedMissed, remaining],
+				nextRunAt: secondsBefore(NOW, 20),
+			}),
+			makeMember({ id: 2 }, secondsBefore(NOW, 10)),
+		];
+
+		const result = applyCoalesceOwnerPolicy(planned);
+
+		const capped = byId(result, 1);
+		expect(capped.plan.catchUpAt).toBeNull();
+		expect(capped.plan.occurrences).toEqual([remaining]);
+		expect(capped.plan.retireBefore).toEqual(cappedMissed);
+		expect(capped.plan.nextRunAt).toEqual(secondsBefore(NOW, 20));
+	});
+});

packages/@n8n/scheduler/src/core/materializer/__tests__/materialize.test.ts

@@ -24,6 +24,7 @@ const makeJob = (id: number): ScheduledJob => ({
 	maxAttempts: 1,
 	misfirePolicy: ScheduledJobMisfirePolicy.Coalesce,
 	misfireGraceSeconds: 60,
+	ownerKey: null,
 });
 
 const makeSkipJob = (id: number): ScheduledJob => ({
@@ -110,6 +111,7 @@ describe('materialize', () => {
 			occurrences: [NOW],
 			skippedOccurrences: 0,
 			catchUpAt: null,
+			retireBefore: null,
 			nextRunAt: new Date('2026-01-01T00:00:10.000Z'),
 			lastFiredAt: NOW,
 		};
@@ -412,6 +414,163 @@ describe('materialize', () => {
 		expect(onSkippedDuplicates).not.toHaveBeenCalled();
 	});
 
+	describe('siblings sharing an owner under the owner-wide coalesce policy', () => {
+		const BEHIND = new Date('2025-12-31T23:58:00.000Z');
+
+		const makeSibling = (id: number): ScheduledJob => ({
+			...makeJob(id),
+			misfirePolicy: ScheduledJobMisfirePolicy.CoalesceOwner,
+			ownerKey: 'owner-a',
+			nextRunAt: BEHIND,
+		});
+
+		const claimSiblings = (tx: MockProxy<MaterializerTransaction>) => {
+			tx.claimDueJobs.mockResolvedValue({
+				now: NOW,
+				jobs: [makeSibling(1), makeSibling(2), makeSibling(3)],
+			});
+			tx.recordOccurrences.mockResolvedValue({ recorded: 1, created: [] });
+		};
+
+		it('records one late run for the whole owner, retires every sibling, and advances every clock', async () => {
+			const tx = makeTx();
+			claimSiblings(tx);
+
+			await materialize(runnerWith(tx), options);
+
+			expect(tx.recordOccurrences).toHaveBeenCalledTimes(1);
+			const rows = tx.recordOccurrences.mock.calls[0][0];
+			expect(rows).toHaveLength(1);
+			expect(rows[0]).toMatchObject({ runAt: NOW, scheduledFor: NOW });
+
+			expect(tx.retireSuperseded).toHaveBeenCalledWith([
+				{ jobId: 1, before: NOW },
+				{ jobId: 2, before: NOW },
+				{ jobId: 3, before: NOW },
+			]);
+
+			const planned = tx.advanceJobs.mock.calls[0][0];
+			expect(planned).toHaveLength(3);
+			for (const { plan } of planned) {
+				expect(plan.nextRunAt).toEqual(new Date('2026-01-01T00:00:10.000Z'));
+				expect(plan.lastFiredAt).toEqual(NOW);
+			}
+		});
+
+		it('records the surviving occurrences of a superseded sibling at their own instants', async () => {
+			const tx = makeTx();
+			tx.claimDueJobs.mockResolvedValue({
+				now: NOW,
+				jobs: [
+					{ ...makeSibling(1), nextRunAt: new Date('2025-12-31T23:58:00.000Z') },
+					{ ...makeSibling(2), nextRunAt: new Date('2025-12-31T23:58:05.000Z') },
+				],
+			});
+			tx.recordOccurrences.mockResolvedValue({ recorded: 7, created: [] });
+
+			await materialize(runnerWith(tx), { ...options, windowSeconds: 30 });
+
+			const rows = tx.recordOccurrences.mock.calls[0][0];
+
+			const superseded = rows.filter((row) => row.jobId === 2);
+			expect(superseded.map((row) => row.scheduledFor)).toEqual([
+				new Date('2026-01-01T00:00:05.000Z'),
+				new Date('2026-01-01T00:00:15.000Z'),
+				new Date('2026-01-01T00:00:25.000Z'),
+			]);
+			for (const row of superseded) {
+				expect(row.runAt).toEqual(row.scheduledFor);
+			}
+
+			const winner = rows.filter((row) => row.jobId === 1);
+			expect(winner.map((row) => row.scheduledFor)).toEqual([
+				NOW,
+				new Date('2026-01-01T00:00:10.000Z'),
+				new Date('2026-01-01T00:00:20.000Z'),
+				new Date('2026-01-01T00:00:30.000Z'),
+			]);
+			expect(winner[0].runAt).toEqual(NOW);
+			for (const row of winner.slice(1)) {
+				expect(row.runAt).toEqual(row.scheduledFor);
+			}
+		});
+
+		it("a group's dropped late run does not itself count as a misfire when the member discarded no backlog", async () => {
+			const tx = makeTx();
+			const hourly = (id: number, nextRunAt: string): ScheduledJob => ({
+				...makeSibling(id),
+				intervalSeconds: 3600,
+				nextRunAt: new Date(nextRunAt),
+			});
+			tx.claimDueJobs.mockResolvedValue({
+				now: NOW,
+				jobs: [hourly(1, '2025-12-31T23:58:00.000Z'), hourly(2, '2025-12-31T23:57:00.000Z')],
+			});
+			tx.recordOccurrences.mockResolvedValue({ recorded: 1, created: [] });
+
+			const summary = await materialize(runnerWith(tx), options);
+
+			const planned = tx.advanceJobs.mock.calls[0][0];
+			const dropped = planned.find(({ job }) => job.id === 2)!;
+			expect(dropped.plan.skippedOccurrences).toBe(0);
+
+			expect(summary.misfires).toEqual([]);
+		});
+
+		it('counts a per-job policy discard in the same batch as a group, apart from the group', async () => {
+			const tx = makeTx();
+			tx.claimDueJobs.mockResolvedValue({
+				now: NOW,
+				jobs: [
+					makeSibling(1),
+					makeSibling(2),
+					{ ...makeJob(3), nextRunAt: BEHIND },
+					{ ...makeSkipJob(4), nextRunAt: BEHIND },
+				],
+			});
+			tx.recordOccurrences.mockResolvedValue({ recorded: 2, created: [] });
+
+			const summary = await materialize(runnerWith(tx), options);
+
+			expect(summary.misfires).toEqual([
+				{ taskType: 'test', policy: ScheduledJobMisfirePolicy.CoalesceOwner, discarded: 24 },
+				{ taskType: 'test', policy: ScheduledJobMisfirePolicy.Coalesce, discarded: 12 },
+				{ taskType: 'test', policy: ScheduledJobMisfirePolicy.Skip, discarded: 13 },
+			]);
+		});
+
+		it('records a single late run for the owner when none of their schedules has a further run', async () => {
+			const tx = makeTx();
+			const oneOff = (id: number, fireAt: string): ScheduledJob => ({
+				...makeSibling(id),
+				kind: 'one_off',
+				intervalSeconds: null,
+				fireAt: new Date(fireAt),
+				nextRunAt: new Date(fireAt),
+			});
+			tx.claimDueJobs.mockResolvedValue({
+				now: NOW,
+				jobs: [
+					oneOff(1, '2025-12-31T23:58:00.000Z'),
+					oneOff(2, '2025-12-31T23:58:10.000Z'),
+					oneOff(3, '2025-12-31T23:58:20.000Z'),
+				],
+			});
+			tx.recordOccurrences.mockResolvedValue({ recorded: 1, created: [] });
+
+			await materialize(runnerWith(tx), options);
+
+			const rows = tx.recordOccurrences.mock.calls[0][0];
+			expect(rows.map((row) => row.jobId)).toEqual([3]);
+			expect(rows[0].runAt).toEqual(NOW);
+			expect(tx.retireSuperseded).toHaveBeenCalledWith([
+				{ jobId: 1, before: new Date('2025-12-31T23:58:00.000Z') },
+				{ jobId: 2, before: new Date('2025-12-31T23:58:10.000Z') },
+				{ jobId: 3, before: new Date('2025-12-31T23:58:20.000Z') },
+			]);
+		});
+	});
+
 	it('still defers and completes the pass when the plan-error reporter itself throws', async () => {
 		const good = makeJob(1);
 		const bad: ScheduledJob = {

packages/@n8n/scheduler/src/core/materializer/__tests__/plan.property.test.ts

@@ -23,6 +23,7 @@ const makeIntervalJob = (intervalSeconds: number): ScheduledJob => ({
 	maxAttempts: 1,
 	misfirePolicy: ScheduledJobMisfirePolicy.Coalesce,
 	misfireGraceSeconds: 60,
+	ownerKey: null,
 });
 
 const makeCronJob = (cronExpression: string, timezone: string): ScheduledJob => ({
@@ -41,6 +42,7 @@ const makeCronJob = (cronExpression: string, timezone: string): ScheduledJob =>
 	maxAttempts: 1,
 	misfirePolicy: ScheduledJobMisfirePolicy.Coalesce,
 	misfireGraceSeconds: 60,
+	ownerKey: null,
 });
 
 const makeOneOffJob = (fireAt: Date): ScheduledJob => ({
@@ -61,6 +63,7 @@ const makeOneOffJob = (fireAt: Date): ScheduledJob => ({
 	maxAttempts: 1,
 	misfirePolicy: ScheduledJobMisfirePolicy.Coalesce,
 	misfireGraceSeconds: 60,
+	ownerKey: null,
 });
 
 /**

packages/@n8n/scheduler/src/core/materializer/__tests__/plan.test.ts

@@ -25,6 +25,7 @@ const makeJob = (overrides: Partial<ScheduledJob> = {}): ScheduledJob => ({
 	maxAttempts: 1,
 	misfirePolicy: ScheduledJobMisfirePolicy.Coalesce,
 	misfireGraceSeconds: 60,
+	ownerKey: null,
 	...overrides,
 });
 

packages/@n8n/scheduler/src/core/materializer/coalesce-group.ts

@@ -0,0 +1,82 @@
+import { ScheduledJobMisfirePolicy } from '@n8n/constants';
+
+import type { PlannedJob } from './transaction';
+
+interface GroupMember {
+	entry: PlannedJob;
+	latestMissedAt: number;
+}
+
+/**
+ * Applies `coalesce_owner` across one planning pass: per owner, only the job
+ * with the latest missed occurrence keeps its late run; the other jobs drop
+ * theirs. Jobs on another policy, without an owner, or with nothing missed
+ * are left untouched.
+ */
+export function applyCoalesceOwnerPolicy(planned: PlannedJob[]): PlannedJob[] {
+	const losers = new Set(
+		[...groupByOwner(planned).values()].flatMap((members) => {
+			const winner = members.reduce(latestThenLowestId);
+			return members.filter((member) => member !== winner).map((member) => member.entry);
+		}),
+	);
+
+	if (losers.size === 0) {
+		return planned;
+	}
+	return planned.map((entry) => (losers.has(entry) ? dropLateRun(entry) : entry));
+}
+
+/**
+ * Grouped by `ownerKey` alone: `coalesce_owner` keeps one late run across an
+ * owner's rules unconditionally, so the whole trigger fires once, however
+ * many rules missed occurrences. Per-rule deduplication isn't this policy's
+ * job: plain `coalesce` already keeps one late run per job, and
+ * distinguishing rules is a workflow-definition concern, not a misfire one.
+ */
+function groupByOwner(planned: PlannedJob[]): Map<string, GroupMember[]> {
+	return planned
+		.flatMap(toOwnedMember)
+		.reduce(
+			(groups, { key, member }) => groups.set(key, [...(groups.get(key) ?? []), member]),
+			new Map<string, GroupMember[]>(),
+		);
+}
+
+/** Empty when the policy does not touch the job: no owner, another policy, or nothing missed. */
+function toOwnedMember(entry: PlannedJob): Array<{ key: string; member: GroupMember }> {
+	const { job, plan } = entry;
+	if (
+		job.ownerKey === null ||
+		plan.catchUpAt === null ||
+		job.misfirePolicy !== ScheduledJobMisfirePolicy.CoalesceOwner
+	) {
+		return [];
+	}
+	return [{ key: job.ownerKey, member: { entry, latestMissedAt: plan.catchUpAt.getTime() } }];
+}
+
+function latestThenLowestId(current: GroupMember, candidate: GroupMember): GroupMember {
+	if (candidate.latestMissedAt !== current.latestMissedAt) {
+		return candidate.latestMissedAt > current.latestMissedAt ? candidate : current;
+	}
+	return candidate.entry.job.id < current.entry.job.id ? candidate : current;
+}
+
+/**
+ * Removes the loser's late run from its plan. `nextRunAt`/`lastFiredAt` stay
+ * as already planned: the loser's clock still has to advance past its
+ * backlog, or it reclaims the same backlog next pass. `retireBefore` also
+ * stays, so its already-recorded pending occurrence still retires even though
+ * it no longer runs late.
+ */
+function dropLateRun({ job, plan }: PlannedJob): PlannedJob {
+	return {
+		job,
+		plan: {
+			...plan,
+			occurrences: plan.occurrences.slice(1),
+			catchUpAt: null,
+		},
+	};
+}

packages/@n8n/scheduler/src/core/materializer/materialize.ts

@@ -1,6 +1,7 @@
 import { Time } from '@n8n/constants';
 
 import type { ScheduledJob } from '../types';
+import { applyCoalesceOwnerPolicy } from './coalesce-group';
 import { countMisfires, type MisfireCount } from './misfire';
 import { DEFAULT_MATERIALIZER_OPTIONS, type MaterializerOptions } from './options';
 import { planOccurrences } from './plan';
@@ -115,11 +116,12 @@ export async function materialize(
 				retiredOccurrences: 0,
 			};
 		}
-		const { occurrencesPlanned, numberOfJobsDeferred } = planOrDeferJobs(
+		const { occurrencesPlanned: jobsPlanned, numberOfJobsDeferred } = planOrDeferJobs(
 			claimed,
 			options,
 			hooks.onPlanError,
 		);
+		const occurrencesPlanned = applyCoalesceOwnerPolicy(jobsPlanned);
 		const rows = toNewOccurrences(occurrencesPlanned, claimed.now);
 		const { recorded, created } = await tx.recordOccurrences(rows);
 		signal?.throwIfAborted();
@@ -169,10 +171,14 @@ function toNewOccurrences(planned: PlannedJob[], now: Date): NewOccurrence[] {
 	);
 }
 
-/** The jobs whose plan recorded a catch-up run, paired with that instant. */
+/**
+ * Jobs whose plan set `retireBefore`: normally the instant of their own late
+ * run, but also a group's loser, whose dropped late run still leaves an
+ * already-recorded pending occurrence that needs retiring.
+ */
 function toSuperseded(planned: PlannedJob[]): SupersededOccurrences[] {
 	return planned.flatMap(({ job, plan }) =>
-		plan.catchUpAt === null ? [] : [{ jobId: job.id, before: plan.catchUpAt }],
+		plan.retireBefore === null ? [] : [{ jobId: job.id, before: plan.retireBefore }],
 	);
 }
 
@@ -228,6 +234,7 @@ function planOrDeferJob(
 					occurrences: [],
 					skippedOccurrences: 0,
 					catchUpAt: null,
+					retireBefore: null,
 					nextRunAt: retryAt,
 					lastFiredAt: job.lastFiredAt,
 				},

packages/@n8n/scheduler/src/core/materializer/misfire.ts

@@ -41,7 +41,8 @@ export function applyMisfirePolicy(
 	}
 
 	switch (job.misfirePolicy) {
-		case ScheduledJobMisfirePolicy.Coalesce: {
+		case ScheduledJobMisfirePolicy.Coalesce:
+		case ScheduledJobMisfirePolicy.CoalesceOwner: {
 			if (truncated && ahead.length === 0) return { occurrences: [], catchUpAt: null };
 			const catchUpAt = behind[behind.length - 1];
 			return { occurrences: [catchUpAt, ...ahead], catchUpAt };

packages/@n8n/scheduler/src/core/materializer/plan.ts

@@ -24,10 +24,17 @@ export interface OccurrencePlan {
 
 	/**
 	 * The one occurrence a `coalesce` policy kept to stand in for a backlog, always the
-	 * first of {@link occurrences}. `null` when no policy applied.
+	 * first of {@link occurrences}. `null` when no policy applied, or a sibling in
+	 * its owner group won instead.
 	 */
 	catchUpAt: Date | null;
 
+	/**
+	 * The instant before which this job's already-recorded pending occurrences
+	 * should be retired.
+	 */
+	retireBefore: Date | null;
+
 	/**
 	 * The next instant not yet recorded,
 	 * or `null` once the schedule has no more fires.
@@ -56,6 +63,7 @@ export function planOccurrences(
 			occurrences: [],
 			skippedOccurrences: 0,
 			catchUpAt: null,
+			retireBefore: null,
 			nextRunAt: null,
 			lastFiredAt: job.lastFiredAt,
 		};
@@ -78,6 +86,7 @@ export function planOccurrences(
 		occurrences,
 		skippedOccurrences: due.length - occurrences.length,
 		catchUpAt,
+		retireBefore: catchUpAt,
 		nextRunAt: fire.done ? null : fire.value,
 		// Tracks every instant this pass consumed, not only the recorded ones: a
 		// discarded occurrence is still one the schedule has moved past.

packages/@n8n/scheduler/src/core/recurrence/__tests__/resolve.test.ts

@@ -22,6 +22,7 @@ function makeJob(overrides: Partial<ScheduledJob> = {}): ScheduledJob {
 		maxAttempts: 1,
 		misfirePolicy: 'coalesce',
 		misfireGraceSeconds: 60,
+		ownerKey: null,
 		...overrides,
 	};
 }

packages/@n8n/scheduler/src/core/types/schedule.ts

@@ -91,4 +91,10 @@ export interface ScheduledJob {
 	misfirePolicy: ScheduledJobMisfirePolicy;
 	/** How late an occurrence may be before {@link misfirePolicy} applies to it. */
 	misfireGraceSeconds: number;
+	/**
+	 * Groups jobs for `coalesce_owner`: same value, same group. The scheduler
+	 * only compares values, it never reads them. `null` means the job stands
+	 * alone. Required so an adapter that forgets to map it fails to compile.
+	 */
+	ownerKey: string | null;
 }

packages/cli/src/scheduling/__tests__/durable-job-provisioner.test.ts

@@ -1,5 +1,6 @@
 import type { Logger } from '@n8n/backend-common';
 import type { GlobalConfig } from '@n8n/config';
+import { ScheduledJobMisfirePolicy } from '@n8n/constants';
 import type {
 	DataSource,
 	ScheduledJob,
@@ -9,7 +10,6 @@ import type {
 import type { DesiredJob, ProvisionSummary, ScheduleDefinition } from '@n8n/scheduler';
 import type { EntityManager } from '@n8n/typeorm';
 import type { Tracing } from 'n8n-core';
-import { ScheduledJobMisfirePolicy } from '@n8n/constants';
 import { mock } from 'vitest-mock-extended';
 
 import { DurableJobProvisioner } from '../durable-job-provisioner';

packages/cli/src/scheduling/__tests__/owner-key.test.ts

@@ -0,0 +1,53 @@
+import { ownerKeyFor, withOwnerKeys } from '../owner-key';
+
+const key = (workflowId: string, nodeId: string) => `${workflowId}\0${nodeId}`;
+
+describe('ownerKeyFor', () => {
+	it.each([
+		['both ids present', 'wf-1', 'node-1', key('wf-1', 'node-1')],
+		['workflowId is null', null, 'node-1', null],
+		['nodeId is null', 'wf-1', null, null],
+		['both are null', null, null, null],
+	])('%s', (_case, workflowId, nodeId, expected) => {
+		expect(ownerKeyFor({ workflowId, nodeId })).toBe(expected);
+	});
+
+	it('distinguishes owners whose ids differ only in where the boundary falls', () => {
+		expect(ownerKeyFor({ workflowId: 'wf-1:node', nodeId: '1' })).not.toBe(
+			ownerKeyFor({ workflowId: 'wf-1', nodeId: 'node:1' }),
+		);
+	});
+
+	it('does not derive the key from the job name', () => {
+		const first = { workflowId: 'wf-1', nodeId: 'node-1', name: 'wf-1:node-1:0:fingerprint-a' };
+		const second = { workflowId: 'wf-1', nodeId: 'node-1', name: 'wf-1:node-1:1:fingerprint-b' };
+
+		expect(ownerKeyFor(first)).toBe(ownerKeyFor(second));
+		expect(ownerKeyFor(first)).not.toBe(first.name);
+		expect(ownerKeyFor(first)).not.toContain('fingerprint-a');
+	});
+});
+
+describe('withOwnerKeys', () => {
+	it('stamps each job with its owner key, keeping the claim clock', () => {
+		const now = new Date('2026-01-01T00:00:00.000Z');
+
+		const result = withOwnerKeys({
+			now,
+			jobs: [
+				{ id: 1, workflowId: 'wf-1', nodeId: 'node-1' },
+				{ id: 2, workflowId: 'wf-1', nodeId: 'node-1' },
+				{ id: 3, workflowId: 'wf-2', nodeId: 'node-9' },
+				{ id: 4, workflowId: null, nodeId: null },
+			],
+		});
+
+		expect(result.now).toBe(now);
+		expect(result.jobs).toEqual([
+			{ id: 1, workflowId: 'wf-1', nodeId: 'node-1', ownerKey: key('wf-1', 'node-1') },
+			{ id: 2, workflowId: 'wf-1', nodeId: 'node-1', ownerKey: key('wf-1', 'node-1') },
+			{ id: 3, workflowId: 'wf-2', nodeId: 'node-9', ownerKey: key('wf-2', 'node-9') },
+			{ id: 4, workflowId: null, nodeId: null, ownerKey: null },
+		]);
+	});
+});

packages/cli/src/scheduling/durable-job-provisioner.ts

@@ -19,6 +19,7 @@ import type {
 import { Tracing } from 'n8n-core';
 import { UnexpectedError } from 'n8n-workflow';
 
+import { withOwnerKeys } from './owner-key';
 import { createSchedulerTracer } from './scheduler-tracer';
 
 /**
@@ -319,7 +320,9 @@ export class DurableJobProvisioner {
 					const claimed = (await this.jobs.findManyByIds(manager, [...jobIds])).filter(
 						(job) => job.enabled && job.nextRunAt !== null,
 					);
-					return claimed.length > 0 ? { now, jobs: claimed } : undefined;
+					// Grouping never triggers here: every seeded job starts from a freshly
+					// computed `nextRunAt`, so none of them has missed anything yet.
+					return claimed.length > 0 ? withOwnerKeys({ now, jobs: claimed }) : undefined;
 				},
 				recordOccurrences: async (occurrences) =>
 					await this.tasks.insertIgnoringDuplicates(manager, occurrences),

packages/cli/src/scheduling/durable-scheduler.ts

@@ -13,6 +13,7 @@ import { InstanceSettings, Tracing } from 'n8n-core';
 
 import { PrometheusSchedulerMetricsService } from '@/metrics/prometheus/scheduler-metrics.service';
 
+import { withOwnerKeys } from './owner-key';
 import { PollTriggerTaskHandler } from './poll-trigger-node/poll-trigger-task-handler';
 import { ScheduleTriggerTaskHandler } from './schedule-trigger-node/schedule-trigger-task-handler';
 import { createSchedulerTracer } from './scheduler-tracer';
@@ -170,8 +171,10 @@ export function buildMaterializerTransaction(
 		await dataSource.transaction(
 			async (manager) =>
 				await work({
-					claimDueJobs: async (limit, lookaheadMs) =>
-						await jobs.claimDue(manager, limit, lookaheadMs),
+					claimDueJobs: async (limit, lookaheadMs) => {
+						const claimed = await jobs.claimDue(manager, limit, lookaheadMs);
+						return claimed === undefined ? undefined : withOwnerKeys(claimed);
+					},
 					recordOccurrences: async (occurrences) =>
 						await tasks.insertIgnoringDuplicates(manager, occurrences),
 					retireSuperseded: async (superseded) => await tasks.updateToMissed(manager, superseded),

packages/cli/src/scheduling/owner-key.ts

@@ -0,0 +1,27 @@
+import type { ScheduledJob as ScheduledJobEntity } from '@n8n/db';
+
+type JobOwner = Pick<ScheduledJobEntity, 'workflowId' | 'nodeId'>;
+
+type OwnedJob<T extends JobOwner> = T & { ownerKey: string | null };
+
+/**
+ * `null` when either id is missing. Derived only from `workflowId`/`nodeId`,
+ * never the job's `name`: that carries the per-rule fingerprint, and using it
+ * would make every group a singleton.
+ */
+export function ownerKeyFor(job: JobOwner): string | null {
+	if (job.workflowId === null || job.nodeId === null) {
+		return null;
+	}
+	return `${job.workflowId}\0${job.nodeId}`;
+}
+
+export function withOwnerKeys<T extends JobOwner>(claimed: {
+	now: Date;
+	jobs: T[];
+}): { now: Date; jobs: Array<OwnedJob<T>> } {
+	return {
+		now: claimed.now,
+		jobs: claimed.jobs.map((job) => ({ ...job, ownerKey: ownerKeyFor(job) })),
+	};
+}

packages/cli/src/scheduling/poll-trigger-node/__tests__/poll-trigger-job-registrar.test.ts

@@ -1,6 +1,6 @@
 /* eslint-disable @typescript-eslint/unbound-method */
-import { ScheduledJobMisfirePolicy } from '@n8n/constants';
 import { mockLogger } from '@n8n/backend-test-utils';
+import { ScheduledJobMisfirePolicy } from '@n8n/constants';
 import type { GlobalConfig } from '@n8n/config';
 import type { EntityManager } from '@n8n/db';
 import type { CronDefinition } from '@n8n/scheduler';

packages/cli/src/scheduling/schedule-trigger-node/__tests__/schedule-trigger-job-registrar.test.ts

@@ -1,6 +1,6 @@
 /* eslint-disable @typescript-eslint/unbound-method */
-import { ScheduledJobMisfirePolicy } from '@n8n/constants';
 import type { Logger } from '@n8n/backend-common';
+import { ScheduledJobMisfirePolicy } from '@n8n/constants';
 import { mockLogger } from '@n8n/backend-test-utils';
 import type { GlobalConfig, WorkflowsConfig } from '@n8n/config';
 import type { EntityManager } from '@n8n/db';
@@ -187,6 +187,24 @@ describe('ScheduleTriggerJobRegistrar', () => {
 			);
 		});
 
+		it('provisions a multi-rule node in one call, under the skip policy', async () => {
+			const session = makeRegistrar().createSession();
+			const collector = session.createCollector(workflow, scheduleNode);
+			collector.registerCron(dailyAtNine, vi.fn());
+			collector.registerCron(everyThreeWeeksMonday, vi.fn());
+			collector.registerCron(
+				{ expression: '0 30 9 * * *' as CronExpression, recurrence: { activated: false } },
+				vi.fn(),
+			);
+
+			await session.commit(WORKFLOW_ID, NODE_ID);
+
+			expect(jobProvisioner.provision).toHaveBeenCalledTimes(1);
+			const [, , , , desired, misfirePolicy] = jobProvisioner.provision.mock.calls.at(-1)!;
+			expect(desired).toHaveLength(3);
+			expect(misfirePolicy).toBe(ScheduledJobMisfirePolicy.Skip);
+		});
+
 		it('provisions a 5-field custom cron (no seconds) and plans its first fire', async () => {
 			const session = makeRegistrar().createSession();
 			const collector = session.createCollector(workflow, scheduleNode);
@@ -392,12 +410,26 @@ describe('ScheduleTriggerJobRegistrar', () => {
 		it.each<[string, INodeParameters | undefined, ScheduledJobMisfirePolicy]>([
 			['1.4', undefined, ScheduledJobMisfirePolicy.Skip],
 			['1.4', { misfirePolicy: 'coalesce' }, ScheduledJobMisfirePolicy.Coalesce],
+			['1.4', { misfirePolicy: 'coalesce_owner' }, ScheduledJobMisfirePolicy.CoalesceOwner],
 			['1.4', { misfirePolicy: 'skip' }, ScheduledJobMisfirePolicy.Skip],
 			['1.3', undefined, ScheduledJobMisfirePolicy.Skip],
 			['1.4', { misfirePolicy: 'nonsense' } as INodeParameters, ScheduledJobMisfirePolicy.Skip],
+			['1.4', { misfirePolicy: 'wrong' } as INodeParameters, ScheduledJobMisfirePolicy.Skip],
+			['1.4', { misfirePolicy: '' } as INodeParameters, ScheduledJobMisfirePolicy.Skip],
 			['1.4', { misfirePolicy: 'Coalesce' } as INodeParameters, ScheduledJobMisfirePolicy.Skip],
 			['1.4', { misfirePolicy: ' coalesce' } as INodeParameters, ScheduledJobMisfirePolicy.Skip],
 			['1.4', { misfirePolicy: 'coalesce ' } as INodeParameters, ScheduledJobMisfirePolicy.Skip],
+			[
+				'1.4',
+				{ misfirePolicy: 'CoalesceOwner' } as INodeParameters,
+				ScheduledJobMisfirePolicy.Skip,
+			],
+			[
+				'1.4',
+				{ misfirePolicy: 'coalesce_owner ' } as INodeParameters,
+				ScheduledJobMisfirePolicy.Skip,
+			],
+			['1.4', { misfirePolicy: '__proto__' } as INodeParameters, ScheduledJobMisfirePolicy.Skip],
 		])(
 			'resolves misfirePolicy %s with parameters %s to %s',
 			async (typeVersionLabel, parameters, expected) => {

packages/cli/src/scheduling/schedule-trigger-node/schedule-trigger-job-registrar.ts

@@ -320,6 +320,8 @@ export class ScheduleTriggerJobRegistrar {
 		});
 
 		const payload: ScheduleTriggerTaskPayload = { workflowId, nodeId };
+		// `skip` matches the legacy engine, which never runs a missed occurrence
+		// late. Running late is a per-node opt-in (see `resolveMisfirePolicy`).
 		const summary = await this.jobProvisioner.provision(
 			workflowId,
 			nodeId,
@@ -421,15 +423,20 @@ function withResolvedTimezone(schedule: Schedule, defaultTimezone: string): Sche
 }
 
 /**
- * Anything other than an explicit `coalesce` resolves to skipping, so an
+ * Any value other than an explicit policy resolves to skipping, so an
  * unrecognised value does not fail the activation. No `typeVersion` check is
  * needed: `Workflow`'s constructor drops a parameter its `displayOptions` hide,
  * so a node older than the option cannot arrive carrying it.
  */
 function resolveMisfirePolicy(node: INode): ScheduledJobMisfirePolicy {
-	return node.parameters?.misfirePolicy === ScheduledJobMisfirePolicy.Coalesce
-		? ScheduledJobMisfirePolicy.Coalesce
-		: ScheduledJobMisfirePolicy.Skip;
+	switch (node.parameters?.misfirePolicy) {
+		case 'coalesce':
+			return ScheduledJobMisfirePolicy.Coalesce;
+		case 'coalesce_owner':
+			return ScheduledJobMisfirePolicy.CoalesceOwner;
+		default:
+			return ScheduledJobMisfirePolicy.Skip;
+	}
 }
 
 function resolveMisfireGraceSeconds(

packages/cli/test/integration/scheduling/scheduler-materialize.test.ts

@@ -1,13 +1,15 @@
-import { testDb } from '@n8n/backend-test-utils';
+import { createWorkflowWithHistory, testDb } from '@n8n/backend-test-utils';
 import {
 	DataSource,
 	type ScheduledJob,
 	ScheduledJobRepository,
 	ScheduledTaskRepository,
+	WorkflowPublishedVersionRepository,
 } from '@n8n/db';
 import { Container } from '@n8n/di';
 import { createScheduler, totalDiscarded } from '@n8n/scheduler';
 import type { SchedulerDeps } from '@n8n/scheduler';
+import { v4 as uuid } from 'uuid';
 
 import { buildMaterializerTransaction } from '@/scheduling/durable-scheduler';
 
@@ -370,4 +372,78 @@ describe('scheduler materialization', () => {
 			expect(job.nextRunAt!.getTime() - job.lastFiredAt!.getTime()).toBe(3600 * 1000);
 		}
 	});
+
+	describe('rules of one node under the owner-wide coalesce policy', () => {
+		let workflowId: string;
+		let nodeId: string;
+
+		const createRule = async () =>
+			await createJob({
+				workflowId,
+				nodeId,
+				intervalSeconds: 3600,
+				misfirePolicy: 'coalesce_owner',
+				misfireGraceSeconds: 3600,
+				nextRunAt: secondsFromNow(3600),
+			});
+
+		beforeEach(async () => {
+			const workflow = await createWorkflowWithHistory({ active: true });
+			await Container.get(WorkflowPublishedVersionRepository).setPublishedVersion(
+				workflow.id,
+				workflow.versionId,
+			);
+			workflowId = workflow.id;
+			nodeId = uuid();
+		});
+
+		it('leaves one catch-up run pending and retires the occurrences it supersedes', async () => {
+			const rules = [await createRule(), await createRule(), await createRule()];
+			await jobRepo.backdateNextRunAt(workflowId, nodeId, 200);
+
+			const firstPass = await runMaterialization(0);
+			expect(firstPass).toMatchObject({ claimedJobs: 3, occurrences: 3 });
+			const queued = await taskRepo.find();
+			expect(queued).toHaveLength(3);
+			expect(queued.every((task) => task.status === 'pending')).toBe(true);
+
+			await jobRepo.update({ workflowId, nodeId }, { misfireGraceSeconds: 30 });
+			await jobRepo.backdateNextRunAt(workflowId, nodeId, 100);
+
+			const secondPass = await runMaterialization(0);
+
+			expect(secondPass).toMatchObject({
+				claimedJobs: 3,
+				occurrences: 1,
+				retiredOccurrences: 3,
+			});
+
+			const tasks = await taskRepo.find();
+			const pending = tasks.filter((task) => task.status === 'pending');
+			expect(pending).toHaveLength(1);
+			expect(pending[0].jobId).toBe(Math.min(...rules.map((rule) => rule.id)));
+			expect(pending[0].runAt.getTime()).toBeGreaterThan(pending[0].scheduledFor.getTime());
+
+			const missedIds = tasks
+				.filter((task) => task.status === 'missed')
+				.map((task) => task.id)
+				.sort();
+			expect(missedIds).toEqual(queued.map((task) => task.id).sort());
+		});
+
+		it('advances every rule clock even though only one catch-up run was recorded', async () => {
+			const rules = [await createRule(), await createRule(), await createRule()];
+			await jobRepo.update({ workflowId, nodeId }, { misfireGraceSeconds: 30 });
+			await jobRepo.backdateNextRunAt(workflowId, nodeId, 100);
+
+			await runMaterialization(0);
+
+			for (const rule of rules) {
+				const advanced = await jobRepo.findOneByOrFail({ id: rule.id });
+				expect(advanced.nextRunAt!.getTime()).toBeGreaterThan(Date.now());
+				expect(advanced.lastFiredAt).not.toBeNull();
+			}
+			expect(await taskRepo.count()).toBe(1);
+		});
+	});
 });

packages/cli/test/migration/1786666615643-add-coalesce-owner-misfire-policy.test.ts

@@ -0,0 +1,262 @@
+import {
+	createTestMigrationContext,
+	initDbUpToMigration,
+	runSingleMigration,
+	undoLastSingleMigration,
+	type TestMigrationContext,
+} from '@n8n/backend-test-utils';
+import { DbConnection } from '@n8n/db';
+import { Container } from '@n8n/di';
+import { DataSource } from '@n8n/typeorm';
+
+const MIGRATION_NAME = 'AddCoalesceOwnerMisfirePolicy1786666615643';
+
+const SCHEDULE_TRIGGER_TASK_TYPE = 'workflow:schedule-trigger';
+const SYSTEM_TASK_TYPE = 'system:prune-executions';
+
+describe('AddCoalesceOwnerMisfirePolicy Migration', () => {
+	let dataSource: DataSource;
+
+	beforeAll(async () => {
+		const dbConnection = Container.get(DbConnection);
+		await dbConnection.init();
+		dataSource = Container.get(DataSource);
+	});
+
+	beforeEach(async () => {
+		const context = createTestMigrationContext(dataSource);
+		await context.queryRunner.clearDatabase();
+		await context.queryRunner.release();
+		await initDbUpToMigration(MIGRATION_NAME);
+	});
+
+	afterAll(async () => {
+		const dbConnection = Container.get(DbConnection);
+		await dbConnection.close();
+	});
+
+	async function insertJob(
+		context: TestMigrationContext,
+		name: string,
+		taskType: string,
+		misfirePolicy: string,
+	) {
+		await context.runQuery(
+			`INSERT INTO ${context.escape.tableName('scheduled_job')}
+			   ("name", "kind", "intervalSeconds", "taskType", "misfirePolicy", "createdAt", "updatedAt")
+			 VALUES ('${name}', 'interval', 60, '${taskType}', '${misfirePolicy}', CURRENT_TIMESTAMP, CURRENT_TIMESTAMP)`,
+		);
+	}
+
+	async function insertTaskFor(context: TestMigrationContext, jobName: string) {
+		const [job] = (await context.queryRunner.query(
+			`SELECT "id" FROM ${context.escape.tableName('scheduled_job')} WHERE "name" = '${jobName}'`,
+		)) as Array<{ id: number }>;
+		await context.runQuery(
+			`INSERT INTO ${context.escape.tableName('scheduled_task')}
+			   ("jobId", "taskType", "scheduledFor", "runAt", "createdAt")
+			 VALUES (${job.id}, 'test', CURRENT_TIMESTAMP, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP)`,
+		);
+	}
+
+	async function policies(context: TestMigrationContext): Promise<Record<string, string>> {
+		const rows = (await context.queryRunner.query(
+			`SELECT "name", "misfirePolicy" FROM ${context.escape.tableName('scheduled_job')}`,
+		)) as Array<{ name: string; misfirePolicy: string }>;
+		return Object.fromEntries(rows.map((row) => [row.name, row.misfirePolicy]));
+	}
+
+	async function countRows(context: TestMigrationContext, table: string): Promise<number> {
+		const [{ count }] = (await context.queryRunner.query(
+			`SELECT COUNT(*) AS "count" FROM ${context.escape.tableName(table)}`,
+		)) as Array<{ count: number | string }>;
+		return Number(count);
+	}
+
+	async function taskJobNames(context: TestMigrationContext): Promise<string[]> {
+		const rows = (await context.queryRunner.query(
+			`SELECT "job"."name" AS "name"
+			   FROM ${context.escape.tableName('scheduled_task')} "task"
+			   INNER JOIN ${context.escape.tableName('scheduled_job')} "job" ON "job"."id" = "task"."jobId"`,
+		)) as Array<{ name: string }>;
+		return rows.map((row) => row.name).sort();
+	}
+
+	async function foreignKeyViolations(context: TestMigrationContext): Promise<unknown[]> {
+		if (!context.isSqlite) return [];
+		return (await context.queryRunner.query(
+			`PRAGMA foreign_key_check(${context.escape.tableName('scheduled_task')})`,
+		)) as unknown[];
+	}
+
+	async function columnNames(context: TestMigrationContext, table: string): Promise<string[]> {
+		if (context.isSqlite) {
+			const rows = (await context.queryRunner.query(
+				`PRAGMA table_info(${context.escape.tableName(table)})`,
+			)) as Array<{ name: string }>;
+			return rows.map((row) => row.name);
+		}
+		const rows = (await context.queryRunner.query(
+			'SELECT column_name FROM information_schema.columns WHERE table_name = $1',
+			[`${context.tablePrefix}${table}`],
+		)) as Array<{ column_name: string }>;
+		return rows.map((row) => row.column_name);
+	}
+
+	async function jobIndexDefinitions(context: TestMigrationContext): Promise<string[]> {
+		const table = `${context.tablePrefix}scheduled_job`;
+		if (context.isSqlite) {
+			const rows = (await context.queryRunner.query(
+				`SELECT sql FROM sqlite_master WHERE type = 'index' AND tbl_name = '${table}'`,
+			)) as Array<{ sql: string | null }>;
+			return rows.map((row) => row.sql ?? '');
+		}
+		const rows = (await context.queryRunner.query(
+			'SELECT indexdef FROM pg_indexes WHERE tablename = $1',
+			[table],
+		)) as Array<{ indexdef: string }>;
+		return rows.map((row) => row.indexdef);
+	}
+
+	async function seedJobsAndTasks() {
+		const context = createTestMigrationContext(dataSource);
+		await insertJob(context, 'trigger_a', SCHEDULE_TRIGGER_TASK_TYPE, 'coalesce');
+		await insertJob(context, 'trigger_b', SCHEDULE_TRIGGER_TASK_TYPE, 'coalesce');
+		await insertJob(context, 'trigger_skip', SCHEDULE_TRIGGER_TASK_TYPE, 'skip');
+		await insertJob(context, 'system_job', SYSTEM_TASK_TYPE, 'coalesce');
+		await insertJob(context, 'system_skip', SYSTEM_TASK_TYPE, 'skip');
+		await insertTaskFor(context, 'trigger_a');
+		await insertTaskFor(context, 'system_job');
+		await context.queryRunner.release();
+	}
+
+	describe('up', () => {
+		it('accepts coalesce_owner and still rejects an unknown policy', async () => {
+			await runSingleMigration(MIGRATION_NAME);
+			const context = createTestMigrationContext(dataSource);
+
+			await expect(
+				insertJob(context, 'owner', SCHEDULE_TRIGGER_TASK_TYPE, 'coalesce_owner'),
+			).resolves.not.toThrow();
+			await expect(
+				insertJob(context, 'garbage', SCHEDULE_TRIGGER_TASK_TYPE, 'not_a_policy'),
+			).rejects.toThrow();
+
+			await context.queryRunner.release();
+		});
+
+		it('keeps queued occurrences pointing at the jobs they belonged to', async () => {
+			await seedJobsAndTasks();
+
+			await runSingleMigration(MIGRATION_NAME);
+
+			const context = createTestMigrationContext(dataSource);
+			expect(await countRows(context, 'scheduled_task')).toBe(2);
+			expect(await taskJobNames(context)).toEqual(['system_job', 'trigger_a']);
+			expect(await foreignKeyViolations(context)).toEqual([]);
+			await context.queryRunner.release();
+		});
+
+		it('keeps every job row, the other CHECK constraints and the indexes', async () => {
+			await seedJobsAndTasks();
+
+			await runSingleMigration(MIGRATION_NAME);
+
+			const context = createTestMigrationContext(dataSource);
+			expect(await countRows(context, 'scheduled_job')).toBe(5);
+			expect(await columnNames(context, 'scheduled_job')).toEqual(
+				expect.arrayContaining([
+					'misfirePolicy',
+					'misfireGraceSeconds',
+					'recurrenceUnit',
+					'recurrenceSize',
+				]),
+			);
+			expect(await columnNames(context, 'scheduled_task')).toContain('missedAfter');
+
+			await expect(
+				context.runQuery(
+					`INSERT INTO ${context.escape.tableName('scheduled_job')}
+					   ("name", "kind", "intervalSeconds", "taskType", "createdAt", "updatedAt")
+					 VALUES ('bad_kind', 'nonsense', 60, 'test', CURRENT_TIMESTAMP, CURRENT_TIMESTAMP)`,
+				),
+			).rejects.toThrow();
+			await expect(
+				context.runQuery(
+					`INSERT INTO ${context.escape.tableName('scheduled_job')}
+					   ("name", "kind", "intervalSeconds", "taskType", "misfireGraceSeconds", "createdAt", "updatedAt")
+					 VALUES ('bad_grace', 'interval', 60, 'test', 0, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP)`,
+				),
+			).rejects.toThrow();
+
+			const definitions = await jobIndexDefinitions(context);
+			expect(definitions.some((sql) => sql.includes('nextRunAt'))).toBe(true);
+			expect(definitions.some((sql) => sql.includes('workflowId'))).toBe(true);
+			expect(definitions.some((sql) => sql.includes('name'))).toBe(true);
+
+			await context.queryRunner.release();
+		});
+
+		it('moves schedule trigger jobs from coalesce to skip, leaving other task types alone', async () => {
+			await seedJobsAndTasks();
+
+			await runSingleMigration(MIGRATION_NAME);
+
+			const context = createTestMigrationContext(dataSource);
+			expect(await policies(context)).toEqual({
+				trigger_a: 'skip',
+				trigger_b: 'skip',
+				trigger_skip: 'skip',
+				system_job: 'coalesce',
+				system_skip: 'skip',
+			});
+			await context.queryRunner.release();
+		});
+	});
+
+	describe('down', () => {
+		it('moves every trigger policy back to coalesce without deleting a job or a task', async () => {
+			await seedJobsAndTasks();
+			await runSingleMigration(MIGRATION_NAME);
+			// Simulates a job a user later moved onto coalesce_owner.
+			const seeded = createTestMigrationContext(dataSource);
+			await insertJob(seeded, 'trigger_owner', SCHEDULE_TRIGGER_TASK_TYPE, 'coalesce_owner');
+			await seeded.queryRunner.release();
+
+			await undoLastSingleMigration();
+
+			const context = createTestMigrationContext(dataSource);
+			expect(await policies(context)).toEqual({
+				trigger_a: 'coalesce',
+				trigger_b: 'coalesce',
+				// Seeded as skip before up(), so it cannot be told apart from the
+				// rows up() moved to skip; down() folds it to coalesce with them.
+				trigger_skip: 'coalesce',
+				trigger_owner: 'coalesce',
+				system_job: 'coalesce',
+				// Not a schedule trigger, so neither direction touches it.
+				system_skip: 'skip',
+			});
+			expect(await countRows(context, 'scheduled_job')).toBe(6);
+			expect(await countRows(context, 'scheduled_task')).toBe(2);
+			expect(await taskJobNames(context)).toEqual(['system_job', 'trigger_a']);
+			expect(await foreignKeyViolations(context)).toEqual([]);
+			await context.queryRunner.release();
+		});
+
+		it('narrows the CHECK back so coalesce_owner is rejected', async () => {
+			await runSingleMigration(MIGRATION_NAME);
+			await undoLastSingleMigration();
+
+			const context = createTestMigrationContext(dataSource);
+			await expect(
+				insertJob(context, 'owner', SCHEDULE_TRIGGER_TASK_TYPE, 'coalesce_owner'),
+			).rejects.toThrow();
+			await expect(
+				insertJob(context, 'plain', SCHEDULE_TRIGGER_TASK_TYPE, 'coalesce'),
+			).resolves.not.toThrow();
+
+			await context.queryRunner.release();
+		});
+	});
+});

packages/nodes-base/nodes/Schedule/ScheduleTrigger.node.ts

@@ -433,7 +433,8 @@ export class ScheduleTrigger implements INodeType {
 				type: 'options',
 				default: 'skip',
 				options: [
-					{ name: 'Run the Most Recent Missed Execution', value: 'coalesce' },
+					{ name: 'Run the Most Recent Missed Execution Per Rule', value: 'coalesce' },
+					{ name: 'Run the Most Recent Missed Execution', value: 'coalesce_owner' },
 					{ name: "Don't Run Missed Executions", value: 'skip' },
 				],
 				hint: 'Applies once an execution is later than the grace period set below',

packages/nodes-base/nodes/Schedule/test/ScheduleTrigger.node.test.ts

@@ -25,7 +25,7 @@ describe('ScheduleTrigger', () => {
 			expect(node.description.version).toContain(1.4);
 		});
 
-		it('defines misfirePolicy as a node setting offering exactly two values, defaulting to skip', () => {
+		it('defines misfirePolicy as a node setting offering exactly three values, defaulting to skip', () => {
 			const misfirePolicy = properties.find((property) => property.name === 'misfirePolicy');
 
 			expect(misfirePolicy).toMatchObject({
@@ -34,7 +34,8 @@ describe('ScheduleTrigger', () => {
 				isNodeSetting: true,
 				noDataExpression: true,
 				options: [
-					{ name: 'Run the Most Recent Missed Execution', value: 'coalesce' },
+					{ name: 'Run the Most Recent Missed Execution Per Rule', value: 'coalesce' },
+					{ name: 'Run the Most Recent Missed Execution', value: 'coalesce_owner' },
 					{ name: "Don't Run Missed Executions", value: 'skip' },
 				],
 			});