fix(schedule-engine): split fromTimestamp from carried lastScheduleTime

Address CodeRabbit review on PR #3476:

1. Engine — split fromTimestamp from lastScheduleTime
   `RegisterScheduleInstanceParams.fromTimestamp` was doing double duty as
   both the next-cron-slot anchor and the "previous fire time" embedded in
   the next worker job's payload. On skipped ticks (inactive schedule, dev
   env disconnected, etc.) the engine still re-registered with
   `fromTimestamp = scheduleTimestamp`, so a long pause/disconnect would
   quietly overwrite the real last-fire time with a stream of skipped
   slots — defeating the workerCatalog accuracy benefit for the very case
   it was meant to handle.

   `RegisterScheduleInstanceParams` now has a separate optional
   `lastScheduleTime` field. `fromTimestamp` advances on every tick;
   `lastScheduleTime` only advances on real fires. After a skip the
   re-registration carries forward the existing `params.lastScheduleTime`
   (or the legacy `instance.lastScheduledTimestamp` fallback).

2. ScheduleListPresenter — guard cron back-calculation
   `previousScheduledTimestamp(...)` calls cron-parser, which throws on
   malformed expressions. A single bad row would have failed the whole
   schedules-list response. Wrapped per-row in try/catch so a degraded
   row falls back to `lastRun: undefined` instead of taking down the page.

3. scheduleEngine integration test — anchor on observed values
   The test compared against a precomputed "next minute boundary" Date,
   which is flaky if the test setup straddles a minute boundary. Switched
   to deriving expectations from the first observed `exactScheduleTime`
   and asserting relative invariants (60s gap, second lastTimestamp
   equals first timestamp).

Author: ericallam

apps/webapp/app/presenters/v3/ScheduleListPresenter.server.ts

@@ -249,12 +249,19 @@ export class ScheduleListPresenter extends BasePresenter {
       // Approximate "last run" from the cron's previous slot. If that slot
       // predates the schedule itself, the schedule hasn't fired yet — show
       // undefined rather than a misleading timestamp. UI is best-effort;
-      // accurate run history is on the schedule's runs page.
-      const cronPrev = previousScheduledTimestamp(
-        schedule.generatorExpression,
-        schedule.timezone
-      );
-      const lastRun = cronPrev.getTime() > schedule.createdAt.getTime() ? cronPrev : undefined;
+      // accurate run history is on the schedule's runs page. cron-parser
+      // throws on malformed expressions, so degrade to undefined per-row
+      // rather than failing the whole list.
+      let lastRun: Date | undefined;
+      try {
+        const cronPrev = previousScheduledTimestamp(
+          schedule.generatorExpression,
+          schedule.timezone
+        );
+        lastRun = cronPrev.getTime() > schedule.createdAt.getTime() ? cronPrev : undefined;
+      } catch {
+        lastRun = undefined;
+      }
 
       return {
         id: schedule.id,

internal-packages/schedule-engine/src/engine/index.ts

@@ -194,15 +194,16 @@ export class ScheduleEngine {
           timezone: instance.taskSchedule.timezone,
         });
 
-        // Enqueue the scheduled task. Pass fromTimestamp through as the next
-        // job's lastScheduleTime so the dequeueing engine can populate
-        // payload.lastTimestamp without re-reading state from the DB. When
-        // fromTimestamp is undefined (first-ever registration / recovery),
-        // lastScheduleTime is also undefined → consumer reports undefined.
+        // Enqueue the scheduled task. The next job's `lastScheduleTime`
+        // payload is the *actual* previous fire time (passed in by the
+        // caller), not `fromTimestamp` — `fromTimestamp` advances on every
+        // tick (including skips) so it can't be used as the previous-fire
+        // anchor without leaking skipped slots into customer-visible
+        // payload.lastTimestamp.
         await this.enqueueScheduledTask(
           params.instanceId,
           nextScheduledTimestamp,
-          params.fromTimestamp
+          params.lastScheduleTime
         );
 
         // Record metrics
@@ -535,13 +536,22 @@ export class ScheduleEngine {
           });
         }
 
-        // Register the next run, calculating from the timestamp we just fired (or
-        // skipped) so we don't need to round-trip through DB state.
-        // Rewritten try/catch to use tryCatch utility
+        // Register the next run. `fromTimestamp` advances on every tick so
+        // the next cron slot keeps marching forward even through skips.
+        // `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
+        // a long pause/disconnect doesn't quietly overwrite the real
+        // last-fire timestamp with a series of skipped slots.
+        const carriedLastScheduleTime = shouldTrigger
+          ? scheduleTimestamp
+          : params.lastScheduleTime ?? instance.lastScheduledTimestamp ?? undefined;
+
         const [nextRunError] = await tryCatch(
           this.registerNextTaskScheduleInstance({
             instanceId: params.instanceId,
             fromTimestamp: scheduleTimestamp,
+            lastScheduleTime: carriedLastScheduleTime,
           })
         );
         if (nextRunError) {

internal-packages/schedule-engine/src/engine/types.ts

@@ -79,5 +79,17 @@ 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.
+   */
   fromTimestamp?: Date;
+  /**
+   * The actual previous fire time to embed in the next worker job's payload,
+   * which becomes that job's `payload.lastTimestamp` on dequeue. Distinct
+   * from `fromTimestamp` so that skipped ticks (inactive schedule, dev env
+   * disconnected, etc.) do NOT advance this — only real fires do.
+   */
+  lastScheduleTime?: Date;
 }

