fix(core): Close the durable scheduler's record-then-dispatch effect boundary (#34014)

Co-authored-by: Emilia <100027345+sovietspaceship@users.noreply.github.com>
Co-authored-by: Claude Opus 4.8 <noreply@anthropic.com>

docs/generated/postgres-schema/README.md

@@ -94,7 +94,7 @@ Auto-generated from the PostgreSQL migrations in @n8n/db. Do not edit by hand.
 | [public.role_mapping_rule_project](public.role_mapping_rule_project.md) | 2 |  | BASE TABLE |
 | [public.role_scope](public.role_scope.md) | 2 |  | BASE TABLE |
 | [public.scheduled_job](public.scheduled_job.md) | 19 |  | BASE TABLE |
-| [public.scheduled_task](public.scheduled_task.md) | 16 |  | BASE TABLE |
+| [public.scheduled_task](public.scheduled_task.md) | 17 |  | BASE TABLE |
 | [public.scope](public.scope.md) | 3 |  | BASE TABLE |
 | [public.secrets_provider_connection](public.secrets_provider_connection.md) | 7 |  | BASE TABLE |
 | [public.settings](public.settings.md) | 3 |  | BASE TABLE |
@@ -1132,6 +1132,7 @@ erDiagram
   integer attempts
   varchar_255_ claimedBy
   timestamp_3__with_time_zone createdAt
+  timestamp_3__with_time_zone dispatchedAt
   text errorMessage
   timestamp_3__with_time_zone finishedAt
   bigint id

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

@@ -89,6 +89,7 @@ erDiagram
   integer attempts
   varchar_255_ claimedBy
   timestamp_3__with_time_zone createdAt
+  timestamp_3__with_time_zone dispatchedAt
   text errorMessage
   timestamp_3__with_time_zone finishedAt
   bigint id

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

@@ -7,6 +7,7 @@
 | attempts | integer | 0 | false |  |  | Execution attempts started so far; compared against maxAttempts. |
 | claimedBy | varchar(255) |  | true |  |  | Id of the instance currently holding the lease; NULL when unclaimed. |
 | createdAt | timestamp(3) with time zone | CURRENT_TIMESTAMP(3) | false |  |  |  |
+| dispatchedAt | timestamp(3) with time zone |  | true |  |  | When the current attempt handed off its effect; NULL until then. Splits dispatch-attempted (startedAt) from effect-happened, so the reaper completes a dispatched occurrence rather than redelivering it. |
 | errorMessage | text |  | true |  |  | Failure detail from the last attempt. |
 | finishedAt | timestamp(3) with time zone |  | true |  |  | When the occurrence reached a terminal state; drives retention pruning. |
 | id | bigint |  | false |  |  |  |
@@ -62,6 +63,7 @@ erDiagram
   integer attempts
   varchar_255_ claimedBy
   timestamp_3__with_time_zone createdAt
+  timestamp_3__with_time_zone dispatchedAt
   text errorMessage
   timestamp_3__with_time_zone finishedAt
   bigint id

docs/generated/sqlite-schema/README.md

@@ -94,7 +94,7 @@ Auto-generated from the SQLite migrations in @n8n/db. Do not edit by hand.
 | [role_mapping_rule_project](role_mapping_rule_project.md) | 2 |  | table |
 | [role_scope](role_scope.md) | 2 |  | table |
 | [scheduled_job](scheduled_job.md) | 19 |  | table |
-| [scheduled_task](scheduled_task.md) | 16 |  | table |
+| [scheduled_task](scheduled_task.md) | 17 |  | table |
 | [scope](scope.md) | 3 |  | table |
 | [secrets_provider_connection](secrets_provider_connection.md) | 7 |  | table |
 | [settings](settings.md) | 3 |  | table |
@@ -1119,6 +1119,7 @@ erDiagram
   INTEGER attempts
   varchar_255_ claimedBy
   datetime_3_ createdAt
+  datetime_3_ dispatchedAt
   TEXT errorMessage
   datetime_3_ finishedAt
   INTEGER id

docs/generated/sqlite-schema/scheduled_job.md

@@ -90,6 +90,7 @@ erDiagram
   INTEGER attempts
   varchar_255_ claimedBy
   datetime_3_ createdAt
+  datetime_3_ dispatchedAt
   TEXT errorMessage
   datetime_3_ finishedAt
   INTEGER id

docs/generated/sqlite-schema/scheduled_task.md

@@ -6,7 +6,7 @@
 <summary><strong>Table Definition</strong></summary>
 
 ```sql
-CREATE TABLE "scheduled_task" ("id" integer PRIMARY KEY NOT NULL, "jobId" integer NOT NULL, "taskType" varchar(128) NOT NULL, "payload" text NOT NULL DEFAULT ('{}'), "scheduledFor" datetime(3) NOT NULL, "runAt" datetime(3) NOT NULL, "status" varchar(16) NOT NULL DEFAULT ('pending'), "attempts" integer NOT NULL DEFAULT (0), "maxAttempts" integer NOT NULL DEFAULT (1), "claimedBy" varchar(255), "leaseExpiresAt" datetime(3), "leaseEpoch" integer NOT NULL DEFAULT (0), "startedAt" datetime(3), "finishedAt" datetime(3), "errorMessage" text, "createdAt" datetime(3) NOT NULL DEFAULT (STRFTIME('%Y-%m-%d %H:%M:%f', 'NOW')), CONSTRAINT "CHK_scheduled_task_running_lease" CHECK ("status" <> 'running' OR "leaseExpiresAt" IS NOT NULL), CONSTRAINT "CHK_scheduled_task_status" CHECK ("status" IN ('pending', 'running', 'succeeded', 'failed', 'missed', 'cancelled')), CONSTRAINT "FK_scheduled_task_jobId" FOREIGN KEY ("jobId") REFERENCES "scheduled_job" ("id") ON DELETE CASCADE)
+CREATE TABLE "scheduled_task" ("id" integer PRIMARY KEY NOT NULL, "jobId" integer NOT NULL, "taskType" varchar(128) NOT NULL, "payload" text NOT NULL DEFAULT ('{}'), "scheduledFor" datetime(3) NOT NULL, "runAt" datetime(3) NOT NULL, "status" varchar(16) NOT NULL DEFAULT ('pending'), "attempts" integer NOT NULL DEFAULT (0), "maxAttempts" integer NOT NULL DEFAULT (1), "claimedBy" varchar(255), "leaseExpiresAt" datetime(3), "leaseEpoch" integer NOT NULL DEFAULT (0), "startedAt" datetime(3), "finishedAt" datetime(3), "errorMessage" text, "createdAt" datetime(3) NOT NULL DEFAULT (STRFTIME('%Y-%m-%d %H:%M:%f', 'NOW')), "dispatchedAt" datetime(3), CONSTRAINT "CHK_scheduled_task_running_lease" CHECK ("status" <> 'running' OR "leaseExpiresAt" IS NOT NULL), CONSTRAINT "CHK_scheduled_task_status" CHECK ("status" IN ('pending', 'running', 'succeeded', 'failed', 'missed', 'cancelled')), CONSTRAINT "FK_scheduled_task_jobId" FOREIGN KEY ("jobId") REFERENCES "scheduled_job" ("id") ON DELETE CASCADE)
 ```
 
 </details>
