prevent multiple schedules after downtime

carderne · carderne · commit 0044bbfe4627 · 2026-08-11T18:29:46.000+01:00
diff --git a/internal-packages/schedule-engine/src/engine/index.ts b/internal-packages/schedule-engine/src/engine/index.ts
@@ -5,7 +5,7 @@ import type { PrismaClient } from "@trigger.dev/database";
 import { Worker, type JobHandlerParams } from "@trigger.dev/redis-worker";
 import { calculateDistributedExecutionTime } from "./distributedScheduling.js";
 import {
-  calculateNextNominalTimestamp,
+  calculateNextSchedulableOccurrence,
   nextScheduledTimestamps,
   previousScheduledTimestamp,
 } from "./scheduleCalculation.js";
@@ -15,11 +15,7 @@ import type {
   TriggerScheduledTaskCallback,
   TriggerScheduleParams,
 } from "./types.js";
-import {
-  calculateEffectiveScheduleTime,
-  calculateSchedulePhase,
-  type NormalizedScheduleWindow,
-} from "./scheduleTiming.js";
+import { calculateSchedulePhase, type NormalizedScheduleWindow } from "./scheduleTiming.js";
 import { scheduleWorkerCatalog } from "./workerCatalog.js";
 import { tryCatch } from "@trigger.dev/core/utils";
 
@@ -217,33 +213,29 @@ export class ScheduleEngine {
         );
         span.setAttribute("schedule_phase", schedulePhase);
 
-        const fromTimestamp = params.fromTimestamp ?? new Date();
+        const registrationTime = new Date();
+        const fromTimestamp = params.fromTimestamp ?? registrationTime;
         span.setAttribute("from_timestamp", fromTimestamp.toISOString());
 