internal-packages/schedule-engine/test/scheduleEngine.test.ts

@@ -98,23 +98,11 @@ describe("ScheduleEngine Integration", () => {
           },
         });
 
-        // Calculate the expected next execution time (next minute boundary)
-        const now = new Date();
-        const expectedExecutionTime = new Date(now);
-        expectedExecutionTime.setMinutes(now.getMinutes() + 1, 0, 0); // Next minute, 0 seconds, 0 milliseconds
-
-        // Calculate the expected upcoming execution times (next 10 minutes after the first execution)
-        const expectedUpcoming = [];
-        for (let i = 1; i <= 10; i++) {
-          const upcoming = new Date(expectedExecutionTime);
-          upcoming.setMinutes(expectedExecutionTime.getMinutes() + i);
-          expectedUpcoming.push(upcoming);
-        }
-
         // Manually enqueue the first scheduled task to kick off the lifecycle.
-        // Engine no longer persists nextScheduledTimestamp — the same time can be
-        // reproduced from the cron expression alone, so we use expectedExecutionTime
-        // directly downstream.
+        // Anchor expectations to the first observed `exactScheduleTime` rather
+        // than a precomputed wall-clock value — registration that happens to
+        // straddle a minute boundary used to flake tests asserting against a
+        // pre-baked "next minute".
         await engine.registerNextTaskScheduleInstance({ instanceId: scheduleInstance.id });
 
         // Wait for the first execution
@@ -176,6 +164,17 @@ describe("ScheduleEngine Integration", () => {
           }
         }
 
+        // Anchor all expectations to what the engine actually fired with, so
+        // the test stays deterministic regardless of when within a minute it
+        // started.
+        const firstScheduledTime = firstExecution.params.exactScheduleTime;
+        const secondScheduledTime = secondExecution.params.exactScheduleTime;
+        expect(firstScheduledTime).toBeDefined();
+        expect(secondScheduledTime).toBeDefined();
+
+        // Each cron slot for "* * * * *" is exactly 60s apart.
+        expect(secondScheduledTime!.getTime() - firstScheduledTime!.getTime()).toBe(60_000);
+
         // Verify the first execution parameters
         expect(firstExecution.params).toEqual({
           taskIdentifier: "test-task",
@@ -196,24 +195,23 @@ describe("ScheduleEngine Integration", () => {
           payload: {
             scheduleId: "sched_abc123",
             type: "DECLARATIVE",
-            timestamp: expectedExecutionTime,
-            // First-ever fire: cron's previous slot predates instance.createdAt
-            // (which was set ~now), so lastTimestamp is undefined. This preserves
-            // the `if (!payload.lastTimestamp)` first-run sentinel customers rely on.
+            timestamp: firstScheduledTime,
+            // First-ever fire: no `lastScheduleTime` carried in the worker
+            // payload and `instance.lastScheduledTimestamp` is null on a
+            // fresh instance, so lastTimestamp is undefined. This preserves
+            // the `if (!payload.lastTimestamp)` first-run sentinel customers
+            // rely on.
             lastTimestamp: undefined,
             externalId: "ext-123",
             timezone: "UTC",
             upcoming: expect.arrayContaining([expect.any(Date)]),
           },
           scheduleInstanceId: scheduleInstance.id,
           scheduleId: taskSchedule.id,
-          exactScheduleTime: expectedExecutionTime,
+          exactScheduleTime: firstScheduledTime,
         });
 
         // Verify the second execution parameters
-        const expectedSecondExecution = new Date(expectedExecutionTime);
-        expectedSecondExecution.setMinutes(expectedExecutionTime.getMinutes() + 1);
-
         expect(secondExecution.params).toEqual({
           taskIdentifier: "test-task",
           environment: expect.objectContaining({
@@ -223,16 +221,17 @@ describe("ScheduleEngine Integration", () => {
           payload: {
             scheduleId: "sched_abc123",
             type: "DECLARATIVE",
-            timestamp: expectedSecondExecution,
-            // Previous slot before second execution = first execution time.
-            lastTimestamp: expectedExecutionTime,
+            timestamp: secondScheduledTime,
+            // The previous fire's exactScheduleTime is carried through the
+            // worker payload as `lastScheduleTime` and surfaced here.
+            lastTimestamp: firstScheduledTime,
             externalId: "ext-123",
             timezone: "UTC",
             upcoming: expect.arrayContaining([expect.any(Date)]),
           },
           scheduleInstanceId: scheduleInstance.id,
           scheduleId: taskSchedule.id,
-          exactScheduleTime: expectedSecondExecution,
+          exactScheduleTime: secondScheduledTime,
         });
       } finally {
         // Clean up: stop the worker