@@ -18,6 +18,7 @@ CREATE TABLE "scheduled_task" ("id" integer PRIMARY KEY NOT NULL, "jobId" intege
 | attempts | INTEGER | 0 | false |  |  |  |
 | claimedBy | varchar(255) |  | true |  |  |  |
 | createdAt | datetime(3) | STRFTIME('%Y-%m-%d %H:%M:%f', 'NOW') | false |  |  |  |
+| dispatchedAt | datetime(3) |  | true |  |  |  |
 | errorMessage | TEXT |  | true |  |  |  |
 | finishedAt | datetime(3) |  | true |  |  |  |
 | id | INTEGER |  | false |  |  |  |
@@ -61,6 +62,7 @@ erDiagram
   INTEGER attempts
   varchar_255_ claimedBy
   datetime_3_ createdAt
+  datetime_3_ dispatchedAt
   TEXT errorMessage
   datetime_3_ finishedAt
   INTEGER id

packages/@n8n/db/src/entities/execution-entity.ts

@@ -31,6 +31,11 @@ export type ExecutionDataStorageLocation = 'db' | 'fs' | 's3' | 'az';
 // Partial index (Postgres only) — supports paginated list queries filtered by
 // workflowId + status without full sequential scans. See migration 1784000000029.
 @Index(['workflowId', 'status', 'id'], { where: '"deletedAt" IS NULL' })
+// Partial unique index, critical for the durable scheduler: this index, not
+// the scheduler's claim, lease, or epoch fencing, is what suppresses a duplicate
+// effect per deduplicationKey when the at-least-once scheduler redelivers an
+// occurrence. Dropping it silently removes that protection.
+@Index(['deduplicationKey'], { unique: true, where: '"deduplicationKey" IS NOT NULL' })
 export class ExecutionEntity {
 	@Generated()
 	@PrimaryColumn({ transformer: idStringifier })

packages/@n8n/db/src/entities/scheduled-task.ts

@@ -24,10 +24,17 @@ export {
  * - and who is currently running it ({@link claimedBy} and the lease columns)
  *
  * "Claiming" a row means a worker briefly reserves it
- * so two workers don't run the same task at once.
+ * so two workers don't *claim* the same task at once.
  *
  * That reservation (a "lease") expires after a while,
  * so if the worker dies mid-run another worker can take over.
+ *
+ * The reverse also holds: a worker that outlives its lease may still be running
+ * the task while another worker claims it, so the claim alone does not prevent
+ * two workers running one task. The scheduler is at-least-once (a lost run is
+ * worse than a duplicate), and running the same occurrence's effect twice is
+ * suppressed best-effort by the unique `deduplicationKey` index on
+ * `execution_entity`.
  */
 @Entity({ name: 'scheduled_task' })
 @Index(['jobId', 'scheduledFor'], { unique: true })
@@ -152,11 +159,26 @@ export class ScheduledTask extends WithCreatedAt {
 	leaseEpoch: number;
 
 	/**
-	 * When the current try started running.
+	 * When the current attempt started running, i.e. when the executor handed the
+	 * occurrence to its handler. Set by the executor's pre-dispatch compare-and-set
+	 * *before* the handler runs (guarded on this being `null`), so it doubles as the
+	 * mutex that runs each occurrence at most once per lease. Cleared when the row
+	 * goes back to `pending` (reclaim, reschedule, release) so a redelivery can
+	 * re-acquire it.
 	 */
 	@DateTimeColumn({ nullable: true })
 	startedAt: Date | null;
 
+	/**
+	 * When the current attempt handed off its effect, i.e. when the handler reported
+	 * dispatch. `null` until then, and a crash between {@link startedAt} and here
+	 * leaves it `null`. The reaper reads this, not {@link startedAt}, to resolve an
+	 * expired lease: set means the effect happened (complete it, never redeliver),
+	 * `null` means it did not (redeliver, so the run is not lost).
+	 */
+	@DateTimeColumn({ nullable: true })
+	dispatchedAt: Date | null;
+
 	/**
 	 * When this run finished, whether it succeeded or failed.
 	 * Used to clean up old rows.

packages/@n8n/db/src/migrations/common/1784000000049-AddScheduledTaskDispatchedAt.ts

@@ -0,0 +1,35 @@
+import type { MigrationContext, ReversibleMigration } from '../migration-types';
+
+const table = 'scheduled_task';
+const columnName = 'dispatchedAt';
+
+/**
+ * Adds `dispatchedAt` to `scheduled_task`, splitting the single dispatch marker
+ * into two:
+ *   - `startedAt`   is now set *before* the handler runs (the pre-dispatch mutex
+ *     the executor uses to run each occurrence at most once per lease),
+ *   - `dispatchedAt` is set *after* the handler reports its effect handed off.
+ *
+ * The reaper reads `dispatchedAt`, not `startedAt`, to decide an expired lease: a
+ * set value means the effect happened (complete it, never redeliver), a NULL means
+ * a crash before dispatch (redeliver, so the run is not lost). Nullable, so added
+ * with a raw ALTER that skips the SQLite table rebuild.
+ */
+export class AddScheduledTaskDispatchedAt1784000000049 implements ReversibleMigration {
+	async up({ runQuery, escape, isPostgres }: MigrationContext) {
+		const tableName = escape.tableName(table);
+		const column = escape.columnName(columnName);
+		const type = isPostgres ? 'timestamptz(3)' : 'datetime(3)';
+		await runQuery(`ALTER TABLE ${tableName} ADD COLUMN ${column} ${type}`);
+		if (isPostgres) {
+			await runQuery(
+				`COMMENT ON COLUMN ${tableName}.${column} IS ` +
+					"'When the current attempt handed off its effect; NULL until then. Splits dispatch-attempted (startedAt) from effect-happened, so the reaper completes a dispatched occurrence rather than redelivering it.'",
+			);
+		}
+	}
+
+	async down({ schemaBuilder: { dropColumns } }: MigrationContext) {
+		await dropColumns(table, [columnName], { recreatesOnSqlite: true });
+	}
+}

packages/@n8n/db/src/migrations/postgresdb/index.ts

@@ -222,6 +222,7 @@ import { AddRecurringCronScheduleKind1784000000045 } from '../common/17840000000
 import { CreateInstanceAiEventsTable1784000000046 } from '../common/1784000000046-CreateInstanceAiEventsTable';
 import { BackfillPreScopingOAuthGrantScopes1784000000047 } from '../common/1784000000047-BackfillPreScopingOAuthGrantScopes';
 import { AddTriggerKindToWorkflowPublicationTriggerStatus1784000000048 } from '../common/1784000000048-AddTriggerKindToWorkflowPublicationTriggerStatus';
+import { AddScheduledTaskDispatchedAt1784000000049 } from '../common/1784000000049-AddScheduledTaskDispatchedAt';
 import type { Migration } from '../migration-types';
 
 export const postgresMigrations: Migration[] = [
@@ -449,4 +450,5 @@ export const postgresMigrations: Migration[] = [
 	CreateInstanceAiEventsTable1784000000046,
 	BackfillPreScopingOAuthGrantScopes1784000000047,
 	AddTriggerKindToWorkflowPublicationTriggerStatus1784000000048,
+	AddScheduledTaskDispatchedAt1784000000049,
 ];

packages/@n8n/db/src/migrations/sqlite/index.ts

@@ -214,6 +214,7 @@ import { AddPartialIndexForGlobalCredentials1784000000044 } from '../common/1784
 import { CreateInstanceAiEventsTable1784000000046 } from '../common/1784000000046-CreateInstanceAiEventsTable';
 import { BackfillPreScopingOAuthGrantScopes1784000000047 } from '../common/1784000000047-BackfillPreScopingOAuthGrantScopes';
 import { AddTriggerKindToWorkflowPublicationTriggerStatus1784000000048 } from '../common/1784000000048-AddTriggerKindToWorkflowPublicationTriggerStatus';
+import { AddScheduledTaskDispatchedAt1784000000049 } from '../common/1784000000049-AddScheduledTaskDispatchedAt';
 
 const sqliteMigrations: Migration[] = [
 	InitialMigration1588102412422,
@@ -431,6 +432,7 @@ const sqliteMigrations: Migration[] = [
 	CreateInstanceAiEventsTable1784000000046,
 	BackfillPreScopingOAuthGrantScopes1784000000047,
 	AddTriggerKindToWorkflowPublicationTriggerStatus1784000000048,
+	AddScheduledTaskDispatchedAt1784000000049,
 ];
 
 export { sqliteMigrations };

packages/@n8n/db/src/repositories/scheduled-task.repository.ts

@@ -1,6 +1,6 @@
 import { DatabaseConfig } from '@n8n/config';
 import { Service } from '@n8n/di';
-import { DataSource, type EntityManager, In, Repository } from '@n8n/typeorm';
+import { DataSource, type EntityManager, In, IsNull, Repository } from '@n8n/typeorm';
 import type { QueryDeepPartialEntity } from '@n8n/typeorm/query-builder/QueryPartialEntity';
 import { UnexpectedError } from 'n8n-workflow';
 
@@ -109,16 +109,18 @@ export type ScheduledTaskMetricSnapshot = {
 export class ScheduledTaskRepository extends Repository<ScheduledTask> {
 	private readonly isPostgres: boolean;
 	private readonly tableName: string;
-	// Quoted here so the reaper update below doesn't have to quote this
-	// camelCase column itself for Postgres (SQLite accepts the same
+	// Quoted here so the reaper update below doesn't have to quote these
+	// camelCase columns itself for Postgres (SQLite accepts the same
 	// double-quoted identifier).
 	private readonly leaseExpiresAtColumn: string;
+	private readonly dispatchedAtColumn: string;
 
 	constructor(dataSource: DataSource, config: DatabaseConfig) {
 		super(ScheduledTask, dataSource.manager);
 		this.isPostgres = config.type === 'postgresdb';
 		this.tableName = this.manager.connection.driver.escape(`${config.tablePrefix}scheduled_task`);
 		this.leaseExpiresAtColumn = this.manager.connection.driver.escape('leaseExpiresAt');
+		this.dispatchedAtColumn = this.manager.connection.driver.escape('dispatchedAt');
 	}
 
 	/**
@@ -284,14 +286,45 @@ export class ScheduledTaskRepository extends Repository<ScheduledTask> {
 	}
 
 	/**
-	 * Set `startedAt` on a task about to dispatch, guarded so it only affects the row
-	 * this `claim` still owns (`running`, same `host` and `leaseEpoch`). Doubles as the
-	 * pre-dispatch existence check: 0 rows means the row was deleted or reclaimed, so
-	 * don't dispatch. Returns rows affected (0 = benign, 1 = proceed).
+	 * Pre-dispatch mutex: atomically claim the sole right to run this occurrence's
+	 * handler for this lease, and refresh the lease for the execution window
+	 * (`leaseExpiresAt = now + leaseMs`). Guarded on the `claim` AND `startedAt IS
+	 * NULL`, so the compare-and-set stamps `startedAt` and returns 1 for the single
+	 * winner; a second fire on the same lease, or a fire on a row already reclaimed
+	 * (epoch bumped) or deleted, matches no row and returns 0. 0 means do not run the
+	 * handler. This is the executor's at-most-once-execute-per-lease guarantee; a
+	 * redelivery only wins after a reclaim cleared `startedAt`.
 	 */
-	async markStarted(claim: HostedClaimedRef): Promise<number> {
+	async beginDispatch(claim: HostedClaimedRef, leaseMs: number): Promise<number> {
+		// Object criteria (not a raw where string) so TypeORM quotes the camelCase
+		// `claimedBy`/`leaseEpoch`/`startedAt` columns correctly on Postgres.
+		const result = await this.update(
+			{
+				id: claim.id,
+				status: ScheduledTaskStatus.Running,
+				claimedBy: claim.host,
+				leaseEpoch: claim.claimedEpoch,
+				startedAt: IsNull(),
+			},
+			{
+				startedAt: () => dbNowLiteral(this.isPostgres),
+				leaseExpiresAt: () => dbNowPlusMsLiteral(this.isPostgres, leaseMs),
+			},
+		);
+		return result.affected ?? 0;
+	}
+
+	/**
+	 * Stamp `dispatchedAt`, the effect-boundary marker, once the handler reports its
+	 * effect was handed off. Guarded so it only affects the row this `claim` still
+	 * owns; 0 rows is a benign no-op (the row was reclaimed meanwhile — the new owner
+	 * stamps its own). Unlike {@link beginDispatch} it is not fenced on the marker
+	 * being null: the scheduler is at-least-once, so this only records that the effect
+	 * happened, letting the reaper complete rather than redeliver the occurrence.
+	 */
+	async markDispatched(claim: HostedClaimedRef): Promise<number> {
 		return await this.runGuardedUpdate(claim, {
-			startedAt: () => dbNowLiteral(this.isPostgres),
+			dispatchedAt: () => dbNowLiteral(this.isPostgres),
 		});
 	}
 
@@ -334,9 +367,10 @@ export class ScheduledTaskRepository extends Repository<ScheduledTask> {
 			errorMessage,
 			claimedBy: null,
 			leaseExpiresAt: null,
-			// The next attempt sets its own start; clear this one's so a pending row
-			// doesn't carry a stale `startedAt`.
+			// The next attempt re-acquires the dispatch mutex and stamps its own markers;
+			// clear both so a pending row carries no stale `startedAt`/`dispatchedAt`.
 			startedAt: null,
+			dispatchedAt: null,
 		});
 	}
 
@@ -353,6 +387,7 @@ export class ScheduledTaskRepository extends Repository<ScheduledTask> {
 			claimedBy: null,
 			leaseExpiresAt: null,
 			startedAt: null,
+			dispatchedAt: null,
 		});
 	}
 
@@ -443,32 +478,64 @@ export class ScheduledTaskRepository extends Repository<ScheduledTask> {
 	 * `@n8n/scheduler`.)
 	 */
 	async reclaimExpired(ref: ClaimedRef, backoffMs: number, errorMessage: string): Promise<number> {
-		return await this.runReaperUpdate(ref, {
-			status: ScheduledTaskStatus.Pending,
-			runAt: () => dbNowPlusMsLiteral(this.isPostgres, backoffMs),
-			attempts: () => 'attempts + 1',
-			leaseEpoch: () => 'leaseEpoch + 1',
-			claimedBy: null,
-			leaseExpiresAt: null,
-			errorMessage,
-			// The next attempt sets its own start; clear this one's so a pending row
-			// doesn't carry a stale `startedAt`.
-			startedAt: null,
-		});
+		return await this.runReaperUpdate(
+			ref,
+			{
+				status: ScheduledTaskStatus.Pending,
+				runAt: () => dbNowPlusMsLiteral(this.isPostgres, backoffMs),
+				attempts: () => 'attempts + 1',
+				leaseEpoch: () => 'leaseEpoch + 1',
+				claimedBy: null,
+				leaseExpiresAt: null,
+				errorMessage,
+				// Reclaim is the pre-dispatch path (fenced on `dispatchedAt IS NULL` below; a
+				// dispatched one is completed instead). Clear `startedAt` so the redelivery can
+				// re-acquire the dispatch mutex; `dispatchedAt` is already null here, cleared
+				// alongside to keep the pending row clean.
+				startedAt: null,
+				dispatchedAt: null,
+			},
+			'pre-dispatch',
+		);
 	}
 
 	/**
 	 * Reaper dead-letter: an expired-lease `running` task at its last attempt to
-	 * terminal `failed`. Same guard as {@link reclaimExpired}; terminal, so no epoch
-	 * bump (the `status` change alone fences a stale owner). Returns rows affected.
+	 * terminal `failed`. Same guard as {@link reclaimExpired}, including the
+	 * `dispatchedAt IS NULL` fence: a dispatched occurrence is never failed (the reaper
+	 * completes it instead). Terminal, so no epoch bump (the `status` change alone
+	 * fences a stale owner). Returns rows affected.
 	 */
 	async deadLetterExpired(ref: ClaimedRef, errorMessage: string): Promise<number> {
-		return await this.runReaperUpdate(ref, {
-			status: ScheduledTaskStatus.Failed,
-			finishedAt: () => dbNowLiteral(this.isPostgres),
-			attempts: () => 'attempts + 1',
-			errorMessage,
-		});
+		return await this.runReaperUpdate(
+			ref,
+			{
+				status: ScheduledTaskStatus.Failed,
+				finishedAt: () => dbNowLiteral(this.isPostgres),
+				attempts: () => 'attempts + 1',
+				errorMessage,
+			},
+			'pre-dispatch',
+		);
+	}
+
+	/**
+	 * Reaper completion: an expired-lease `running` task that was already dispatched
+	 * (its effect happened) to terminal `succeeded`. Recording it failed would blame the
+	 * scheduler for work that was done, and reclaiming it would dispatch the same
+	 * occurrence twice; completing it does neither. Same guard as {@link deadLetterExpired}
+	 * but fenced on `dispatchedAt IS NOT NULL` (the post-dispatch counterpart); terminal,
+	 * so no epoch bump. Returns rows affected.
+	 */
+	async completeExpired(ref: ClaimedRef): Promise<number> {
+		return await this.runReaperUpdate(
+			ref,
+			{
+				status: ScheduledTaskStatus.Succeeded,
+				finishedAt: () => dbNowLiteral(this.isPostgres),
+			},
+			'post-dispatch',
+		);
 	}
 
 	/**
@@ -501,19 +568,33 @@ export class ScheduledTaskRepository extends Repository<ScheduledTask> {
 	 * does not guard on `claimedBy` (the reaper is not the owner) and re-asserts the
 	 * expiry so a lease renewed between the sweep's read and this write is left alone.
 	 * Returns rows affected; 0 is benign (another reaper won it, or the owner finished).
+	 *
+	 * `dispatched` fences the outcome on the effect boundary so it stays consistent
+	 * with the marker at write time, not just at the sweep's read: a pre-dispatch
+	 * outcome (reclaim, dead-letter) only lands while `dispatchedAt` is still null, and
+	 * a post-dispatch one (complete) only while it is set. A `markDispatched` that
+	 * raced in between the sweep's read and this write therefore turns a would-be
+	 * dead-letter/reclaim into a benign no-op instead of failing (or redelivering) a
+	 * dispatched occurrence; the next sweep reads the marker and completes the row.
 	 */
 	private async runReaperUpdate(
 		ref: ClaimedRef,
 		values: QueryDeepPartialEntity<ScheduledTask>,
+		dispatched: 'pre-dispatch' | 'post-dispatch',
 	): Promise<number> {
 		const { id, claimedEpoch } = ref;
-		// No alias on an UPDATE builder, so quote the camelCase column ourselves for
+		// No alias on an UPDATE builder, so quote the camelCase columns ourselves for
 		// Postgres (SQLite accepts the same double-quoted identifier).
+		const dispatchedFence =
+			dispatched === 'post-dispatch'
+				? `${this.dispatchedAtColumn} IS NOT NULL`
+				: `${this.dispatchedAtColumn} IS NULL`;
 		const result = await this.createQueryBuilder()
 			.update(ScheduledTask)
 			.set(values)
 			.where({ id, status: ScheduledTaskStatus.Running, leaseEpoch: claimedEpoch })
 			.andWhere(`${this.leaseExpiresAtColumn} < ${dbNowLiteral(this.isPostgres)}`)
+			.andWhere(dispatchedFence)
 			.execute();
 		return result.affected ?? 0;
 	}

packages/@n8n/scheduler/src/core/__tests__/executor-lookahead.test.ts

@@ -68,7 +68,10 @@ describe('executor claims far enough ahead to fire on time', () => {
 			);
 		}
 
-		async markStarted(): Promise<number> {
+		async beginDispatch(): Promise<number> {
+			return await Promise.resolve(1);
+		}
+		async markDispatched(): Promise<number> {
 			return await Promise.resolve(1);
 		}
 		async completeTask(): Promise<number> {
@@ -101,8 +104,9 @@ describe('executor claims far enough ahead to fire on time', () => {
 		const registry = new TaskHandlerRegistry();
 		const firedAt = new Map<string, number>();
 		registry.register(TASK_TYPE, {
-			execute: async (t) => {
+			execute: async (t, onDispatch) => {
 				firedAt.set(t.id, Date.now());
+				onDispatch();
 				await Promise.resolve();
 			},
 		});

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

@@ -10,6 +10,7 @@ import { createScheduler, DEFAULT_DISPATCH_LAG_WARN_THRESHOLD_SECONDS } from '..
 import type { SchedulerDeps, SchedulerEvent, SchedulerTaskStore } from '../factory';
 import { PASS_TIMED_OUT } from '../lifecycle';
 import type { MaterializerTransaction, RunInTransaction } from '../materializer';
+import type { ExpiredLeaseRow } from '../reaper';
 import { DEFAULT_RETENTION_OPTIONS } from '../retention';
 import type { ClaimedTask, ScheduledJob } from '../types';
 
@@ -194,6 +195,13 @@ const claimedTask = (overrides: Partial<ClaimedTask> = {}): ClaimedTask => ({
 	...overrides,
 });
 
+/** An expired-lease row for the reaper; pre-dispatch (`dispatchedAt` null) by default. */
+const expiredRow = (overrides: Partial<ExpiredLeaseRow> = {}): ExpiredLeaseRow => ({
+	...claimedTask(),
+	dispatchedAt: null,
+	...overrides,
+});
+
 describe('createScheduler execute', () => {
 	it('claims nothing while no handler is registered, then scopes the claim to registered types', async () => {
 		const { scheduler, taskStore } = makeScheduler();
@@ -232,7 +240,7 @@ describe('createScheduler execute', () => {
 			id: '1',
 			claimedEpoch: 1,
 		});
-		expect(taskStore.markStarted).not.toHaveBeenCalled();
+		expect(taskStore.beginDispatch).not.toHaveBeenCalled();
 	});
 
 	it('routes a mid-fire failure to an error event and leaves the row to the reaper', async () => {
@@ -240,7 +248,7 @@ describe('createScheduler execute', () => {
 		scheduler.registerTaskHandler('test-task', { execute: vi.fn() });
 		taskStore.claimDueTasks.mockResolvedValue([claimedTask()]);
 		// The outcome write fails outside the handler-failure path.
-		taskStore.markStarted.mockRejectedValue(new Error('db down'));
+		taskStore.beginDispatch.mockRejectedValue(new Error('db down'));
 
 		await scheduler.execute();
 
@@ -595,7 +603,7 @@ describe('createScheduler late dispatch', () => {
 		taskStore.claimDueTasks.mockResolvedValue([
 			claimedTask({ runAt: new Date('2020-01-01T00:00:00.000Z') }),
 		]);
-		taskStore.markStarted.mockResolvedValue(1);
+		taskStore.markDispatched.mockResolvedValue(1);
 		taskStore.completeTask.mockResolvedValue(1);
 
 		await scheduler.execute();
@@ -628,7 +636,7 @@ describe('createScheduler late dispatch', () => {
 		taskStore.claimDueTasks.mockResolvedValue([
 			claimedTask({ runAt: new Date('2020-01-01T00:00:00.000Z') }),
 		]);
-		taskStore.markStarted.mockResolvedValue(1);
+		taskStore.markDispatched.mockResolvedValue(1);
 		taskStore.completeTask.mockResolvedValue(1);
 
 		await scheduler.execute();
@@ -803,7 +811,7 @@ describe('createScheduler pass timeout and overlap', () => {
 			id: '1',
 			claimedEpoch: 1,
 		});
-		expect(taskStore.markStarted).not.toHaveBeenCalled();
+		expect(taskStore.beginDispatch).not.toHaveBeenCalled();
 
 		await scheduler.stop();
 	});
@@ -901,7 +909,7 @@ describe('createScheduler reap', () => {
 	it('routes a row recovery failure to an error event and finishes the sweep', async () => {
 		const { scheduler, taskStore, onEvent } = makeScheduler();
 		taskStore.findExpiredLeases.mockResolvedValue([
-			{ id: '7', attempts: 0, maxAttempts: 3, leaseEpoch: 1 },
+			expiredRow({ id: '7', attempts: 0, maxAttempts: 3, leaseEpoch: 1 }),
 		]);
 		taskStore.reclaimExpired.mockRejectedValue(new Error('deadlock'));
 
@@ -918,7 +926,7 @@ describe('createScheduler reap', () => {
 	it('routes a dead-lettered task to a warn event carrying the task identity', async () => {
 		const { scheduler, taskStore, onEvent } = makeScheduler();
 		taskStore.findExpiredLeases.mockResolvedValue([
-			{ id: '7', attempts: 2, maxAttempts: 3, leaseEpoch: 1 },
+			expiredRow({ id: '7', attempts: 2, maxAttempts: 3, leaseEpoch: 1 }),
 		]);
 		taskStore.deadLetterExpired.mockResolvedValue(1);
 
@@ -1067,7 +1075,7 @@ describe('createScheduler tracing', () => {
 		const { scheduler, taskStore } = makeScheduler({ tracer });
 		scheduler.registerTaskHandler('test-task', { execute: vi.fn().mockResolvedValue(undefined) });
 		taskStore.claimDueTasks.mockResolvedValue([claimedTask()]);
-		taskStore.markStarted.mockResolvedValue(1);
+		taskStore.beginDispatch.mockResolvedValue(1);
 		taskStore.completeTask.mockResolvedValue(1);
 
 		await scheduler.execute();
@@ -1098,9 +1106,9 @@ describe('createScheduler tracing', () => {
 		// the wrong attribute key cannot slip past these assertions: two leases with
 		// attempts left are reclaimed, one out of attempts is dead-lettered.
 		taskStore.findExpiredLeases.mockResolvedValue([
-			{ id: '7', attempts: 0, maxAttempts: 3, leaseEpoch: 1 },
-			{ id: '8', attempts: 0, maxAttempts: 3, leaseEpoch: 1 },
-			{ id: '9', attempts: 2, maxAttempts: 3, leaseEpoch: 1 },
+			expiredRow({ id: '7', attempts: 0, maxAttempts: 3, leaseEpoch: 1 }),
+			expiredRow({ id: '8', attempts: 0, maxAttempts: 3, leaseEpoch: 1 }),
+			expiredRow({ id: '9', attempts: 2, maxAttempts: 3, leaseEpoch: 1 }),
 		]);
 		taskStore.reclaimExpired.mockResolvedValue(1);
 		taskStore.deadLetterExpired.mockResolvedValue(1);
@@ -1296,7 +1304,7 @@ describe('createScheduler metrics', () => {
 		const metrics = mock<SchedulerMetrics>();
 		const { scheduler, taskStore } = makeScheduler({ metrics });
 		taskStore.findExpiredLeases.mockResolvedValue([
-			{ id: '7', attempts: 2, maxAttempts: 3, leaseEpoch: 1 },
+			expiredRow({ id: '7', attempts: 2, maxAttempts: 3, leaseEpoch: 1 }),
 		]);
 		taskStore.deadLetterExpired.mockResolvedValue(1);
 
@@ -1329,7 +1337,7 @@ describe('createScheduler metrics', () => {
 		scheduler.registerTaskHandler('test-task', { execute: vi.fn().mockResolvedValue(undefined) });
 		// A task due in the past fires on the next timer tick.
 		taskStore.claimDueTasks.mockResolvedValue([claimedTask()]);
-		taskStore.markStarted.mockResolvedValue(1);
+		taskStore.beginDispatch.mockResolvedValue(1);
 		taskStore.completeTask.mockResolvedValue(1);
 
 		await scheduler.execute();
@@ -1351,7 +1359,7 @@ describe('createScheduler metrics', () => {
 		});
 		// Single attempt: the first failure exhausts it.
 		taskStore.claimDueTasks.mockResolvedValue([claimedTask({ attempts: 0, maxAttempts: 1 })]);
-		taskStore.markStarted.mockResolvedValue(1);
+		taskStore.beginDispatch.mockResolvedValue(1);
 		taskStore.failTaskTerminal.mockResolvedValue(1);
 
 		await scheduler.execute();
@@ -1371,7 +1379,7 @@ describe('createScheduler metrics', () => {
 		});
 		// Attempts remain, so the failure reschedules rather than fails terminally.
 		taskStore.claimDueTasks.mockResolvedValue([claimedTask({ attempts: 0, maxAttempts: 3 })]);
-		taskStore.markStarted.mockResolvedValue(1);
+		taskStore.beginDispatch.mockResolvedValue(1);
 		taskStore.rescheduleTask.mockResolvedValue(1);
 
 		await scheduler.execute();
@@ -1404,7 +1412,7 @@ describe('createScheduler metrics', () => {
 		const execute = vi.fn().mockResolvedValue(undefined);
 		scheduler.registerTaskHandler('test-task', { execute });
 		taskStore.claimDueTasks.mockResolvedValue([claimedTask()]);
-		taskStore.markStarted.mockResolvedValue(1);
+		taskStore.beginDispatch.mockResolvedValue(1);
 		taskStore.completeTask.mockResolvedValue(1);
 
 		await scheduler.execute();

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

@@ -26,12 +26,12 @@ const claimedTask = (id: string): ClaimedTask => ({
 /**
  * The claim/dispatch exactly-once invariant, fuzzed over randomised batches:
  * every claimed task dispatches to its handler at most once, and a task whose
- * `markStarted` guard finds nothing to start (already reclaimed or deleted) is
- * never dispatched. Store, registry and timer are all mocked, and each fire
- * callback is invoked by hand, so the detached-fire path stays deterministic.
+ * `beginDispatch` claims no row (already reclaimed or deleted) is never
+ * dispatched. Store, registry and timer are all mocked, and each fire callback is
+ * invoked by hand, so the detached-fire path stays deterministic.
  */
 describe('Executor claim/dispatch (fast-check)', () => {
-	it('dispatches each claimed task at most once, and never dispatches one whose markStarted found nothing to start', async () => {
+	it('dispatches each claimed task at most once, and never dispatches one whose beginDispatch claims no row', async () => {
 		await fc.assert(
 			fc.asyncProperty(
 				fc.uniqueArray(
@@ -51,10 +51,10 @@ describe('Executor claim/dispatch (fast-check)', () => {
 					registry.resolve.mockReturnValue(handler);
 					store.claimDueTasks.mockResolvedValue(tasks);
 					// One resolved value per task, in claim order. `claimAndSchedule` schedules
-					// them in that order, and each callback below calls `markStarted`
+					// them in that order, and each callback below calls `beginDispatch`
 					// synchronously before its first await, so the queued values line up.
 					for (const entry of entries) {
-						store.markStarted.mockResolvedValueOnce(entry.started ? 1 : 0);
+						store.beginDispatch.mockResolvedValueOnce(entry.started ? 1 : 0);
 					}
 					store.completeTask.mockResolvedValue(1);
 

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

@@ -174,9 +174,9 @@ describe('Executor.claimAndSchedule', () => {
 		registry.registeredTypes.mockReturnValue(['workflow:schedule-trigger']);
 		registry.resolve.mockReturnValue({ execute: vi.fn() });
 		store.claimDueTasks.mockResolvedValue([task]);
-		// Make fire reject at its markStarted call to exercise the detached-error path.
+		// Make fire reject at its beginDispatch call to exercise the detached-error path.
 		const failure = new Error('db down');
-		store.markStarted.mockRejectedValue(failure);
+		store.beginDispatch.mockRejectedValue(failure);
 
 		await executor.claimAndSchedule(HOST);
 		const scheduledCallback = timer.schedule.mock.calls[0][1];
@@ -193,7 +193,7 @@ describe('Executor.fire', () => {
 		const { store, registry, executor } = setup();
 		const handler: TaskHandler = { execute: vi.fn() };
 		registry.resolve.mockReturnValue(handler);
-		store.markStarted.mockResolvedValue(0);
+		store.beginDispatch.mockResolvedValue(0);
 
 		const result = await executor.fire(HOST, claimedTask());
 
@@ -206,7 +206,7 @@ describe('Executor.fire', () => {
 	it('dispatches and completes on handler success', async () => {
 		const { store, registry, executor } = setup();
 		const handler: TaskHandler = { execute: vi.fn().mockResolvedValue(undefined) };
-		store.markStarted.mockResolvedValue(1);
+		store.beginDispatch.mockResolvedValue(1);
 		store.completeTask.mockResolvedValue(1);
 		registry.resolve.mockReturnValue(handler);
 		const task = claimedTask();
@@ -214,7 +214,7 @@ describe('Executor.fire', () => {
 		const result = await executor.fire(HOST, task);
 
 		expect(result).toEqual({ outcome: 'completed' });
-		expect(handler.execute).toHaveBeenCalledWith(task);
+		expect(handler.execute).toHaveBeenCalledWith(task, expect.any(Function));
 		expect(store.completeTask).toHaveBeenCalledWith({
 			host: HOST,
 			id: task.id,
@@ -224,20 +224,148 @@ describe('Executor.fire', () => {
 		expect(store.rescheduleTask).not.toHaveBeenCalled();
 	});
 
+	it('stamps the dispatch marker when the handler reports it, and settles it before completing', async () => {
+		const { store, registry, executor } = setup();
+		const order: string[] = [];
+		const handler: TaskHandler = {
+			execute: vi.fn(async (_task: ClaimedTask, onDispatch: () => void) => {
+				onDispatch();
+				await Promise.resolve();
+			}),
+		};
+		store.beginDispatch.mockResolvedValue(1);
+		store.markDispatched.mockImplementation(async () => {
+			order.push('markDispatched');
+			return await Promise.resolve(1);
+		});
+		store.completeTask.mockImplementation(async () => {
+			order.push('completeTask');
+			return await Promise.resolve(1);
+		});
+		registry.resolve.mockReturnValue(handler);
+		const task = claimedTask();
+
+		const result = await executor.fire(HOST, task);
+
+		expect(result).toEqual({ outcome: 'completed' });
+		expect(store.markDispatched).toHaveBeenCalledWith({
+			host: HOST,
+			id: task.id,
+			claimedEpoch: task.leaseEpoch,
+		});
+		// The marker must land before the terminal write, never after it.
+		expect(order).toEqual(['markDispatched', 'completeTask']);
+	});
+
+	it('stamps the marker at most once even if the handler reports dispatch twice', async () => {
+		const { store, registry, executor } = setup();
+		const handler: TaskHandler = {
+			execute: vi.fn(async (_task: ClaimedTask, onDispatch: () => void) => {
+				onDispatch();
+				onDispatch();
+				await Promise.resolve();
+			}),
+		};
+		store.beginDispatch.mockResolvedValue(1);
+		store.markDispatched.mockResolvedValue(1);
+		store.completeTask.mockResolvedValue(1);
+		registry.resolve.mockReturnValue(handler);
+
+		await executor.fire(HOST, claimedTask());
+
+		expect(store.markDispatched).toHaveBeenCalledTimes(1);
+	});
+
+	it('reports a failed marker write without failing the fire (redelivery is acceptable)', async () => {
+		const { store, registry, hooks, executor } = setup();
+		const failure = new Error('db down');
+		const handler: TaskHandler = {
+			execute: vi.fn(async (_task: ClaimedTask, onDispatch: () => void) => {
+				onDispatch();
+				await Promise.resolve();
+			}),
+		};
+		store.beginDispatch.mockResolvedValue(1);
+		store.markDispatched.mockRejectedValue(failure);
+		store.completeTask.mockResolvedValue(1);
+		registry.resolve.mockReturnValue(handler);
+		const task = claimedTask();
+
+		const result = await executor.fire(HOST, task);
+
+		// Losing the marker only costs a redelivery; the fire itself still succeeded.
+		expect(result).toEqual({ outcome: 'completed' });
+		expect(hooks.onFireError).toHaveBeenCalledWith(task, failure);
+	});
+
+	it('settles the marker write before recording a post-dispatch handler failure', async () => {
+		const { store, registry, executor } = setup();
+		const order: string[] = [];
+		const handler: TaskHandler = {
+			execute: vi.fn(async (_task: ClaimedTask, onDispatch: () => void) => {
+				onDispatch();
+				await Promise.resolve();
+				throw new Error('post-dispatch failure');
+			}),
+		};
+		store.beginDispatch.mockResolvedValue(1);
+		store.markDispatched.mockImplementation(async () => {
+			order.push('markDispatched');
+			return await Promise.resolve(1);
+		});
+		store.rescheduleTask.mockImplementation(async () => {
+			order.push('rescheduleTask');
+			return await Promise.resolve(1);
+		});
+		registry.resolve.mockReturnValue(handler);
+
+		const result = await executor.fire(HOST, claimedTask({ attempts: 0, maxAttempts: 3 }));
+
+		expect(result).toEqual({ outcome: 'rescheduled', errorMessage: 'post-dispatch failure' });
+		expect(order).toEqual(['markDispatched', 'rescheduleTask']);
+	});
+
+	it('completes rather than dead-letters a last attempt whose handler threw after dispatch', async () => {
+		const { store, registry, hooks, executor } = setup();
+		const handler: TaskHandler = {
+			execute: vi.fn(async (_task: ClaimedTask, onDispatch: () => void) => {
+				onDispatch();
+				await Promise.resolve();
+				throw new Error('post-dispatch failure');
+			}),
+		};
+		store.beginDispatch.mockResolvedValue(1);
+		store.markDispatched.mockResolvedValue(1);
+		store.completeTask.mockResolvedValue(1);
+		registry.resolve.mockReturnValue(handler);
+
+		// Last attempt (the default): the effect already happened, so the task
+		// must not be recorded failed for work that was done.
+		const result = await executor.fire(HOST, claimedTask({ attempts: 0, maxAttempts: 1 }));
+
+		expect(result).toEqual({ outcome: 'completed' });
+		expect(store.completeTask).toHaveBeenCalledTimes(1);
+		expect(store.failTaskTerminal).not.toHaveBeenCalled();
+		expect(hooks.onFire).toHaveBeenCalledWith('workflow:schedule-trigger', 'success');
+	});
+
 	it('threads the claimed lease epoch through the terminal calls for fencing', async () => {
 		const { store, registry, executor } = setup();
 		const handler: TaskHandler = { execute: vi.fn().mockResolvedValue(undefined) };
-		store.markStarted.mockResolvedValue(1);
+		store.beginDispatch.mockResolvedValue(1);
 		registry.resolve.mockReturnValue(handler);
 		const task = claimedTask({ leaseEpoch: 7 });
 
 		await executor.fire(HOST, task);
 
-		expect(store.markStarted).toHaveBeenCalledWith({
-			host: HOST,
-			id: task.id,
-			claimedEpoch: 7,
-		});
+		expect(store.beginDispatch).toHaveBeenCalledWith(
+			{
+				host: HOST,
+				id: task.id,
+				claimedEpoch: 7,
+			},
+			60_000,
+		);
 		expect(store.completeTask).toHaveBeenCalledWith({
 			host: HOST,
 			id: task.id,
@@ -248,7 +376,7 @@ describe('Executor.fire', () => {
 	it('propagates when recording success fails, without treating it as a handler failure', async () => {
 		const { store, registry, executor } = setup();
 		const handler: TaskHandler = { execute: vi.fn().mockResolvedValue(undefined) };
-		store.markStarted.mockResolvedValue(1);
+		store.beginDispatch.mockResolvedValue(1);
 		registry.resolve.mockReturnValue(handler);
 		store.completeTask.mockRejectedValue(new Error('db down'));
 		const task = claimedTask({ attempts: 0, maxAttempts: 3 });
@@ -263,7 +391,7 @@ describe('Executor.fire', () => {
 	it('retries with backoff for the next attempt when the handler fails and attempts remain', async () => {
 		const { store, registry, executor } = setup();
 		const handler: TaskHandler = { execute: vi.fn().mockRejectedValue(new Error('boom')) };
-		store.markStarted.mockResolvedValue(1);
+		store.beginDispatch.mockResolvedValue(1);
 		store.rescheduleTask.mockResolvedValue(1);
 		registry.resolve.mockReturnValue(handler);
 		const task = claimedTask({ attempts: 0, maxAttempts: 3, leaseEpoch: 7 });
@@ -284,7 +412,7 @@ describe('Executor.fire', () => {
 	it('uses the next attempt number for backoff on a middle attempt', async () => {
 		const { store, registry, executor } = setup();
 		const handler: TaskHandler = { execute: vi.fn().mockRejectedValue(new Error('boom')) };
-		store.markStarted.mockResolvedValue(1);
+		store.beginDispatch.mockResolvedValue(1);
 		registry.resolve.mockReturnValue(handler);
 		const task = claimedTask({ attempts: 1, maxAttempts: 3, leaseEpoch: 7 });
 
@@ -301,7 +429,7 @@ describe('Executor.fire', () => {
 	it('fails terminally when the handler fails on the single default attempt', async () => {
 		const { store, registry, executor } = setup();
 		const handler: TaskHandler = { execute: vi.fn().mockRejectedValue(new Error('boom')) };
-		store.markStarted.mockResolvedValue(1);
+		store.beginDispatch.mockResolvedValue(1);
 		store.failTaskTerminal.mockResolvedValue(1);
 		registry.resolve.mockReturnValue(handler);
 		const task = claimedTask({ attempts: 0, maxAttempts: 1, leaseEpoch: 7 });
@@ -320,7 +448,7 @@ describe('Executor.fire', () => {
 	it('fails terminally on the final attempt of a multi-attempt task', async () => {
 		const { store, registry, executor } = setup();
 		const handler: TaskHandler = { execute: vi.fn().mockRejectedValue(new Error('boom')) };
-		store.markStarted.mockResolvedValue(1);
+		store.beginDispatch.mockResolvedValue(1);
 		registry.resolve.mockReturnValue(handler);
 		// nextAttempt = 3 == maxAttempts -> terminal, not another retry.
 		const task = claimedTask({ attempts: 2, maxAttempts: 3, leaseEpoch: 7 });
@@ -342,8 +470,8 @@ describe('Executor.fire', () => {
 		const result = await executor.fire(HOST, task);
 
 		expect(result).toEqual({ outcome: 'skipped-no-handler' });
-		// Resolved before markStarted, so a task with no handler is never marked started.
-		expect(store.markStarted).not.toHaveBeenCalled();
+		// Resolved before beginDispatch, so a handler-less task never takes the dispatch mutex.
+		expect(store.beginDispatch).not.toHaveBeenCalled();
 		expect(store.releaseClaim).toHaveBeenCalledWith({
 			host: HOST,
 			id: task.id,
@@ -368,7 +496,7 @@ describe('Executor.fire', () => {
 	it('routes the fire through the tracing hook with the host and task, resolving with its result', async () => {
 		const { store, registry, tracing, executor } = setup();
 		const handler: TaskHandler = { execute: vi.fn().mockResolvedValue(undefined) };
-		store.markStarted.mockResolvedValue(1);
+		store.beginDispatch.mockResolvedValue(1);
 		store.completeTask.mockResolvedValue(1);
 		registry.resolve.mockReturnValue(handler);
 		const task = claimedTask();
@@ -383,7 +511,7 @@ describe('Executor.fire', () => {
 	it('lets a throw from the fire reach the tracing hook instead of turning it into an outcome', async () => {
 		const { store, registry, tracing, executor } = setup();
 		const handler: TaskHandler = { execute: vi.fn().mockResolvedValue(undefined) };
-		store.markStarted.mockResolvedValue(1);
+		store.beginDispatch.mockResolvedValue(1);
 		registry.resolve.mockReturnValue(handler);
 		// Recording the outcome fails after the handler already succeeded, so there
 		// is no result to return; the error must propagate as-is.
@@ -398,7 +526,7 @@ describe('Executor.fire', () => {
 describe('Executor.fire metrics hooks', () => {
 	it('calls onDispatch with the lag against the timer clock when a task fires', async () => {
 		const { store, registry, timer, hooks, executor } = setup();
-		store.markStarted.mockResolvedValue(1);
+		store.beginDispatch.mockResolvedValue(1);
 		store.completeTask.mockResolvedValue(1);
 		registry.resolve.mockReturnValue({ execute: vi.fn().mockResolvedValue(undefined) });
 		const task = claimedTask({ runAt: new Date('2026-07-01T00:00:00.000Z') });
@@ -413,7 +541,7 @@ describe('Executor.fire metrics hooks', () => {
 
 	it('measures dispatch lag from runAt, not the fixed scheduledFor slot', async () => {
 		const { store, registry, timer, hooks, executor } = setup();
-		store.markStarted.mockResolvedValue(1);
+		store.beginDispatch.mockResolvedValue(1);
 		store.completeTask.mockResolvedValue(1);
 		registry.resolve.mockReturnValue({ execute: vi.fn().mockResolvedValue(undefined) });
 		// A retried task: runAt was pushed 60s past the original slot by backoff.
@@ -431,7 +559,7 @@ describe('Executor.fire metrics hooks', () => {
 
 	it('clamps dispatch lag to zero when the timer fires before runAt', async () => {
 		const { store, registry, timer, hooks, executor } = setup();
-		store.markStarted.mockResolvedValue(1);
+		store.beginDispatch.mockResolvedValue(1);
 		store.completeTask.mockResolvedValue(1);
 		registry.resolve.mockReturnValue({ execute: vi.fn().mockResolvedValue(undefined) });
 		const task = claimedTask({ runAt: new Date('2026-07-01T00:00:05.000Z') });
@@ -446,7 +574,7 @@ describe('Executor.fire metrics hooks', () => {
 	it('calls no hooks when the row is gone or reclaimed', async () => {
 		const { store, registry, hooks, executor } = setup();
 		registry.resolve.mockReturnValue({ execute: vi.fn() });
-		store.markStarted.mockResolvedValue(0);
+		store.beginDispatch.mockResolvedValue(0);
 
 		await executor.fire(HOST, claimedTask());
 
@@ -456,7 +584,7 @@ describe('Executor.fire metrics hooks', () => {
 
 	it('calls onFire with success when the handler completes', async () => {
 		const { store, registry, hooks, executor } = setup();
-		store.markStarted.mockResolvedValue(1);
+		store.beginDispatch.mockResolvedValue(1);
 		store.completeTask.mockResolvedValue(1);
 		registry.resolve.mockReturnValue({ execute: vi.fn().mockResolvedValue(undefined) });
 		const task = claimedTask();
@@ -470,7 +598,7 @@ describe('Executor.fire metrics hooks', () => {
 
 	it('calls onRetry when the handler fails and attempts remain', async () => {
 		const { store, registry, hooks, executor } = setup();
-		store.markStarted.mockResolvedValue(1);
+		store.beginDispatch.mockResolvedValue(1);
 		store.rescheduleTask.mockResolvedValue(1);
 		registry.resolve.mockReturnValue({ execute: vi.fn().mockRejectedValue(new Error('boom')) });
 		const task = claimedTask({ attempts: 0, maxAttempts: 3 });
@@ -484,7 +612,7 @@ describe('Executor.fire metrics hooks', () => {
 
 	it('calls onFire with failure when the handler fails terminally', async () => {
 		const { store, registry, hooks, executor } = setup();
-		store.markStarted.mockResolvedValue(1);
+		store.beginDispatch.mockResolvedValue(1);
 		store.failTaskTerminal.mockResolvedValue(1);
 		registry.resolve.mockReturnValue({ execute: vi.fn().mockRejectedValue(new Error('boom')) });
 		const task = claimedTask({ attempts: 0, maxAttempts: 1 });
@@ -498,7 +626,7 @@ describe('Executor.fire metrics hooks', () => {
 
 	it('calls no outcome hook and reports the fire as skipped when a terminal write affects no row (reclaimed on lease overrun)', async () => {
 		const { store, registry, hooks, executor } = setup();
-		store.markStarted.mockResolvedValue(1);
+		store.beginDispatch.mockResolvedValue(1);
 		registry.resolve.mockReturnValue({ execute: vi.fn().mockResolvedValue(undefined) });
 		// Every terminal write resolves 0: the row was reclaimed, so nothing is ours to count.
 		store.completeTask.mockResolvedValue(0);
@@ -527,7 +655,7 @@ describe('Executor.fire metrics hooks', () => {
 		const store = mock<ExecutorTaskStore>();
 		const registry = mock<TaskHandlerRegistry>();
 		const timer = mock<PrecisionTimer>();
-		store.markStarted.mockResolvedValue(1);
+		store.beginDispatch.mockResolvedValue(1);
 		store.completeTask.mockResolvedValue(1);
 		registry.resolve.mockReturnValue({ execute: vi.fn().mockResolvedValue(undefined) });
 		// No hooks: the optional calls must not throw.
@@ -590,7 +718,7 @@ describe('Executor.stop', () => {
 		registry.registeredTypes.mockReturnValue(['workflow:schedule-trigger']);
 		registry.resolve.mockReturnValue({ execute: vi.fn().mockResolvedValue(undefined) });
 		store.claimDueTasks.mockResolvedValue([task]);
-		store.markStarted.mockResolvedValue(1); // fire proceeds and completes
+		store.beginDispatch.mockResolvedValue(1); // fire proceeds and completes
 
 		await executor.claimAndSchedule(HOST);
 		// Simulate the timer firing the task before shutdown.

packages/@n8n/scheduler/src/core/executor/executor.ts

@@ -53,7 +53,21 @@ export interface ExecutorHooks {
 /**
  * Claims due tasks, fires each at its `runAt`, dispatches to the handler registered
  * for its `taskType`, and records the outcome. Runs on every main; the claim's
- * locking guarantees each task is owned by one instance.
+ * locking guarantees no two instances *claim* a task at once. Before running a
+ * handler the executor takes a pre-dispatch mutex ({@link ExecutorTaskStore.beginDispatch}):
+ * an atomic compare-and-set that stamps `startedAt` and returns 1 for a single
+ * winner, so the handler runs at most once per lease. That same write refreshes the
+ * lease, giving the handler a full lease for its execution window.
+ *
+ * The contract is at-least-once. Ownership only lasts as long as the lease: if an
+ * owner is lost past it (crash or partition), the reaper reclaims the row, clears
+ * `startedAt`, and another instance re-acquires the mutex and runs the handler again
+ * so the occurrence is not lost. A genuinely stalled-but-alive owner keeps its
+ * (refreshed) lease, so it is not reclaimed and no second handler overlaps it. The
+ * one residual overlap is a partitioned owner still running while its lease is
+ * reclaimed; there the unique `deduplicationKey` index on `execution_entity`
+ * suppresses the duplicate effect. That index, not the claim, is the effect-level
+ * backstop.
  *
  * This is the executor logic only: a driver (the multi-main loop) calls
  * {@link claimAndSchedule} on a cadence and supplies the instance host id. The
@@ -65,8 +79,7 @@ export interface ExecutorHooks {
  * past its lease and is reaped can't write its stale result over the recovered run:
  * while the row sits `pending` the `status = 'running'` guard rejects it, and once
  * another claim takes it the epoch has advanced, so the stale owner's guarded update
- * matches no row. Handlers are still expected to hand off quickly; the lease-renewal
- * heartbeat for longer ones is future work.
+ * matches no row.
  *
  * Persistence sits behind the {@link ExecutorTaskStore} it is given, so this is only
  * the algorithm and a fake store is enough to test it.
@@ -198,10 +211,10 @@ export class Executor {
 	private async runFire(host: string, task: ClaimedTask): Promise<FireResult> {
 		const claim: ClaimedTaskRef = { host, id: task.id, claimedEpoch: task.leaseEpoch };
 
-		// Resolve the handler before marking the task started: don't mark a task started
-		// we can't run, and skip the write on the missing-handler path. The claim is
-		// scoped to registered types, so this normally resolves; if the handler went away
-		// (e.g. a rolling restart), release without counting an attempt so it isn't lost.
+		// Resolve the handler before the ownership check: don't touch the DB for a task
+		// we can't run, and skip on the missing-handler path. The claim is scoped to
+		// registered types, so this normally resolves; if the handler went away (e.g. a
+		// rolling restart), release without counting an attempt so it isn't lost.
 		const handler = this.registry.resolve(task.taskType);
 		if (handler === undefined) {
 			this.hooks.onMissingHandler?.(task);
@@ -209,32 +222,64 @@ export class Executor {
 			return { outcome: 'skipped-no-handler' };
 		}
 
-		// Guard + set `startedAt` in one write. 0 rows => deleted or reclaimed; don't
-		// dispatch an execution for work that is gone or no longer ours.
-		const started = await this.store.markStarted(claim);
-		if (started === 0) {
+		// Pre-dispatch mutex: atomically claim the sole right to run this occurrence's
+		// handler for this lease, and refresh the lease for the execution window. 0 rows
+		// => the row is gone, was reclaimed (epoch bumped), or was already dispatched on
+		// this lease; in every case don't run the handler. This compare-and-set, not the
+		// later marker, is what keeps the executor from calling a handler twice per lease.
+		const won = await this.store.beginDispatch(claim, this.leaseMs);
+		if (won === 0) {
 			return { outcome: 'skipped-not-owned' };
 		}
 
-		// Now that the task is confirmed ours and started, it is genuinely being dispatched.
-		// Lag is measured against `runAt` (the effective fire time, pushed forward by retry
-		// backoff), not the fixed original slot, so a retry's backoff wait isn't logged as lag.
-		// The timer's clock (the one scheduling used) is used, not a fresh wall clock, so a
-		// skewed instance doesn't bias the lag it also scheduled against; clamp non-negative
-		// since a timer can fire marginally early.
+		// The task is confirmed ours and is being handed to its handler. Lag is measured
+		// against `runAt` (the effective fire time, pushed forward by retry backoff), not
+		// the fixed original slot, so a retry's backoff wait isn't logged as lag. The
+		// timer's clock (the one scheduling used) is used, not a fresh wall clock, so a
+		// skewed instance doesn't bias the lag it also scheduled against; clamp
+		// non-negative since a timer can fire marginally early.
 		const lagMs = this.timer.now() - task.runAt.getTime();
 		const lagSeconds = Math.max(0, lagMs) / Time.seconds.toMilliseconds;
 		this.hooks.onDispatch?.(task.taskType, lagSeconds);
 
+		// The handler reports the instant its effect was handed off; persist it as the
+		// `dispatchedAt` marker so the reaper can tell an occurrence that ran from one
+		// that never did. The write is kicked off from the (synchronous) callback and its
+		// promise captured, then settled before any terminal write below so the marker
+		// can't land on (and be rejected by) an already-terminal row. A failed marker
+		// write is reported, not thrown: losing it only costs a redelivery, which the
+		// at-least-once contract accepts. `??=` makes a second onDispatch call a no-op.
+		let dispatchMark: Promise<void> | undefined;
+		const onDispatch = (): void => {
+			dispatchMark ??= this.store.markDispatched(claim).then(
+				() => undefined,
+				(error: unknown) => this.hooks.onFireError?.(task, error),
+			);
+		};
+
 		// Record success only after the try, so a failure to record it isn't taken for a
 		// handler failure. Such a failure propagates out (caught by the detached `.catch`
 		// in claimAndSchedule) and leaves the row `running` for the reaper.
 		try {
-			await handler.execute(task);
+			await handler.execute(task, onDispatch);
 		} catch (error) {
+			await dispatchMark;
 			const errorMessage = ensureError(error).message;
 			const nextAttempts = task.attempts + 1;
 			if (nextAttempts >= task.maxAttempts) {
+				// If the handler had already handed off its effect (onDispatch ran, so
+				// `dispatchMark` is set) before throwing, the occurrence's work is done.
+				// On this last attempt, recording it failed would blame the scheduler for
+				// work that happened; complete it as succeeded instead, mirroring the
+				// reaper's post-dispatch branch. Only pre-dispatch failures are dead-lettered.
+				if (dispatchMark !== undefined) {
+					const rowsAffected = await this.store.completeTask(claim);
+					if (rowsAffected > 0) {
+						this.hooks.onFire?.(task.taskType, 'success');
+						return { outcome: 'completed' };
+					}
+					return { outcome: 'skipped-not-owned', errorMessage };
+				}
 				// A terminal write resolves 0 (it does not reject) when the row was
 				// reclaimed by the reaper after a lease overrun. The result is then no
 				// longer ours to record: report the fire as skipped, not as a state
@@ -259,6 +304,7 @@ export class Executor {
 			return { outcome: 'skipped-not-owned', errorMessage };
 		}
 
+		await dispatchMark;
 		const rowsAffected = await this.store.completeTask(claim);
 		if (rowsAffected > 0) {
 			this.hooks.onFire?.(task.taskType, 'success');

packages/@n8n/scheduler/src/core/executor/store.ts

@@ -44,8 +44,20 @@ export interface ExecutorTaskStore {
 	 */
 	claimDueTasks(batch: ClaimDueTasksBatch): Promise<ClaimedTask[]>;
 
-	/** Record that a claimed task actually began executing (sets `startedAt`). */
-	markStarted(claim: ClaimedTaskRef): Promise<number>;
+	/**
+	 * Pre-dispatch mutex: atomically claim the sole right to run this occurrence's
+	 * handler for this lease and refresh the lease (`leaseExpiresAt = now + leaseMs`)
+	 * for the execution window. Returns rows affected: 1 for the single winner, 0 when
+	 * the row was deleted, reclaimed, or already dispatched on this lease. 0 means do
+	 * not run the handler. This is the executor's at-most-once-execute-per-lease guard.
+	 */
+	beginDispatch(claim: ClaimedTaskRef, leaseMs: number): Promise<number>;
+
+	/**
+	 * Stamp the effect-boundary marker (`dispatchedAt`) once the handler reports its
+	 * effect handed off. Guarded on the claim; 0 rows affected is a benign no-op.
+	 */
+	markDispatched(claim: ClaimedTaskRef): Promise<number>;
 
 	/** Terminal success. */
 	completeTask(claim: ClaimedTaskRef): Promise<number>;

packages/@n8n/scheduler/src/core/executor/task-handler.ts

@@ -5,9 +5,16 @@ import type { ClaimedTask } from '../types';
  * Runs one claimed task. Registered against a `taskType`; the executor resolves
  * the handler for a task's type and calls `execute`. A throw means the attempt
  * failed (the executor retries with backoff or marks the task failed).
+ *
+ * `onDispatch` lets the handler tell the executor the instant its effect was
+ * actually handed off (e.g. the workflow execution was created and started).
+ * The executor records that as the task's dispatch marker, which the reaper uses
+ * to avoid recording an occurrence that did run as failed. Call it once, exactly
+ * when the effect becomes real: not before, and not when a redelivery finds the
+ * effect already exists. Handlers that never dispatch simply never call it.
  */
 export interface TaskHandler {
-	execute(task: ClaimedTask): Promise<void>;
+	execute(task: ClaimedTask, onDispatch: () => void): Promise<void>;
 }
 
 /**

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

@@ -363,6 +363,13 @@ export function createScheduler(deps: SchedulerDeps): Scheduler & SchedulerPasse
 							...task,
 						});
 					},
+					onCompletedAfterDispatch: ({ taskId }) => {
+						emit(
+							'warn',
+							'Scheduler completed a task whose lease lapsed after dispatch; its effect had already happened',
+							{ taskId },
+						);
+					},
 				},
 				signal,
 			);

packages/@n8n/scheduler/src/core/reaper/__tests__/reap.property.test.ts

@@ -7,8 +7,11 @@ import { reap } from '../reap';
 
 /**
  * The reaper's per-row decision over arbitrary expired-lease batches: every row
- * is either reclaimed or dead-lettered (never both, never neither), the choice
- * is exactly "attempts left?", and reclaim always uses the shared backoff curve.
+ * gets exactly one terminal-or-retry decision (never two, never none). The effect
+ * boundary is the primary split: a dispatched row (`dispatchedAt` set) is completed
+ * as succeeded whatever its attempts, never redelivered. Only a never-dispatched row
+ * splits on the attempt count: reclaimed if it has attempts left, else dead-lettered.
+ * Reclaim always uses the shared backoff curve.
  */
 describe('reap decision (fast-check)', () => {
 	const arbRows = fc
@@ -17,38 +20,63 @@ describe('reap decision (fast-check)', () => {
 				attempts: fc.integer({ min: 0, max: 10 }),
 				maxAttempts: fc.integer({ min: 1, max: 10 }),
 				leaseEpoch: fc.integer({ min: 1, max: 100 }),
+				// The effect-boundary marker: dispatched (a timestamp) or not (null).
+				dispatched: fc.boolean(),
 			}),
 			{ maxLength: 30 },
 		)
 		// Ids must be unique, so derive them from the position.
-		.map((rows) => rows.map((row, index): ExpiredLeaseRow => ({ id: `task-${index}`, ...row })));
+		.map((rows) =>
+			rows.map(
+				({ dispatched, ...row }, index): ExpiredLeaseRow => ({
+					id: `task-${index}`,
+					jobId: 1,
+					taskType: 'test',
+					payload: {},
+					scheduledFor: new Date('2026-01-01T00:00:00.000Z'),
+					runAt: new Date('2026-01-01T00:00:00.000Z'),
+					status: 'running',
+					dispatchedAt: dispatched ? new Date('2026-01-01T00:00:00.000Z') : null,
+					...row,
+				}),
+			),
+		);
 
-	it('reclaims xor dead-letters each row, split on remaining attempts', async () => {
+	it('reclaims, completes, or dead-letters each row exactly once, split on attempts and the effect boundary', async () => {
 		await fc.assert(
 			fc.asyncProperty(arbRows, async (rows) => {
 				const store = mock<ReaperTaskStore>();
 				store.findExpiredLeases.mockResolvedValue(rows);
 				store.reclaimExpired.mockResolvedValue(1);
 				store.deadLetterExpired.mockResolvedValue(1);
+				store.completeExpired.mockResolvedValue(1);
 
 				const result = await reap(store, { batchSize: rows.length || 1 });
 
-				const expectedDeadLetter = rows.filter((r) => r.attempts + 1 >= r.maxAttempts);
-				const expectedReclaim = rows.filter((r) => r.attempts + 1 < r.maxAttempts);
+				const lastAttempt = (r: ExpiredLeaseRow) => r.attempts + 1 >= r.maxAttempts;
+				// Effect boundary first: any dispatched row is completed regardless of attempts.
+				// A never-dispatched row then splits on the attempt count.
+				const expectedComplete = rows.filter((r) => r.dispatchedAt !== null);
+				const expectedReclaim = rows.filter((r) => r.dispatchedAt === null && !lastAttempt(r));
+				const expectedDeadLetter = rows.filter((r) => r.dispatchedAt === null && lastAttempt(r));
 
 				const reclaimCalls = store.reclaimExpired.mock.calls;
+				const completeIds = store.completeExpired.mock.calls.map(([ref]) => ref.id);
 				const deadLetterIds = store.deadLetterExpired.mock.calls.map(([ref]) => ref.id);
 
-				// Every row got exactly one decision, and the split is by attempts.
+				// Every row got exactly one decision, split by attempts then effect boundary.
 				expect(reclaimCalls.map(([ref]) => ref.id).sort()).toEqual(
 					expectedReclaim.map((r) => r.id).sort(),
 				);
+				expect([...completeIds].sort()).toEqual(expectedComplete.map((r) => r.id).sort());
 				expect([...deadLetterIds].sort()).toEqual(expectedDeadLetter.map((r) => r.id).sort());
 
-				// Counts reflect the decisions and cover the whole batch.
+				// Counts reflect the decisions. `deadLettered` is genuine terminal failures
+				// only; a post-dispatch completion is a success, reported via the hook and not
+				// counted here. The three decisions together still cover the whole batch.
 				expect(result.reclaimed).toBe(expectedReclaim.length);
 				expect(result.deadLettered).toBe(expectedDeadLetter.length);
-				expect(result.reclaimed + result.deadLettered).toBe(rows.length);
+				expect(result.reclaimed + result.deadLettered + completeIds.length).toBe(rows.length);
 
 				// Reclaim waits the shared backoff for the just-counted attempt, and the
 				// guarded update fences on the epoch read during the sweep.

packages/@n8n/scheduler/src/core/reaper/__tests__/reap.test.ts

@@ -5,9 +5,17 @@ import { reap, type ExpiredLeaseRow, type ReaperTaskStore } from '../reap';
 
 const expiredTask = (overrides: Partial<ExpiredLeaseRow> = {}): ExpiredLeaseRow => ({
 	id: '1',
+	jobId: 1,
+	taskType: 'test',
+	payload: {},
+	scheduledFor: new Date('2026-01-01T00:00:00.000Z'),
+	runAt: new Date('2026-01-01T00:00:00.000Z'),
+	status: 'running',
 	attempts: 0,
 	maxAttempts: 3,
 	leaseEpoch: 1,
+	// Pre-dispatch by default; post-dispatch tests set a concrete `dispatchedAt`.
+	dispatchedAt: null,
 	...overrides,
 });
 
@@ -15,11 +23,14 @@ const setup = () => {
 	const store = mock<ReaperTaskStore>();
 	const onRowError = vi.fn();
 	const onDeadLetter = vi.fn();
+	const onCompletedAfterDispatch = vi.fn();
 	// Guarded updates report one row changed unless a test overrides them.
 	store.reclaimExpired.mockResolvedValue(1);
 	store.deadLetterExpired.mockResolvedValue(1);
-	const run = async () => await reap(store, { batchSize: 100 }, { onRowError, onDeadLetter });
-	return { store, onRowError, onDeadLetter, run };
+	store.completeExpired.mockResolvedValue(1);
+	const run = async () =>
+		await reap(store, { batchSize: 100 }, { onRowError, onDeadLetter, onCompletedAfterDispatch });
+	return { store, onRowError, onDeadLetter, onCompletedAfterDispatch, run };
 };
 
 describe('reap', () => {
@@ -83,6 +94,67 @@ describe('reap', () => {
 		expect(onDeadLetter).toHaveBeenCalledWith({ taskId: task.id, attempts: 1, maxAttempts: 1 });
 	});
 
+	it('dead-letters a never-dispatched occurrence stranded on its last attempt, without dispatching it', async () => {
+		const { store, onDeadLetter } = setup();
+		// Never dispatched, and this expired lease is its last attempt: the effect
+		// never happened, and there is no dispatch fn to hand it off to any more.
+		const task = expiredTask({ attempts: 0, maxAttempts: 1, dispatchedAt: null });
+		store.findExpiredLeases.mockResolvedValue([task]);
+
+		const result = await reap(store, { batchSize: 100 }, { onDeadLetter });
+
+		expect(store.deadLetterExpired).toHaveBeenCalledWith(
+			{ id: task.id, claimedEpoch: task.leaseEpoch },
+			expect.any(String),
+		);
+		expect(store.reclaimExpired).not.toHaveBeenCalled();
+		expect(store.completeExpired).not.toHaveBeenCalled();
+		expect(result).toEqual({ reclaimed: 0, deadLettered: 1 });
+		expect(onDeadLetter).toHaveBeenCalledWith({ taskId: task.id, attempts: 1, maxAttempts: 1 });
+	});
+
+	it('completes a post-dispatch expired lease as succeeded instead of failing it', async () => {
+		const { store, onCompletedAfterDispatch, onDeadLetter } = setup();
+		// `dispatchedAt` set: the effect already happened before the lease lapsed.
+		const task = expiredTask({ attempts: 0, maxAttempts: 1, dispatchedAt: new Date() });
+		store.findExpiredLeases.mockResolvedValue([task]);
+
+		const result = await reap(
+			store,
+			{ batchSize: 100 },
+			{ onCompletedAfterDispatch, onDeadLetter },
+		);
+
+		expect(store.completeExpired).toHaveBeenCalledWith({
+			id: task.id,
+			claimedEpoch: task.leaseEpoch,
+		});
+		expect(store.deadLetterExpired).not.toHaveBeenCalled();
+		expect(onDeadLetter).not.toHaveBeenCalled();
+		expect(onCompletedAfterDispatch).toHaveBeenCalledWith({ taskId: task.id });
+		// A success: reported via the hook, not counted as dead-lettered.
+		expect(result).toEqual({ reclaimed: 0, deadLettered: 0 });
+	});
+
+	it('completes a post-dispatch expired lease even with attempts left, never reclaiming it', async () => {
+		const { store, onCompletedAfterDispatch } = setup();
+		// Dispatched but not on the last attempt: the effect boundary, not the attempt
+		// count, decides, so it is completed rather than reclaimed for a re-run that
+		// would fire the effect twice.
+		const task = expiredTask({ attempts: 0, maxAttempts: 3, dispatchedAt: new Date() });
+		store.findExpiredLeases.mockResolvedValue([task]);
+
+		const result = await reap(store, { batchSize: 100 }, { onCompletedAfterDispatch });
+
+		expect(store.completeExpired).toHaveBeenCalledWith({
+			id: task.id,
+			claimedEpoch: task.leaseEpoch,
+		});
+		expect(store.reclaimExpired).not.toHaveBeenCalled();
+		expect(onCompletedAfterDispatch).toHaveBeenCalledWith({ taskId: task.id });
+		expect(result).toEqual({ reclaimed: 0, deadLettered: 0 });
+	});
+
 	it('dead-letters when the next attempt reaches maxAttempts', async () => {
 		const { store, run } = setup();
 		// nextAttempt = 3 == maxAttempts -> terminal, not another reclaim.

packages/@n8n/scheduler/src/core/reaper/reap.ts

@@ -1,4 +1,5 @@
 import { backoff } from '../executor/backoff';
+import type { ClaimedTask } from '../types';
 
 /** Recorded on a task the reaper recovers, so the failure has a cause. */
 const LEASE_EXPIRED_MESSAGE = 'Lease expired before completion';
@@ -18,14 +19,13 @@ export interface ExpiredLeaseRef {
 }
 
 /**
- * One expired-lease row the sweep decides on: only the fields the loop reads. The
- * storage layer's full task row has these and more, so it fits without adapting.
+ * One expired-lease row the sweep decides on. It carries the full claimed-task
+ * shape plus `dispatchedAt`, the effect-boundary marker: `null` means the owner
+ * was lost before dispatch, so the occurrence's effect never happened. The storage
+ * layer's full task row has these and more, so it fits without adapting.
  */
-export interface ExpiredLeaseRow {
-	id: string;
-	attempts: number;
-	maxAttempts: number;
-	leaseEpoch: number;
+export interface ExpiredLeaseRow extends ClaimedTask {
+	dispatchedAt: Date | null;
 }
 
 /**
@@ -37,6 +37,11 @@ export interface ReaperTaskStore {
 	findExpiredLeases(limit: number): Promise<ExpiredLeaseRow[]>;
 	reclaimExpired(ref: ExpiredLeaseRef, backoffMs: number, errorMessage: string): Promise<number>;
 	deadLetterExpired(ref: ExpiredLeaseRef, errorMessage: string): Promise<number>;
+	/**
+	 * Terminally complete a post-dispatch expired lease as `succeeded`: its effect
+	 * already happened, so it must not be recorded failed nor dispatched again.
+	 */
+	completeExpired(ref: ExpiredLeaseRef): Promise<number>;
 }
 
 /** Knobs of one reaper sweep. */
@@ -55,6 +60,11 @@ export interface ReaperHooks {
 	onRowError?: (taskId: string, error: unknown) => void;
 	/** Notified when a task is failed terminally: the lease of its last attempt expired. */
 	onDeadLetter?: (task: { taskId: string; attempts: number; maxAttempts: number }) => void;
+	/**
+	 * Notified when a post-dispatch task's lease lapsed on its last attempt and it was
+	 * completed as succeeded (its effect had already happened) instead of failed.
+	 */
+	onCompletedAfterDispatch?: (task: { taskId: string }) => void;
 }
 
 /**
@@ -75,9 +85,11 @@ export interface ReaperHooks {
  * reapers on every main are safe. A row that throws is skipped (reported via
  * `hooks.onRowError`), not allowed to abort the rest of the pass.
  *
- * One pass reclaims (or dead-letters) up to `batchSize` expired-lease tasks: a task
- * with attempts left goes back to `pending` with a backoff and a bumped epoch; one
- * at its last attempt fails terminally. Returns the counts.
+ * One pass resolves up to `batchSize` expired-lease tasks, splitting first on the
+ * effect boundary: a task the owner already dispatched is completed as succeeded
+ * (its effect happened, so it is never redelivered), whatever its attempts. A
+ * never-dispatched task with attempts left goes back to `pending` with a backoff and
+ * a bumped epoch; one at its last attempt fails terminally. Returns the counts.
  *
  * Cancellation (`signal`, aborted when the driving loop times the pass out or
  * shuts down) is task-granular.
@@ -105,14 +117,35 @@ export async function reap(
 			// The expired lease means the in-flight attempt is lost, so count it now,
 			// same as a handler failure would.
 			const nextAttempts = task.attempts + 1;
-			if (nextAttempts >= task.maxAttempts) {
-				const affected = await store.deadLetterExpired(
-					{ id: task.id, claimedEpoch: task.leaseEpoch },
-					LEASE_EXPIRED_MESSAGE,
-				);
+			const ref = { id: task.id, claimedEpoch: task.leaseEpoch };
+			// The effect boundary is the primary split, ahead of the attempt count: a row
+			// the owner dispatched before losing its lease already had its effect happen, so
+			// it is completed whatever the attempts left and never redelivered. Only a
+			// never-dispatched row is reclaimed for another attempt or, on its last one,
+			// dead-lettered.
+			if (task.dispatchedAt !== null) {
+				// Post-dispatch: complete as succeeded rather than record a failure for work
+				// that was done, and never dispatch it again. A success, not a failure, so it
+				// is not counted as dead-lettered; `onCompletedAfterDispatch` reports it.
+				const affected = await store.completeExpired(ref);
+				if (affected > 0) {
+					try {
+						// Stryker disable next-line OptionalChaining: the enclosing catch
+						// already swallows a call on an undefined hook, same as `?.` skipping it.
+						hooks.onCompletedAfterDispatch?.({ taskId: task.id });
+					} catch {
+						// A host-supplied reporter must not break the sweep it observes.
+					}
+				}
+			} else if (nextAttempts >= task.maxAttempts) {
+				// Never dispatched, last attempt: the effect never happened and no attempts
+				// remain, so record the terminal failure. Guarded and epoch-fenced, and fenced
+				// on `dispatchedAt` still being null: a marker that landed during the sweep
+				// turns this into a benign no-op (the next sweep then completes the row) instead
+				// of failing a dispatched occurrence. A lost race (0 rows) likewise means
+				// another actor already resolved it.
+				const affected = await store.deadLetterExpired(ref, LEASE_EXPIRED_MESSAGE);
 				deadLettered += affected;
-				// Only an update that actually won the row is a dead-letter; a lost
-				// race means another actor decided the row and there is nothing to report.
 				if (affected > 0) {
 					try {
 						// Stryker disable next-line OptionalChaining: the enclosing catch
@@ -127,11 +160,7 @@ export async function reap(
 					}
 				}
 			} else {
-				reclaimed += await store.reclaimExpired(
-					{ id: task.id, claimedEpoch: task.leaseEpoch },
-					backoff(nextAttempts),
-					LEASE_EXPIRED_MESSAGE,
-				);
+				reclaimed += await store.reclaimExpired(ref, backoff(nextAttempts), LEASE_EXPIRED_MESSAGE);
 			}
 		} catch (error) {
 			try {

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

@@ -1,7 +1,17 @@
 /**
- * The allocated unit of work. Coordination (the scheduler's core) claims, leases,
- * and runs each `ScheduledTask` exactly once on one main; recurrence is just one
- * way these come to exist.
+ * The allocated unit of work. Coordination (the scheduler's core) claims and
+ * leases each `ScheduledTask` so no two instances *claim* it at once; recurrence
+ * is just one way these come to exist.
+ *
+ * The contract is at-least-once: a crashed or lease-lapsed attempt is redelivered
+ * rather than dropped, because a lost run is worse than a duplicate one (an attempt
+ * that exhausts its retries is dead-lettered). So the claim is not exactly-once: an
+ * owner stalled past its lease can still be inside its handler while the reaper
+ * reclaims the row and another instance claims it, and a redelivery runs the
+ * handler again. Duplicate *effects* are suppressed best-effort by the partial
+ * unique index on `execution_entity.deduplicationKey` (see `ExecutionEntity` in
+ * `@n8n/db`). Handlers must therefore be idempotent per occurrence. Tightening the
+ * duplicate-suppression semantics is deferred to the misfire-policy work.
  *
  * Both types carry only the fields the core reads, named and typed as the
  * `scheduled_task` columns, so the storage row satisfies them structurally and

packages/@n8n/scheduler/src/observability/__tests__/executor-tracing.test.ts

@@ -144,10 +144,12 @@ describe('withHandoffTracing', () => {
 		const { span, tracer } = makeTracer();
 		const handler: TaskHandler = { execute: vi.fn().mockResolvedValue(undefined) };
 		const task = claimedTask();
+		const onDispatch = vi.fn();
 
-		await withHandoffTracing(tracer, handler).execute(task);
+		await withHandoffTracing(tracer, handler).execute(task, onDispatch);
 
-		expect(handler.execute).toHaveBeenCalledWith(task);
+		// The wrapper adds a span, not semantics: the dispatch callback flows through.
+		expect(handler.execute).toHaveBeenCalledWith(task, onDispatch);
 		const options = tracer.startSpan.mock.calls[0][0];
 		expect(options.name).toBe('Scheduler handoff');
 		expect(options.op).toBe('scheduler.handoff');
@@ -163,9 +165,9 @@ describe('withHandoffTracing', () => {
 		const { span, tracer } = makeTracer();
 		const handler: TaskHandler = { execute: vi.fn().mockRejectedValue(new Error('boom')) };
 
-		await expect(withHandoffTracing(tracer, handler).execute(claimedTask())).rejects.toThrow(
-			'boom',
-		);
+		await expect(
+			withHandoffTracing(tracer, handler).execute(claimedTask(), vi.fn()),
+		).rejects.toThrow('boom');
 
 		expect(span.setStatus).toHaveBeenCalledWith({ code: SpanStatus.error, message: 'boom' });
 		expect(span.setStatus).not.toHaveBeenCalledWith({ code: SpanStatus.ok });

packages/@n8n/scheduler/src/observability/executor-tracing.ts

@@ -65,7 +65,7 @@ export function createExecutorTracing(tracer: Tracer): ExecutorTracing {
  */
 export function withHandoffTracing(tracer: Tracer, handler: TaskHandler): TaskHandler {
 	return {
-		async execute(task) {
+		async execute(task, onDispatch) {
 			await tracer.startSpan(
 				{
 					name: 'Scheduler handoff',
@@ -77,7 +77,7 @@ export function withHandoffTracing(tracer: Tracer, handler: TaskHandler): TaskHa
 					},
 				},
 				async (span) => {
-					await handler.execute(task);
+					await handler.execute(task, onDispatch);
 					span.setStatus({ code: SpanStatus.ok });
 				},
 			);

packages/cli/src/executions/__tests__/execution-persistence.test.ts

@@ -67,6 +67,7 @@ describe('ExecutionPersistence', () => {
 			raw: {},
 		});
 		mockTx.update.mockResolvedValue({ affected: 1, generatedMaps: [], raw: {} });
+		mockTx.delete.mockResolvedValue({ affected: 1, raw: {} });
 		return mockTx;
 	};
 
@@ -377,6 +378,104 @@ describe('ExecutionPersistence', () => {
 				);
 			});
 		});
+
+		describe('tombstone reclaim', () => {
+			// A sibling test leaves `jsonStore.write` rejecting, and `vi.clearAllMocks()` does not
+			// reset implementations; restore a resolving write so the fs create path succeeds.
+			beforeEach(() => {
+				jsonStore.write.mockResolvedValue(123);
+			});
+
+			const payloadWithKey: CreateExecutionPayload = {
+				...createPayload,
+				deduplicationKey: 'wf-1:node-1:1700000000000',
+			};
+
+			const mockTombstone = (storedAt: 'db' | 'fs' | 's3' | 'az') =>
+				({
+					id: 'exec-old',
+					workflowId: 'workflow-123',
+					storedAt,
+				}) as unknown as ExecutionEntity;
+
+			it('deletes a reclaimed fs-mode tombstone data blob after the replacement commits', async () => {
+				const fsPersistence = createPersistenceService('fs');
+				const mockTx = createMockTransaction();
+				// A prior attempt left an orphaned `new` tombstone under this key, stored on fs.
+				mockTx.findOne.mockResolvedValue(mockTombstone('fs'));
+				executionRepository.manager.transaction = createMockTx(mockTx);
+
+				const executionId = await fsPersistence.create(payloadWithKey);
+
+				expect(executionId).toBe('exec-1');
+				// The tombstone DB row is deleted inside the transaction, scoped to `new`...
+				expect(mockTx.delete).toHaveBeenCalledWith(ExecutionEntity, {
+					id: 'exec-old',
+					status: 'new',
+				});
+				// ...and its out-of-band blob is cleared after commit, keyed by the old (tombstone) id.
+				expect(jsonStore.delete).toHaveBeenCalledWith([
+					{ workflowId: 'workflow-123', executionId: 'exec-old', storedAt: 'fs' },
+				]);
+			});
+
+			it('does not touch the blob store when the reclaimed tombstone was db-stored', async () => {
+				const dbPersistence = createPersistenceService('db');
+				const mockTx = createMockTransaction();
+				mockTx.findOne.mockResolvedValue(mockTombstone('db'));
+				executionRepository.manager.transaction = createMockTx(mockTx);
+
+				await dbPersistence.create(payloadWithKey);
+
+				// db-stored data cascaded with the row delete; nothing to clear out of band.
+				expect(mockTx.delete).toHaveBeenCalledWith(ExecutionEntity, {
+					id: 'exec-old',
+					status: 'new',
+				});
+				expect(jsonStore.delete).not.toHaveBeenCalled();
+			});
+
+			it('does not clear the blob when the tombstone advanced out of `new` before the delete', async () => {
+				const fsPersistence = createPersistenceService('fs');
+				const mockTx = createMockTransaction();
+				// The tombstone was `new` at read, but a worker started it before the delete,
+				// so the `status: 'new'`-scoped delete affects no row.
+				mockTx.findOne.mockResolvedValue(mockTombstone('fs'));
+				mockTx.delete.mockResolvedValue({ affected: 0, raw: {} });
+				executionRepository.manager.transaction = createMockTx(mockTx);
+
+				await fsPersistence.create(payloadWithKey);
+
+				// No row was removed, so the in-flight execution's blob is left in place.
+				expect(jsonStore.delete).not.toHaveBeenCalled();
+			});
+
+			it('reports but does not fail the create when the post-commit blob cleanup fails', async () => {
+				const fsPersistence = createPersistenceService('fs');
+				const mockTx = createMockTransaction();
+				mockTx.findOne.mockResolvedValue(mockTombstone('fs'));
+				executionRepository.manager.transaction = createMockTx(mockTx);
+				const cleanupError = new Error('blob store down');
+				jsonStore.delete.mockRejectedValueOnce(cleanupError);
+
+				// The new execution is committed, so a failed orphan cleanup is reported, not thrown.
+				const executionId = await fsPersistence.create(payloadWithKey);
+
+				expect(executionId).toBe('exec-1');
+				expect(errorReporter.error).toHaveBeenCalledWith(cleanupError, expect.anything());
+			});
+
+			it('skips the tombstone lookup and cleanup entirely without a deduplicationKey', async () => {
+				const fsPersistence = createPersistenceService('fs');
+				const mockTx = createMockTransaction();
+				executionRepository.manager.transaction = createMockTx(mockTx);
+
+				await fsPersistence.create(createPayload); // no deduplicationKey
+
+				expect(mockTx.findOne).not.toHaveBeenCalled();
+				expect(jsonStore.delete).not.toHaveBeenCalled();
+			});
+		});
 	});
 
 	describe('updateExistingExecution', () => {

packages/cli/src/executions/execution-persistence.ts

@@ -93,8 +93,10 @@ export class ExecutionPersistence {
 		const workflowVersionId = workflowData.versionId ?? null;
 		const executionEntity = { ...rest, createdAt: new Date(), storedAt, workflowVersionId };
 
+		let reclaimedTombstone: DeletionTarget | null = null;
 		try {
-			return await this.executionRepository.manager.transaction(async (tx) => {
+			const executionId = await this.executionRepository.manager.transaction(async (tx) => {
+				reclaimedTombstone = await this.reclaimTombstone(tx, executionEntity.deduplicationKey);
 				const { identifiers } = await tx.insert(ExecutionEntity, executionEntity);
 				const executionId = String(identifiers[0].id);
 				const ref = { workflowId: id, executionId };
@@ -116,6 +118,12 @@ export class ExecutionPersistence {
 
 				return executionId;
 			});
+
+			// Clear the reclaimed tombstone's blob only now, once the replacement has
+			// committed (blob deletes are not transactional; see the method).
+			await this.deleteReclaimedTombstoneData(reclaimedTombstone);
+
+			return executionId;
 		} catch (error) {
 			if (executionEntity.deduplicationKey && this.isDuplicateExecutionError(error)) {
 				throw new DuplicateExecutionError(executionEntity.deduplicationKey, error);
@@ -124,6 +132,81 @@ export class ExecutionPersistence {
 		}
 	}
 
+	/**
+	 * Clear an orphaned tombstone before claiming its dedup key.
+	 *
+	 * A prior attempt can insert the execution under this key and then die before
+	 * dispatching it, so the row never advances past `new`. That tombstone asserts
+	 * an effect that never happened: without clearing it, the redelivered occurrence
+	 * collides on insert and the caller mistakes it for an already-run handoff,
+	 * dropping the occurrence. Deleting the row lets the redelivery take over the key;
+	 * in `db` mode that cascades its data, while blob-stored data (fs/s3/az) lives out
+	 * of band and is returned here so `create` can delete it after the replacement
+	 * commits. Any other status reflects a real dispatch, so it is left in place and the
+	 * insert still surfaces the duplicate.
+	 *
+	 * Known imprecision, accepted under the scheduler's at-least-once contract: in
+	 * queue mode an execution stays `new` between being enqueued and a worker picking
+	 * it up, so a redelivery racing that window deletes a genuinely enqueued row and
+	 * re-dispatches the occurrence (the worker's job then finds no execution and fails
+	 * noisily, but the occurrence still runs). Telling "inserted, never enqueued" from
+	 * "enqueued, not yet picked up" apart needs schema the misfire-policy work owns.
+	 *
+	 * @returns the deleted tombstone's storage location, so `create` can clear its
+	 * out-of-band data after committing, or `null` when there was nothing to reclaim.
+	 */
+	private async reclaimTombstone(
+		tx: EntityManager,
+		deduplicationKey: string | null | undefined,
+	): Promise<DeletionTarget | null> {
+		if (!deduplicationKey) return null;
+
+		// Load the tombstone before deleting it so its out-of-band blob can be cleared
+		// after the replacement commits. The unique `deduplicationKey` index means at
+		// most one row carries this key.
+		const tombstone = await tx.findOne(ExecutionEntity, {
+			where: { deduplicationKey, status: 'new' },
+			select: ['id', 'workflowId', 'storedAt'],
+		});
+		if (!tombstone) return null;
+
+		// Scope the delete to `new`: the `findOne` above takes no lock, so a worker may
+		// start the row (moving it out of `new`) between the read and here. Deleting only
+		// while still `new` leaves a started execution in place; the redelivery's insert
+		// then collides and is handled as an existing handoff. Return null when nothing
+		// was deleted, so no blob cleanup runs for a row we kept.
+		const { affected } = await tx.delete(ExecutionEntity, { id: tombstone.id, status: 'new' });
+		if (!affected) return null;
+		return {
+			workflowId: tombstone.workflowId,
+			executionId: tombstone.id,
+			storedAt: tombstone.storedAt,
+		};
+	}
+
+	/**
+	 * Delete a reclaimed tombstone's out-of-band data blob, best-effort. Called after
+	 * the replacement execution has committed, since blob deletes are not
+	 * transactional: doing it earlier would strand the tombstone's blob if the insert
+	 * rolled back. `toBlobRefs` skips a `db`-stored tombstone, whose data the row
+	 * delete already removed. A failed cleanup only leaks the orphan blob, so it is
+	 * reported rather than allowed to fail the (already-persisted) create.
+	 */
+	private async deleteReclaimedTombstoneData(target: DeletionTarget | null): Promise<void> {
+		if (!target) return;
+		// A `db`-stored tombstone's data cascaded with the row delete, so `toBlobRefs`
+		// narrows it away and there is nothing to clear out of band.
+		const blobRefs = this.toBlobRefs([target]);
+		if (blobRefs.length === 0) return;
+		try {
+			await this.jsonStore.delete(blobRefs);
+		} catch (error) {
+			this.errorReporter.error(error, {
+				extra: { executionId: target.executionId, storedAt: target.storedAt },
+			});
+		}
+	}
+
 	/**
 	 * Update an existing execution and, if the payload includes data fields, its data in the configured storage.
 	 * - In `db` mode, we update both entity and data in the DB in a transaction.

packages/cli/src/scheduling/schedule-trigger-node/__tests__/schedule-trigger-task-handler.test.ts

@@ -40,6 +40,9 @@ describe('ScheduleTriggerTaskHandler', () => {
 		workflowExecutionService,
 	);
 
+	// The executor's dispatch-marker callback; cleared each test by vi.clearAllMocks().
+	const onDispatch = vi.fn();
+
 	const triggerNode = mock<INode>({ id: 'node-1', name: 'Schedule Trigger', disabled: false });
 
 	// Plain data objects, not mock proxies: the handler reads them as values.
@@ -88,7 +91,7 @@ describe('ScheduleTriggerTaskHandler', () => {
 
 	describe('handoff', () => {
 		test('creates a trigger execution with the occurrence-derived dedup key', async () => {
-			await handler.execute(buildTask());
+			await handler.execute(buildTask(), onDispatch);
 
 			expect(triggerExecutionContextFactory.loadPublishedWorkflowData).toHaveBeenCalledWith('wf-1');
 			expect(workflowExecutionService.runWorkflow).toHaveBeenCalledWith(
@@ -104,7 +107,7 @@ describe('ScheduleTriggerTaskHandler', () => {
 		});
 
 		test('stamps the trigger item from the occurrence instant in the workflow timezone', async () => {
-			await handler.execute(buildTask());
+			await handler.execute(buildTask(), onDispatch);
 
 			const [, , data] = workflowExecutionService.runWorkflow.mock.calls[0];
 			expect(data[0][0].json).toMatchObject({
@@ -118,7 +121,7 @@ describe('ScheduleTriggerTaskHandler', () => {
 				buildWorkflowData({ settings: {} }),
 			);
 
-			await handler.execute(buildTask());
+			await handler.execute(buildTask(), onDispatch);
 
 			const [, , data] = workflowExecutionService.runWorkflow.mock.calls[0];
 			expect(data[0][0].json).toMatchObject({
@@ -132,7 +135,7 @@ describe('ScheduleTriggerTaskHandler', () => {
 				buildWorkflowData({ settings: { timezone: 'DEFAULT' } }),
 			);
 
-			await handler.execute(buildTask());
+			await handler.execute(buildTask(), onDispatch);
 
 			const [, , data] = workflowExecutionService.runWorkflow.mock.calls[0];
 			// 'DEFAULT' is a sentinel, not a Moment zone: it must not leak into the
@@ -144,7 +147,7 @@ describe('ScheduleTriggerTaskHandler', () => {
 		});
 
 		test('builds additional data for the published workflow like the activation path', async () => {
-			await handler.execute(buildTask());
+			await handler.execute(buildTask(), onDispatch);
 
 			expect(WorkflowExecuteAdditionalData.getBase).toHaveBeenCalledWith({
 				workflowId: 'wf-1',
@@ -153,7 +156,7 @@ describe('ScheduleTriggerTaskHandler', () => {
 		});
 
 		test('emits workflow-executed for the new execution', async () => {
-			await handler.execute(buildTask());
+			await handler.execute(buildTask(), onDispatch);
 
 			expect(eventService.emit).toHaveBeenCalledWith('workflow-executed', {
 				workflowId: 'wf-1',
@@ -162,6 +165,21 @@ describe('ScheduleTriggerTaskHandler', () => {
 				source: 'trigger',
 			});
 		});
+
+		test('reports the dispatch once the run is initiated, not before', async () => {
+			// Report only after runWorkflow resolves: no dispatch is claimed if the
+			// handoff throws first.
+			let reportedAt: 'before' | 'after' | undefined;
+			workflowExecutionService.runWorkflow.mockImplementation(async () => {
+				reportedAt = onDispatch.mock.calls.length === 0 ? 'before' : 'after';
+				return await Promise.resolve('exec-1');
+			});
+
+			await handler.execute(buildTask(), onDispatch);
+
+			expect(onDispatch).toHaveBeenCalledTimes(1);
+			expect(reportedAt).toBe('before'); // not yet called while runWorkflow is in flight
+		});
 	});
 
 	describe('redelivery', () => {
@@ -172,7 +190,9 @@ describe('ScheduleTriggerTaskHandler', () => {
 				mock<ExecutionEntity>({ id: 'exec-0', status: 'running' }),
 			);
 
-			await expect(handler.execute(buildTask({ attempts: 1 }))).resolves.toBeUndefined();
+			await expect(
+				handler.execute(buildTask({ attempts: 1 }), onDispatch),
+			).resolves.toBeUndefined();
 
 			expect(executionRepository.findOne).toHaveBeenCalledWith({
 				where: { deduplicationKey: '7:2026-07-06T07:30:00.000Z' },
@@ -188,6 +208,8 @@ describe('ScheduleTriggerTaskHandler', () => {
 				shouldBeLogged: false,
 			});
 			expect(eventService.emit).not.toHaveBeenCalled();
+			// A redelivery that finds the effect already exists must not report a dispatch.
+			expect(onDispatch).not.toHaveBeenCalled();
 		});
 	});
 
@@ -195,7 +217,7 @@ describe('ScheduleTriggerTaskHandler', () => {
 		test('rejects a task whose payload is missing workflowId or nodeId', async () => {
 			const task = buildTask({ payload: { nodeId: 'node-1' } });
 
-			await expect(handler.execute(task)).rejects.toThrow(UnexpectedError);
+			await expect(handler.execute(task, onDispatch)).rejects.toThrow(UnexpectedError);
 			expect(workflowExecutionService.runWorkflow).not.toHaveBeenCalled();
 		});
 
@@ -203,7 +225,7 @@ describe('ScheduleTriggerTaskHandler', () => {
 			const error = new UnexpectedError('Published version not found for workflow');
 			triggerExecutionContextFactory.loadPublishedWorkflowData.mockRejectedValue(error);
 
-			await expect(handler.execute(buildTask())).rejects.toThrow(error);
+			await expect(handler.execute(buildTask(), onDispatch)).rejects.toThrow(error);
 			expect(workflowExecutionService.runWorkflow).not.toHaveBeenCalled();
 		});
 
@@ -212,7 +234,7 @@ describe('ScheduleTriggerTaskHandler', () => {
 				buildWorkflowData({ nodes: [] }),
 			);
 
-			await expect(handler.execute(buildTask())).rejects.toThrow(
+			await expect(handler.execute(buildTask(), onDispatch)).rejects.toThrow(
 				'missing or disabled in the published workflow',
 			);
 			expect(workflowExecutionService.runWorkflow).not.toHaveBeenCalled();
@@ -223,7 +245,7 @@ describe('ScheduleTriggerTaskHandler', () => {
 				buildWorkflowData({ nodes: [mock<INode>({ id: 'node-1', disabled: true })] }),
 			);
 
-			await expect(handler.execute(buildTask())).rejects.toThrow(
+			await expect(handler.execute(buildTask(), onDispatch)).rejects.toThrow(
 				'missing or disabled in the published workflow',
 			);
 			expect(workflowExecutionService.runWorkflow).not.toHaveBeenCalled();
@@ -233,7 +255,7 @@ describe('ScheduleTriggerTaskHandler', () => {
 			const error = new Error('db unavailable');
 			workflowExecutionService.runWorkflow.mockRejectedValue(error);
 
-			await expect(handler.execute(buildTask())).rejects.toThrow(error);
+			await expect(handler.execute(buildTask(), onDispatch)).rejects.toThrow(error);
 			expect(eventService.emit).not.toHaveBeenCalled();
 		});
 	});

packages/cli/src/scheduling/schedule-trigger-node/schedule-trigger-task-handler.ts

@@ -43,7 +43,7 @@ export class ScheduleTriggerTaskHandler implements TaskHandler {
 		this.logger = this.logger.scoped('scheduler');
 	}
 
-	async execute(task: ClaimedTask): Promise<void> {
+	async execute(task: ClaimedTask, onDispatch: () => void): Promise<void> {
 		const { workflowId, nodeId } = this.parsePayload(task);
 		const workflowData =
 			await this.triggerExecutionContextFactory.loadPublishedWorkflowData(workflowId);
@@ -76,6 +76,8 @@ export class ScheduleTriggerTaskHandler implements TaskHandler {
 				deduplicationKey,
 			);
 
+			onDispatch();
+
 			this.eventService.emit('workflow-executed', {
 				workflowId,
 				workflowName: workflowData.name,

packages/cli/test/integration/executions/execution-persistence.test.ts

@@ -1,5 +1,6 @@
 import { createWorkflow, testDb } from '@n8n/backend-test-utils';
 import type { CreateExecutionPayload, WorkflowEntity } from '@n8n/db';
+import { ExecutionRepository } from '@n8n/db';
 import { Container } from '@n8n/di';
 import { createEmptyRunExecutionData } from 'n8n-workflow';
 
@@ -8,10 +9,12 @@ import { ExecutionPersistence } from '@/executions/execution-persistence';
 
 describe('ExecutionPersistence', () => {
 	let executionPersistence: ExecutionPersistence;
+	let executionRepository: ExecutionRepository;
 
 	beforeAll(async () => {
 		await testDb.init();
 		executionPersistence = Container.get(ExecutionPersistence);
+		executionRepository = Container.get(ExecutionRepository);
 	});
 
 	beforeEach(async () => {
@@ -26,12 +29,13 @@ describe('ExecutionPersistence', () => {
 		const buildPayload = (
 			workflow: WorkflowEntity,
 			deduplicationKey?: string,
+			status: CreateExecutionPayload['status'] = 'new',
 		): CreateExecutionPayload => ({
 			data: createEmptyRunExecutionData(),
 			workflowData: workflow,
 			mode: 'trigger',
 			finished: false,
-			status: 'new',
+			status,
 			workflowId: workflow.id,
 			deduplicationKey,
 		});
@@ -55,11 +59,12 @@ describe('ExecutionPersistence', () => {
 			expect(id1).not.toBe(id2);
 		});
 
-		it('throws DuplicateExecutionError when deduplicationKey is reused', async () => {
+		it('throws DuplicateExecutionError when a dispatched execution already holds the key', async () => {
 			const workflow = await createWorkflow();
 			const key = 'wf:node:t1';
 
-			await executionPersistence.create(buildPayload(workflow, key));
+			// A dispatched execution (status past `new`) is a real effect, not a tombstone.
+			await executionPersistence.create(buildPayload(workflow, key, 'running'));
 
 			await expect(executionPersistence.create(buildPayload(workflow, key))).rejects.toBeInstanceOf(
 				DuplicateExecutionError,
@@ -68,5 +73,20 @@ describe('ExecutionPersistence', () => {
 				deduplicationKey: key,
 			});
 		});
+
+		it('reclaims a tombstone (never-dispatched `new` row) when the key is reused', async () => {
+			const workflow = await createWorkflow();
+			const key = 'wf:node:t1';
+
+			// A prior attempt inserted the execution `new`, then died before dispatch.
+			const tombstoneId = await executionPersistence.create(buildPayload(workflow, key));
+
+			// The redelivery takes over the key instead of colliding, yielding a fresh row.
+			const reclaimedId = await executionPersistence.create(buildPayload(workflow, key));
+
+			expect(reclaimedId).not.toBe(tombstoneId);
+			expect(await executionRepository.findOneBy({ id: tombstoneId })).toBeNull();
+			expect(await executionRepository.findOneBy({ id: reclaimedId })).not.toBeNull();
+		});
 	});
 });

packages/cli/test/integration/scheduling/durable-scheduler-effect-boundary.test.ts

@@ -0,0 +1,244 @@
+import { createWorkflow, testDb } from '@n8n/backend-test-utils';
+import type { ScheduledJob, WorkflowEntity } from '@n8n/db';
+import { DataSource, ScheduledJobRepository, ScheduledTaskRepository } from '@n8n/db';
+import { Container } from '@n8n/di';
+import { createScheduler } from '@n8n/scheduler';
+import type { ClaimedTask, Scheduler, SchedulerPasses, TaskHandler } from '@n8n/scheduler';
+import { createEmptyRunExecutionData } from 'n8n-workflow';
+
+import { DuplicateExecutionError } from '@/errors/duplicate-execution.error';
+import { ExecutionPersistence } from '@/executions/execution-persistence';
+import { buildMaterializerTransaction } from '@/scheduling/durable-scheduler';
+import {
+	SCHEDULE_TRIGGER_TASK_TYPE,
+	scheduleTriggerDeduplicationKey,
+} from '@/scheduling/schedule-trigger-node/schedule-trigger-task';
+
+/**
+ * The durable-scheduler effect boundary under the at-least-once contract.
+ *
+ * A handler models `ScheduleTriggerTaskHandler` faithfully: it inserts a real
+ * execution row under the occurrence-derived dedup key (hitting the real partial
+ * unique index), then reports the dispatch via `onDispatch` (as the real handler
+ * does right after `runWorkflow`), standing in for the running workflow with a
+ * dispatch spy. Assertions are on that spy, never on the presence of an
+ * `execution_entity` row.
+ *
+ * The cases: a redelivery still dispatches past an orphaned `new` tombstone
+ * (`reclaimTombstone`); and a post-dispatch lease lapse is not recorded failed
+ * (the dispatch marker lets the reaper complete it). Concurrent-handler
+ * behaviour is deliberately not asserted here: at-least-once permits overlap,
+ * and tightening it is deferred to the misfire-policy work.
+ */
+describe('durable scheduler effect boundary', () => {
+	const HOST = 'main-effect-boundary';
+
+	let jobRepo: ScheduledJobRepository;
+	let taskRepo: ScheduledTaskRepository;
+	let executionPersistence: ExecutionPersistence;
+	let workflow: WorkflowEntity;
+	let job: ScheduledJob;
+
+	// Schedulers created per test, stopped in afterEach so their timers/loops drain.
+	const schedulers: Array<Scheduler & SchedulerPasses> = [];
+	// Deferreds any hanging handler awaits, always resolved on teardown.
+	let releases: Array<() => void> = [];
+
+	const dispatchSpy = vi.fn();
+
+	const past = () => new Date(Date.now() - 60_000);
+
+	const deferred = () => {
+		let resolve!: () => void;
+		const promise = new Promise<void>((r) => {
+			resolve = r;
+		});
+		return { promise, resolve };
+	};
+
+	const delay = async (ms: number) => await new Promise((resolve) => setTimeout(resolve, ms));
+
+	const waitFor = async (predicate: () => Promise<boolean> | boolean, timeoutMs = 10_000) => {
+		const deadline = Date.now() + timeoutMs;
+		while (Date.now() < deadline) {
+			if (await predicate()) return;
+			await delay(25);
+		}
+		throw new Error('condition not met in time');
+	};
+
+	const makeScheduler = (
+		handler: TaskHandler,
+		hostId = HOST,
+		executor: { leaseSeconds?: number; lookaheadSeconds?: number; batchSize?: number } = {},
+	) => {
+		const scheduler = createScheduler({
+			hostId,
+			materializerTransaction: buildMaterializerTransaction(
+				Container.get(DataSource),
+				jobRepo,
+				taskRepo,
+			),
+			taskStore: taskRepo,
+			executor: { leaseSeconds: 60, lookaheadSeconds: 5, batchSize: 100, ...executor },
+		});
+		scheduler.registerTaskHandler(SCHEDULE_TRIGGER_TASK_TYPE, handler);
+		schedulers.push(scheduler);
+		return scheduler;
+	};
+
+	/**
+	 * Mirrors `ScheduleTriggerTaskHandler.execute`: insert the execution row under the
+	 * occurrence's dedup key (real unique index), then report the dispatch. A
+	 * pre-existing row makes the insert collide, and `DuplicateExecutionError` is
+	 * swallowed like `recordExistingHandoff` does: no dispatch, and no `onDispatch`
+	 * (the effect already exists and isn't ours).
+	 */
+	const effectBoundaryHandler = (opts: { hangAfterDispatch?: boolean } = {}): TaskHandler => ({
+		execute: async (task: ClaimedTask, onDispatch: () => void) => {
+			const deduplicationKey = scheduleTriggerDeduplicationKey(task);
+			try {
+				// The insert transaction: claims the key (execution-persistence.create).
+				await executionPersistence.create({
+					workflowId: workflow.id,
+					data: createEmptyRunExecutionData(),
+					workflowData: workflow,
+					mode: 'trigger',
+					status: 'new',
+					finished: false,
+					deduplicationKey,
+				});
+			} catch (error) {
+				if (!(error instanceof DuplicateExecutionError)) throw error;
+				// A row already holds the key: swallow and complete, as the handler does.
+				return;
+			}
+			// The insert committed and the run was initiated: the effect is real. Stand in
+			// for the running workflow with the spy, then report the dispatch so the task
+			// carries its marker (as the real handler does after runWorkflow).
+			dispatchSpy(task);
+			onDispatch();
+			if (opts.hangAfterDispatch) {
+				const gate = deferred();
+				releases.push(gate.resolve);
+				await gate.promise;
+			}
+		},
+	});
+
+	const createTask = async (overrides: Record<string, unknown> = {}) =>
+		await taskRepo.save(
+			taskRepo.create({
+				jobId: job.id,
+				taskType: SCHEDULE_TRIGGER_TASK_TYPE,
+				payload: { workflowId: 'wf-1', nodeId: 'node-1' },
+				scheduledFor: new Date('2026-07-06T07:30:00.000Z'),
+				runAt: past(),
+				status: 'pending',
+				attempts: 0,
+				maxAttempts: 1,
+				...overrides,
+			}),
+		);
+
+	beforeAll(async () => {
+		await testDb.init();
+		jobRepo = Container.get(ScheduledJobRepository);
+		taskRepo = Container.get(ScheduledTaskRepository);
+		executionPersistence = Container.get(ExecutionPersistence);
+		workflow = await createWorkflow({ settings: { executionOrder: 'v1' } });
+	});
+
+	beforeEach(async () => {
+		dispatchSpy.mockClear();
+		releases = [];
+		await testDb.truncate(['ScheduledTask', 'ScheduledJob', 'ExecutionEntity']);
+		job = await jobRepo.save(
+			jobRepo.create({
+				name: `job-${Math.random().toString(36).slice(2)}`,
+				workflowId: null,
+				nodeId: null,
+				taskType: SCHEDULE_TRIGGER_TASK_TYPE,
+				payload: {},
+				kind: 'interval',
+				intervalSeconds: 3600,
+				enabled: true,
+				nextRunAt: new Date('2026-01-01T00:00:00.000Z'),
+				maxAttempts: 1,
+			}),
+		);
+	});
+
+	afterEach(async () => {
+		for (const release of releases) release();
+		releases = [];
+		await Promise.all(schedulers.map(async (s) => await s.stop()));
+		schedulers.length = 0;
+	});
+
+	afterAll(async () => {
+		await testDb.terminate();
+	});
+
+	// (a) A tombstone row (a prior attempt inserted the execution as `new`, then died
+	// before dispatching) holds the dedup key. The redelivery must still dispatch the
+	// workflow. On master it does not: the insert collides and the dispatch is skipped,
+	// yet the task is recorded `succeeded`.
+	it('dispatches the workflow even when a tombstone execution row already holds the key', async () => {
+		const taskRow = await createTask();
+		const key = scheduleTriggerDeduplicationKey({
+			jobId: taskRow.jobId,
+			scheduledFor: taskRow.scheduledFor,
+		});
+
+		// The stuck row from a crashed earlier attempt: inserted `new`, never dispatched.
+		await executionPersistence.create({
+			workflowId: workflow.id,
+			data: createEmptyRunExecutionData(),
+			workflowData: workflow,
+			mode: 'trigger',
+			status: 'new',
+			finished: false,
+			deduplicationKey: key,
+		});
+
+		const scheduler = makeScheduler(effectBoundaryHandler());
+		const claimed = await scheduler.execute();
+		expect(claimed).toHaveLength(1);
+
+		await waitFor(
+			async () => (await taskRepo.findOneByOrFail({ id: taskRow.id })).status === 'succeeded',
+		);
+
+		// The occurrence's workflow must have been dispatched; on master it never is.
+		expect(dispatchSpy).toHaveBeenCalledTimes(1);
+	}, 15_000);
+
+	// (b) The workflow is dispatched (the marker is stamped), then the lease lapses
+	// before the outcome write. The reaper sees the marker and completes the row rather
+	// than recording `failed` for work that was done.
+	it('does not record a task failed after its workflow was dispatched', async () => {
+		const scheduler = makeScheduler(effectBoundaryHandler({ hangAfterDispatch: true }));
+
+		const taskRow = await createTask({ maxAttempts: 1 });
+
+		// Fire it: the handler dispatches and reports it, then stalls (the instance hangs
+		// before the outcome write). Wait for the dispatch marker to be persisted so the
+		// reaper below reads it: this is the deterministic post-dispatch state.
+		await scheduler.execute();
+		await waitFor(
+			async () => (await taskRepo.findOneByOrFail({ id: taskRow.id })).dispatchedAt !== null,
+		);
+		expect(dispatchSpy).toHaveBeenCalledTimes(1);
+
+		// The stalled owner's lease lapses; the reaper resolves the row.
+		await taskRepo.update({ id: taskRow.id }, { leaseExpiresAt: past() });
+		const result = await scheduler.reap();
+		// A completion is a success, not a dead-letter, so it is not counted as one.
+		expect(result.deadLettered).toBe(0);
+
+		// The marker proves the effect happened, so the row is completed, not failed.
+		const final = await taskRepo.findOneByOrFail({ id: taskRow.id });
+		expect(final.status).toBe('succeeded');
+	}, 15_000);
+});

packages/cli/test/integration/scheduling/executor-repository.test.ts

@@ -219,16 +219,88 @@ describe('ScheduledTaskRepository executor methods', () => {
 			return { id: claimed.id, epoch: claimed.leaseEpoch };
 		}
 
-		it('markStarted sets startedAt for the owner, and is a no-op for a non-owner host', async () => {
+		it('beginDispatch takes the dispatch mutex for the owner, stamps startedAt, refreshes the lease, and refuses a non-owner', async () => {
 			const { id, epoch } = await claimOne();
+			const leaseBefore = (await reload(id)).leaseExpiresAt!.getTime();
 
-			expect(await taskRepository.markStarted({ host: HOST_B, id, claimedEpoch: epoch })).toBe(0);
+			// Non-owner wins no row and writes nothing.
+			expect(
+				await taskRepository.beginDispatch({ host: HOST_B, id, claimedEpoch: epoch }, 120_000),
+			).toBe(0);
 			expect((await reload(id)).startedAt).toBeNull();
 
-			expect(await taskRepository.markStarted({ host: HOST_A, id, claimedEpoch: epoch })).toBe(1);
+			// Owner wins the mutex: 1 row, startedAt stamped, lease pushed out for the run.
+			expect(
+				await taskRepository.beginDispatch({ host: HOST_A, id, claimedEpoch: epoch }, 120_000),
+			).toBe(1);
+			const row = await reload(id);
+			expect(row.startedAt).not.toBeNull();
+			expect(row.leaseExpiresAt!.getTime()).toBeGreaterThan(leaseBefore);
+		});
+
+		it('beginDispatch is a one-shot per lease: a second call wins no row', async () => {
+			const { id, epoch } = await claimOne();
+
+			expect(
+				await taskRepository.beginDispatch({ host: HOST_A, id, claimedEpoch: epoch }, 60_000),
+			).toBe(1);
+			// startedAt is now set, so the mutex is taken: the handler cannot be run twice on
+			// this lease (the executor's at-most-once-execute guarantee).
+			expect(
+				await taskRepository.beginDispatch({ host: HOST_A, id, claimedEpoch: epoch }, 60_000),
+			).toBe(0);
+		});
+
+		it('serialises concurrent beginDispatch calls: exactly one wins the mutex', async () => {
+			// The at-most-once-execute guarantee under contention: several fires racing the
+			// same claim (e.g. a duplicated timer, or a reclaim in flight) must not all run
+			// the handler. The `startedAt IS NULL` guard plus row locking make the UPDATE an
+			// atomic compare-and-set, so exactly one call affects the row and gets 1.
+			const { id, epoch } = await claimOne();
+
+			const results = await Promise.all(
+				Array.from(
+					{ length: 5 },
+					async () =>
+						await taskRepository.beginDispatch({ host: HOST_A, id, claimedEpoch: epoch }, 60_000),
+				),
+			);
+
+			expect(results.filter((n) => n === 1)).toHaveLength(1); // exactly one winner
+			expect(results.filter((n) => n === 0)).toHaveLength(4); // the rest are benign no-ops
 			expect((await reload(id)).startedAt).not.toBeNull();
 		});
 
+		it('markDispatched stamps dispatchedAt for the owner, and is a no-op for a non-owner host', async () => {
+			const { id, epoch } = await claimOne();
+
+			expect(await taskRepository.markDispatched({ host: HOST_B, id, claimedEpoch: epoch })).toBe(
+				0,
+			);
+			expect((await reload(id)).dispatchedAt).toBeNull();
+
+			expect(await taskRepository.markDispatched({ host: HOST_A, id, claimedEpoch: epoch })).toBe(
+				1,
+			);
+			expect((await reload(id)).dispatchedAt).not.toBeNull();
+		});
+
+		it('markDispatched is not fenced on an existing marker (at-least-once allows redelivery)', async () => {
+			const { id, epoch } = await claimOne();
+
+			expect(await taskRepository.markDispatched({ host: HOST_A, id, claimedEpoch: epoch })).toBe(
+				1,
+			);
+			expect((await reload(id)).dispatchedAt).not.toBeNull();
+
+			// A redelivered occurrence is allowed back to its handler, so a second
+			// `markDispatched` at the same claim still lands (no `dispatchedAt IS NULL` fence).
+			expect(await taskRepository.markDispatched({ host: HOST_A, id, claimedEpoch: epoch })).toBe(
+				1,
+			);
+			expect((await reload(id)).dispatchedAt).not.toBeNull();
+		});
+
 		it('completeTask marks succeeded for the owner only', async () => {
 			const { id, epoch } = await claimOne();
 
@@ -314,7 +386,9 @@ describe('ScheduledTaskRepository executor methods', () => {
 			const { id, epoch } = await claimOne();
 			const stale = epoch + 1;
 
-			expect(await taskRepository.markStarted({ host: HOST_A, id, claimedEpoch: stale })).toBe(0);
+			expect(await taskRepository.markDispatched({ host: HOST_A, id, claimedEpoch: stale })).toBe(
+				0,
+			);
 			expect(await taskRepository.completeTask({ host: HOST_A, id, claimedEpoch: stale })).toBe(0);
 			expect(
 				await taskRepository.failTaskTerminal({ host: HOST_A, id, claimedEpoch: stale }, 'x'),
@@ -326,14 +400,16 @@ describe('ScheduledTaskRepository executor methods', () => {
 
 			const row = await reload(id);
 			expect(row.status).toBe('running');
-			expect(row.startedAt).toBeNull();
+			expect(row.dispatchedAt).toBeNull();
 		});
 
 		it('treats a transition on a deleted row as a benign no-op (cascade-delete safety)', async () => {
 			const { id, epoch } = await claimOne();
 			await taskRepository.delete({ id });
 
-			expect(await taskRepository.markStarted({ host: HOST_A, id, claimedEpoch: epoch })).toBe(0);
+			expect(await taskRepository.markDispatched({ host: HOST_A, id, claimedEpoch: epoch })).toBe(
+				0,
+			);
 			expect(await taskRepository.completeTask({ host: HOST_A, id, claimedEpoch: epoch })).toBe(0);
 			expect(
 				await taskRepository.failTaskTerminal({ host: HOST_A, id, claimedEpoch: epoch }, 'x'),
@@ -361,7 +437,9 @@ describe('ScheduledTaskRepository executor methods', () => {
 			expect(await taskRepository.completeTask({ host: HOST_A, id, claimedEpoch: epoch })).toBe(1);
 
 			expect(await taskRepository.completeTask({ host: HOST_A, id, claimedEpoch: epoch })).toBe(0);
-			expect(await taskRepository.markStarted({ host: HOST_A, id, claimedEpoch: epoch })).toBe(0);
+			expect(await taskRepository.markDispatched({ host: HOST_A, id, claimedEpoch: epoch })).toBe(
+				0,
+			);
 
 			const row = await reload(id);
 			expect(row.status).toBe('succeeded');
@@ -395,13 +473,13 @@ describe('ScheduledTaskRepository executor methods', () => {
 			// epoch is a 0-row no-op regardless of which transition the stalled owner calls.
 			const start = await claimOne();
 			expect(
-				await taskRepository.markStarted({
+				await taskRepository.markDispatched({
 					host: HOST_A,
 					id: start.id,
 					claimedEpoch: start.epoch - 1,
 				}),
 			).toBe(0);
-			expect((await reload(start.id)).startedAt).toBeNull();
+			expect((await reload(start.id)).dispatchedAt).toBeNull();
 
 			const fail = await claimOne();
 			expect(
@@ -473,7 +551,11 @@ describe('ScheduledTaskRepository executor methods', () => {
 				claimedBy: HOST_A,
 				leaseExpiresAt: past(),
 				leaseEpoch: 1,
+				// A running row whose lease lapsed after the owner started it (`beginDispatch` ran)
+				// but before the effect was handed off: the pre-dispatch shape the reaper reclaims
+				// or dead-letters. Post-dispatch cases override `dispatchedAt` with a timestamp.
 				startedAt: past(),
+				dispatchedAt: null,
 				maxAttempts: 3,
 				...overrides,
 			});
@@ -535,6 +617,30 @@ describe('ScheduledTaskRepository executor methods', () => {
 				expect(row.runAt.getTime()).toBeGreaterThan(Date.now());
 			});
 
+			it('clears the dispatch mutex (startedAt) on reclaim so the redelivery can re-acquire it', async () => {
+				// Reclaim is the pre-dispatch path: the reaper only reclaims a row whose
+				// effect never landed (`dispatchedAt` null).
+				const task = await createExpiredRunning({
+					leaseEpoch: 2,
+					startedAt: past(),
+					dispatchedAt: null,
+				});
+
+				expect(
+					await taskRepository.reclaimExpired(
+						{ id: task.id, claimedEpoch: 2 },
+						30_000,
+						'lease expired',
+					),
+				).toBe(1);
+
+				// `startedAt` is cleared so the next attempt re-acquires the `beginDispatch` mutex,
+				// and `dispatchedAt` stays null (the effect still has not happened).
+				const row = await reload(task.id);
+				expect(row.startedAt).toBeNull();
+				expect(row.dispatchedAt).toBeNull();
+			});
+
 			it('is a no-op at a stale epoch (a concurrent reaper already reclaimed it)', async () => {
 				const task = await createExpiredRunning({ leaseEpoch: 5 });
 
@@ -596,6 +702,22 @@ describe('ScheduledTaskRepository executor methods', () => {
 				expect(row.attempts).toBe(1); // counted once, not twice
 				expect(row.leaseEpoch).toBe(2); // bumped once, not twice
 			});
+
+			it('is a no-op on a dispatched row: a dispatched occurrence is completed, not reclaimed', async () => {
+				// The pre-dispatch fence (`dispatchedAt IS NULL`) stops the reaper redelivering an
+				// occurrence whose effect already happened, e.g. a marker that landed after the
+				// sweep's read. The next sweep then completes it.
+				const task = await createExpiredRunning({
+					leaseEpoch: 1,
+					attempts: 0,
+					dispatchedAt: past(),
+				});
+
+				expect(
+					await taskRepository.reclaimExpired({ id: task.id, claimedEpoch: 1 }, 30_000, 'x'),
+				).toBe(0);
+				expect((await reload(task.id)).status).toBe('running');
+			});
 		});
 
 		describe('deadLetterExpired', () => {
@@ -669,6 +791,80 @@ describe('ScheduledTaskRepository executor methods', () => {
 				expect(row.status).toBe('failed');
 				expect(row.attempts).toBe(1); // counted once, not twice
 			});
+
+			it('is a no-op on a dispatched row: a dispatched occurrence is never failed', async () => {
+				// The pre-dispatch fence (`dispatchedAt IS NULL`) keeps a dispatch marker that
+				// landed during the sweep from being overwritten with a terminal failure.
+				const task = await createExpiredRunning({
+					leaseEpoch: 1,
+					attempts: 0,
+					maxAttempts: 1,
+					dispatchedAt: past(),
+				});
+
+				expect(await taskRepository.deadLetterExpired({ id: task.id, claimedEpoch: 1 }, 'x')).toBe(
+					0,
+				);
+				expect((await reload(task.id)).status).toBe('running');
+			});
+		});
+
+		describe('completeExpired', () => {
+			it('completes the task as succeeded, stamping finishedAt without failing it', async () => {
+				const task = await createExpiredRunning({
+					attempts: 0,
+					maxAttempts: 1,
+					leaseEpoch: 1,
+					dispatchedAt: past(),
+				});
+
+				expect(await taskRepository.completeExpired({ id: task.id, claimedEpoch: 1 })).toBe(1);
+
+				const row = await reload(task.id);
+				expect(row.status).toBe('succeeded');
+				expect(row.finishedAt).not.toBeNull();
+				expect(row.errorMessage).toBeNull();
+			});
+
+			it('is a no-op at a stale epoch', async () => {
+				const task = await createExpiredRunning({ leaseEpoch: 5, dispatchedAt: past() });
+
+				expect(await taskRepository.completeExpired({ id: task.id, claimedEpoch: 4 })).toBe(0);
+				expect((await reload(task.id)).status).toBe('running');
+			});
+
+			it('is a no-op when the lease is still live (renewed since the sweep read)', async () => {
+				const task = await createExpiredRunning({
+					leaseEpoch: 1,
+					leaseExpiresAt: new Date(Date.now() + 60_000),
+					dispatchedAt: past(),
+				});
+
+				expect(await taskRepository.completeExpired({ id: task.id, claimedEpoch: 1 })).toBe(0);
+				expect((await reload(task.id)).status).toBe('running');
+			});
+
+			it('is a no-op on a pre-dispatch row: the effect never happened, so it is not completed', async () => {
+				// The post-dispatch fence (`dispatchedAt IS NOT NULL`) keeps a never-dispatched
+				// row out of the completion path, so a marker that has not landed cannot be
+				// mistaken for a done effect.
+				const task = await createExpiredRunning({ leaseEpoch: 1, dispatchedAt: null });
+
+				expect(await taskRepository.completeExpired({ id: task.id, claimedEpoch: 1 })).toBe(0);
+				expect((await reload(task.id)).status).toBe('running');
+			});
+
+			it('never lets two concurrent reapers complete the same expired lease', async () => {
+				const task = await createExpiredRunning({ leaseEpoch: 1, dispatchedAt: past() });
+
+				const [a, b] = await Promise.all([
+					taskRepository.completeExpired({ id: task.id, claimedEpoch: 1 }),
+					taskRepository.completeExpired({ id: task.id, claimedEpoch: 1 }),
+				]);
+
+				expect([a, b].sort()).toEqual([0, 1]); // exactly one call completed the row
+				expect((await reload(task.id)).status).toBe('succeeded');
+			});
 		});
 	});
 });

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

@@ -37,8 +37,11 @@ describe('scheduler execution over the storage bindings', () => {
 			taskStore: taskRepo,
 		});
 		scheduler.registerTaskHandler(TASK_TYPE, {
-			execute: async (task) => {
+			execute: async (task, onDispatch) => {
 				executed.push(task);
+				// The fake's effect is the push above; report it so the task carries its
+				// effect marker (`dispatchedAt`) like a real handler would.
+				onDispatch();
 			},
 		});
 	});
@@ -99,7 +102,9 @@ describe('scheduler execution over the storage bindings', () => {
 		expect(executed[0].payload).toEqual({ answer: 42 });
 		const done = await taskRepo.findOneByOrFail({ jobId: job.id });
 		expect(done.finishedAt).not.toBeNull();
+		// `beginDispatch` stamped `startedAt`, and the handler's `onDispatch` stamped `dispatchedAt`.
 		expect(done.startedAt).not.toBeNull();
+		expect(done.dispatchedAt).not.toBeNull();
 		// Terminal rows keep the claim as the record of who ran them.
 		expect(done.claimedBy).toMatch(/^main-/);
 	}, 15_000);

packages/cli/test/integration/scheduling/scheduler-multi-main.test.ts

@@ -165,12 +165,13 @@ describe('scheduler across two mains over one database', () => {
 		expect(done.claimedBy).toBe('main-b');
 	}, 15_000);
 
-	it('dead-letters a claim stranded on its last attempt instead of reclaiming it', async () => {
+	it('dead-letters a never-dispatched claim stranded on its last attempt', async () => {
 		const job = await createJob({ maxAttempts: 3 });
 		const past = new Date(Date.now() - 60_000);
-		// main-a claimed this occurrence for its final attempt, then died: the
-		// lease is expired with no attempts left, so the reaper fails it
-		// terminally instead of retrying.
+		// main-a claimed this occurrence for its final attempt, then died before
+		// dispatching it (no `dispatchedAt`). No attempts remain, so the reaper can't
+		// retry it and never dispatches it either: the run is lost, and the row is
+		// resolved terminally.
 		const doomed = await taskRepo.save(
 			taskRepo.create({
 				jobId: job.id,
@@ -189,17 +190,52 @@ describe('scheduler across two mains over one database', () => {
 
 		expect(await mainB.reap()).toEqual({ reclaimed: 0, deadLettered: 1 });
 
+		// Never dispatched: the handler never ran on either main.
+		expect(executedA).toHaveLength(0);
+		expect(executedB).toHaveLength(0);
+
 		const done = await taskRepo.findOneByOrFail({ id: doomed.id });
-		expect(done.status).toBe('failed');
 		expect(done.attempts).toBe(3);
+		expect(done.status).toBe('failed');
 		expect(done.errorMessage).toMatch(/lease expired/i);
 
-		// Terminal: a further reap sweep has nothing left to do, and the dead task
-		// is never claimed or fired by either main.
+		// Terminal: a further sweep has nothing to do, and neither main re-claims it.
 		expect(await mainB.reap()).toEqual({ reclaimed: 0, deadLettered: 0 });
 		expect(await mainA.execute()).toEqual([]);
 		expect(await mainB.execute()).toEqual([]);
+	}, 15_000);
+
+	it('completes a dispatched claim stranded on its last attempt without re-running it', async () => {
+		const job = await createJob({ maxAttempts: 3 });
+		const past = new Date(Date.now() - 60_000);
+		// main-a dispatched this occurrence (its `dispatchedAt` is set) then died before
+		// recording the outcome. Its effect already happened, so the reaper must not
+		// re-run it nor blame it: it completes the row as succeeded.
+		const dispatched = await taskRepo.save(
+			taskRepo.create({
+				jobId: job.id,
+				taskType: TASK_TYPE,
+				payload: {},
+				scheduledFor: past,
+				runAt: past,
+				status: 'running',
+				claimedBy: 'main-a',
+				leaseExpiresAt: new Date(Date.now() - 1000),
+				leaseEpoch: 1,
+				dispatchedAt: past,
+				attempts: 2,
+				maxAttempts: 3,
+			}),
+		);
+
+		// A completion is a success, not a dead-letter: neither reclaimed nor dead-lettered.
+		expect(await mainB.reap()).toEqual({ reclaimed: 0, deadLettered: 0 });
+
+		// Not re-run: the effect already happened.
 		expect(executedA).toHaveLength(0);
 		expect(executedB).toHaveLength(0);
+
+		const done = await taskRepo.findOneByOrFail({ id: dispatched.id });
+		expect(done.status).toBe('succeeded');
 	}, 15_000);
 });

packages/cli/test/integration/scheduling/scheduler-trigger-handler.test.ts

@@ -194,6 +194,7 @@ describe('schedule-trigger occurrence to a real execution', () => {
 				claimedBy: null,
 				leaseExpiresAt: null,
 				startedAt: null,
+				dispatchedAt: null,
 				finishedAt: null,
 				runAt: new Date(Date.now() - 1000),
 			},