-        const nominalAt = calculateNextNominalTimestamp(
-          instance.taskSchedule.generatorExpression,
-          instance.taskSchedule.timezone,
-          fromTimestamp
-        );
-        const nextNominalAt = calculateNextNominalTimestamp(
-          instance.taskSchedule.generatorExpression,
-          instance.taskSchedule.timezone,
-          nominalAt
-        );
         const {
-          effectiveAt: candidateEffectiveAt,
+          nominalAt,
+          candidateEffectiveAt,
+          effectiveAt,
           effectiveRangeMs,
           windowMs,
           offsetMs: candidateDelayMs,
           intervalMs,
           windowWasCappedToInterval,
-        } = calculateEffectiveScheduleTime({
-          nominalAt,
-          nextNominalAt,
+          skippedExpiredOccurrences,
+        } = calculateNextSchedulableOccurrence({
+          schedule: instance.taskSchedule.generatorExpression,
+          timezone: instance.taskSchedule.timezone,
+          afterNominal: fromTimestamp,
+          now: registrationTime,
           schedulePhase,
           window: scheduleWindow,
+          cronSpreadEnabled: this.options.cronSpreadEnabled,
         });
-        const effectiveAt = this.options.cronSpreadEnabled ? candidateEffectiveAt : nominalAt;
         const appliedDelayMs = effectiveAt.getTime() - nominalAt.getTime();
 
         span.setAttribute("cron_spread_enabled", this.options.cronSpreadEnabled);
@@ -256,6 +248,14 @@ export class ScheduleEngine {
         span.setAttribute("schedule_window_ms", windowMs);
         span.setAttribute("effective_range_ms", effectiveRangeMs);
         span.setAttribute("schedule_window_was_capped_to_interval", windowWasCappedToInterval);
+        span.setAttribute("schedule_expired_occurrences_skipped", skippedExpiredOccurrences);
+
+        if (skippedExpiredOccurrences) {
+          span.addEvent("schedule_expired_occurrences_skipped", {
+            from_nominal_time: fromTimestamp.toISOString(),
+            selected_nominal_time: nominalAt.toISOString(),
+          });
+        }
 
         if (windowWasCappedToInterval) {
           span.addEvent("schedule_window_capped_to_interval", {
@@ -268,7 +268,7 @@ export class ScheduleEngine {
           });
         }
 
-        const schedulingDelayMs = effectiveAt.getTime() - Date.now();
+        const schedulingDelayMs = effectiveAt.getTime() - registrationTime.getTime();
         span.setAttribute("scheduling_delay_ms", schedulingDelayMs);
 
         this.logger.debug("Calculated next schedule timestamps", {
@@ -283,6 +283,7 @@ export class ScheduleEngine {
           appliedDelayMs,
           effectiveRangeMs,
           windowWasCappedToInterval,
+          skippedExpiredOccurrences,
           schedulingDelayMs,
           generatorExpression: instance.taskSchedule.generatorExpression,
           timezone: instance.taskSchedule.timezone,
@@ -674,8 +675,9 @@ export class ScheduleEngine {
           });
         }
 
-        // Register the next run. `fromTimestamp` advances on every tick so
-        // the next cron slot keeps marching forward even through skips.
+        // Register the next run. `fromTimestamp` anchors nominal chaining;
+        // registration preserves an upcoming effective occurrence and skips
+        // expired intermediate ticks after downtime.
         // `lastScheduleTime` is the actual previous fire time the next job
         // will report as `payload.lastTimestamp` — only advance it when we
         // actually triggered, otherwise carry forward the existing value so
diff --git a/internal-packages/schedule-engine/src/engine/scheduleCalculation.test.ts b/internal-packages/schedule-engine/src/engine/scheduleCalculation.test.ts
@@ -1,4 +1,9 @@
-import { calculateNextNominalTimestamp, nextScheduledTimestamps } from "./scheduleCalculation.js";
+import {
+  calculateNextNominalTimestamp,
+  calculateNextSchedulableOccurrence,
+  nextScheduledTimestamps,
+} from "./scheduleCalculation.js";
+import { SCHEDULE_PHASE_DENOMINATOR } from "./scheduleTiming.js";
 
 describe("calculateNextNominalTimestamp", () => {
   it("advances from the previous nominal tick instead of wall-clock time", () => {
@@ -38,6 +43,90 @@ describe("calculateNextNominalTimestamp", () => {
   });
 });
 
+describe("calculateNextSchedulableOccurrence", () => {
+  const hourlySchedule = "0 * * * *";
+  const window = { type: "percentage", percentage: 100 } as const;
+
+  it("restores wall-clock catch-up behavior when spreading is disabled", () => {
+    const occurrence = calculateNextSchedulableOccurrence({
+      schedule: hourlySchedule,
+      timezone: "UTC",
+      afterNominal: new Date("2026-08-11T09:00:00.000Z"),
+      now: new Date("2026-08-11T12:30:00.000Z"),
+      schedulePhase: (SCHEDULE_PHASE_DENOMINATOR * 3) / 4,
+      window,
+      cronSpreadEnabled: false,
+    });
+
+    expect(occurrence.nominalAt).toEqual(new Date("2026-08-11T13:00:00.000Z"));
+    expect(occurrence.effectiveAt).toEqual(occurrence.nominalAt);
+    expect(occurrence.skippedExpiredOccurrences).toBe(true);
+  });
+
+  it("keeps strict nominal chaining when the next effective time is upcoming", () => {
+    const occurrence = calculateNextSchedulableOccurrence({
+      schedule: hourlySchedule,
+      timezone: "UTC",
+      afterNominal: new Date("2026-08-11T09:00:00.000Z"),
+      now: new Date("2026-08-11T10:00:01.000Z"),
+      schedulePhase: (SCHEDULE_PHASE_DENOMINATOR * 3) / 4,
+      window,
+      cronSpreadEnabled: true,
+    });
+
+    expect(occurrence.nominalAt).toEqual(new Date("2026-08-11T10:00:00.000Z"));
+    expect(occurrence.effectiveAt).toEqual(new Date("2026-08-11T10:45:00.000Z"));
+    expect(occurrence.skippedExpiredOccurrences).toBe(false);
+  });
+
+  it("keeps the latest nominal occurrence when its effective time is upcoming", () => {
+    const occurrence = calculateNextSchedulableOccurrence({
+      schedule: hourlySchedule,
+      timezone: "UTC",
+      afterNominal: new Date("2026-08-11T09:00:00.000Z"),
+      now: new Date("2026-08-11T12:30:00.000Z"),
+      schedulePhase: (SCHEDULE_PHASE_DENOMINATOR * 3) / 4,
+      window,
+      cronSpreadEnabled: true,
+    });
+
+    expect(occurrence.nominalAt).toEqual(new Date("2026-08-11T12:00:00.000Z"));
+    expect(occurrence.effectiveAt).toEqual(new Date("2026-08-11T12:45:00.000Z"));
+    expect(occurrence.skippedExpiredOccurrences).toBe(true);
+  });
+
+  it("skips to the next future nominal occurrence when the latest effective time expired", () => {
+    const occurrence = calculateNextSchedulableOccurrence({
+      schedule: hourlySchedule,
+      timezone: "UTC",
+      afterNominal: new Date("2026-08-11T09:00:00.000Z"),
+      now: new Date("2026-08-11T12:30:00.000Z"),
+      schedulePhase: SCHEDULE_PHASE_DENOMINATOR / 4,
+      window,
+      cronSpreadEnabled: true,
+    });
+
+    expect(occurrence.nominalAt).toEqual(new Date("2026-08-11T13:00:00.000Z"));
+    expect(occurrence.effectiveAt).toEqual(new Date("2026-08-11T13:15:00.000Z"));
+    expect(occurrence.skippedExpiredOccurrences).toBe(true);
+  });
+
+  it("includes a nominal occurrence exactly at now when it is still eligible", () => {
+    const occurrence = calculateNextSchedulableOccurrence({
+      schedule: hourlySchedule,
+      timezone: "UTC",
+      afterNominal: new Date("2026-08-11T09:00:00.000Z"),
+      now: new Date("2026-08-11T12:00:00.000Z"),
+      schedulePhase: 0,
+      window,
+      cronSpreadEnabled: true,
+    });
+
+    expect(occurrence.nominalAt).toEqual(new Date("2026-08-11T12:00:00.000Z"));
+    expect(occurrence.effectiveAt).toEqual(new Date("2026-08-11T12:00:00.000Z"));
+  });
+});
+
 describe("nextScheduledTimestamps", () => {
   it("advances every timestamp from the preceding nominal tick", () => {
     const upcoming = nextScheduledTimestamps(
diff --git a/internal-packages/schedule-engine/src/engine/scheduleCalculation.ts b/internal-packages/schedule-engine/src/engine/scheduleCalculation.ts
@@ -1,4 +1,9 @@
 import { parseExpression } from "cron-parser";
+import {
+  calculateEffectiveScheduleTime,
+  type EffectiveScheduleTime,
+  type NormalizedScheduleWindow,
+} from "./scheduleTiming.js";
 
 export function calculateNextScheduledTimestampFromNow(schedule: string, timezone: string | null) {
   return calculateNextScheduledTimestamp(schedule, timezone, new Date());
@@ -37,6 +42,82 @@ function calculateNextStep(schedule: string, timezone: string | null, currentDat
     .toDate();
 }
 
+type SchedulableOccurrence = Omit<EffectiveScheduleTime, "effectiveAt"> & {
+  candidateEffectiveAt: Date;
+  effectiveAt: Date;
+  skippedExpiredOccurrences: boolean;
+};
+
+/**
+ * Selects the next occurrence that has not passed its actual eligibility time.
+ *
+ * The usual path advances strictly from the preceding nominal tick. If that occurrence expired
+ * during downtime, selection jumps directly to the latest nominal tick that could still be
+ * eligible, or to the first future nominal tick. This preserves one late catch-up without
+ * replaying every missed occurrence.
+ */
+export function calculateNextSchedulableOccurrence({
+  schedule,
+  timezone,
+  afterNominal,
+  now,
+  schedulePhase,
+  window,
+  cronSpreadEnabled,
+}: {
+  schedule: string;
+  timezone: string | null;
+  afterNominal: Date;
+  now: Date;
+  schedulePhase: number;
+  window?: NormalizedScheduleWindow;
+  cronSpreadEnabled: boolean;
+}): SchedulableOccurrence {
+  const occurrenceAt = (
+    nominalAt: Date
+  ): Omit<SchedulableOccurrence, "skippedExpiredOccurrences"> => {
+    const nextNominalAt = calculateNextNominalTimestamp(schedule, timezone, nominalAt);
+    const { effectiveAt: candidateEffectiveAt, ...timing } = calculateEffectiveScheduleTime({
+      nominalAt,
+      nextNominalAt,
+      schedulePhase,
+      window,
+    });
+
+    return {
+      ...timing,
+      candidateEffectiveAt,
+      effectiveAt: cronSpreadEnabled ? candidateEffectiveAt : nominalAt,
+    };
+  };
+
+  const firstNominalAt = calculateNextNominalTimestamp(schedule, timezone, afterNominal);
+  const firstOccurrence = occurrenceAt(firstNominalAt);
+
+  if (firstOccurrence.effectiveAt.getTime() >= now.getTime()) {
+    return { ...firstOccurrence, skippedExpiredOccurrences: false };
+  }
+
+  // `prev()` is strictly before its current date. Advancing by one millisecond includes a cron
+  // tick exactly at `now`, whose effective time may still be upcoming.
+  const latestNominalAt = previousScheduledTimestamp(
+    schedule,
+    timezone,
+    new Date(now.getTime() + 1)
+  );
+
+  if (latestNominalAt.getTime() > afterNominal.getTime()) {
+    const latestOccurrence = occurrenceAt(latestNominalAt);
+
+    if (latestOccurrence.effectiveAt.getTime() >= now.getTime()) {
+      return { ...latestOccurrence, skippedExpiredOccurrences: true };
+    }
+  }
+
+  const nextOccurrence = occurrenceAt(calculateNextNominalTimestamp(schedule, timezone, now));
+  return { ...nextOccurrence, skippedExpiredOccurrences: true };
+}
+
 /**
  * Cron's previous slot relative to `fromTimestamp`. For a continuously-
  * running schedule this equals the actual last fire time; for paused or
diff --git a/internal-packages/schedule-engine/src/engine/types.ts b/internal-packages/schedule-engine/src/engine/types.ts
@@ -84,9 +84,9 @@ export interface TriggerScheduleParams {
 export interface RegisterScheduleInstanceParams {
   instanceId: string;
   /**
-   * Anchor for computing the next cron slot. Defaults to now() when omitted.
-   * This advances on every tick (fired or skipped) so the next slot keeps
-   * marching forward regardless of skip reasons.
+   * Nominal anchor for selecting the next non-expired cron occurrence. Defaults
+   * to now() when omitted. The engine advances from this timestamp when the
+   * next occurrence is still eligible and skips expired intermediate ticks.
    */
   fromTimestamp?: Date;
   /**
diff --git a/internal-packages/schedule-engine/test/scheduleEngine2.test.ts b/internal-packages/schedule-engine/test/scheduleEngine2.test.ts
@@ -100,6 +100,7 @@ describe("ScheduleEngine Integration (part 2)", () => {
         // Call triggerScheduledTask directly without lastScheduleTime or an
         // effective time, simulating an in-flight Redis job from the old engine.
         const exactScheduleTime = new Date("2026-04-30T10:05:00.000Z");
+        const beforeTrigger = new Date();
         await engine.triggerScheduledTask({
           instanceId: scheduleInstance.id,
           finalAttempt: false,
@@ -120,16 +121,19 @@ describe("ScheduleEngine Integration (part 2)", () => {
           exactScheduleTime: string;
           effectiveScheduleTime: string;
         };
-        const nextNominalAt = new Date("2026-04-30T10:10:00.000Z");
+        const nextNominalAt = new Date(nextJobPayload.exactScheduleTime);
 
-        // The next job advances from the legacy job's nominal T, not from the
-        // current wall clock. With cron spread disabled, actual eligibility
-        // remains nominal even though registration still calculates candidate E.
-        expect(new Date(nextJobPayload.exactScheduleTime)).toEqual(nextNominalAt);
+        // The legacy occurrence fires once, then expired intermediate ticks are
+        // skipped instead of being replayed. With spread disabled, eligibility
+        // remains nominal and the next job is in the future.
+        expect(nextNominalAt.getTime()).toBeGreaterThan(beforeTrigger.getTime());
         expect(new Date(nextJobPayload.effectiveScheduleTime)).toEqual(nextNominalAt);
         expect(nextJob!.timestamp).toEqual(
           calculateDistributedExecutionTime(nextNominalAt, 10, scheduleInstance.id)
         );
+        expect(new Date((nextJob!.item as { lastScheduleTime: string }).lastScheduleTime)).toEqual(
+          exactScheduleTime
+        );
 
         const updatedInstance = await prisma.taskScheduleInstance.findUniqueOrThrow({
           where: { id: scheduleInstance.id },
@@ -287,8 +291,9 @@ describe("ScheduleEngine Integration (part 2)", () => {
         });
         expect(preservedInstance.schedulePhase).toBe(pinnedPhase);
 
-        const exactScheduleTime = new Date("2026-04-30T10:00:00.000Z");
-        const effectiveScheduleTime = new Date("2026-04-30T10:00:45.000Z");
+        const intervalMs = 5 * 60_000;
+        const exactScheduleTime = new Date(Math.floor(Date.now() / intervalMs) * intervalMs);
+        const effectiveScheduleTime = new Date(exactScheduleTime.getTime() + 45_000);
         await engine.triggerScheduledTask({
           instanceId: scheduleInstance.id,
           finalAttempt: false,
@@ -306,8 +311,8 @@ describe("ScheduleEngine Integration (part 2)", () => {
           exactScheduleTime: string;
           effectiveScheduleTime: string;
         };
-        const nextNominalAt = new Date("2026-04-30T10:05:00.000Z");
-        const followingNominalAt = new Date("2026-04-30T10:10:00.000Z");
+        const nextNominalAt = new Date(exactScheduleTime.getTime() + intervalMs);
+        const followingNominalAt = new Date(nextNominalAt.getTime() + intervalMs);
         const { effectiveAt: nextEffectiveAt } = calculateEffectiveScheduleTime({
           nominalAt: nextNominalAt,
           nextNominalAt: followingNominalAt,
