fix(cron): clear future orphan run markers (#111743)

Co-authored-by: Peter Steinberger <steipete@gmail.com>

src/cron/service.overdue-scheduling.test.ts

@@ -2,6 +2,7 @@
 import { describe, expect, it } from "vitest";
 import { createMockCronStateForJobs } from "./service.test-harness.js";
 import { recomputeNextRunsForMaintenance } from "./service/jobs.js";
+import { reserveQueuedCronRun } from "./service/run-admission.js";
 import type { CronJob } from "./types.js";
 
 function createCronSystemEventJob(now: number, overrides: Partial<CronJob> = {}): CronJob {
@@ -155,6 +156,82 @@ describe("issue #13992 regression - cron jobs skip execution", () => {
     expect(job.state.nextRunAtMs).toBe(futureTime);
   });
 
+  it("clears an orphaned queued marker from before a clock rollback", () => {
+    const now = Date.now();
+    const futureQueuedAt = now + 3 * 60 * 60_000;
+
+    const job = createCronSystemEventJob(now, {
+      state: {
+        nextRunAtMs: now + 60_000,
+        queuedAtMs: futureQueuedAt,
+      },
+    });
+
+    const state = createMockCronStateForJobs({ jobs: [job], nowMs: now });
+    recomputeNextRunsForMaintenance(state);
+
+    expect(job.state.queuedAtMs).toBeUndefined();
+  });
+
+  it("clears an orphaned running marker from before a clock rollback", () => {
+    const now = Date.now();
+    const pastDue = now - 60_000;
+    const futureRunningAt = now + 3 * 60 * 60_000;
+
+    const job = createCronSystemEventJob(now, {
+      state: {
+        nextRunAtMs: pastDue,
+        runningAtMs: futureRunningAt,
+        lastRunAtMs: pastDue - 60_000,
+      },
+    });
+
+    const state = createMockCronStateForJobs({ jobs: [job], nowMs: now });
+    recomputeNextRunsForMaintenance(state, { recomputeExpired: true });
+
+    expect(job.state.runningAtMs).toBeUndefined();
+    expect(job.state.nextRunAtMs).toBe(pastDue);
+  });
+
+  it.each(["queuedAtMs", "runningAtMs"] as const)(
+    "preserves a future %s marker owned by a live reservation",
+    (markerField) => {
+      const now = Date.now();
+      const futureMarker = now + 3 * 60 * 60_000;
+      const job = createCronSystemEventJob(now, {
+        state: {
+          nextRunAtMs: now + 60_000,
+          [markerField]: futureMarker,
+        },
+      });
+      const state = createMockCronStateForJobs({ jobs: [job], nowMs: now });
+      reserveQueuedCronRun(state, job.id, futureMarker);
+
+      recomputeNextRunsForMaintenance(state);
+
+      expect(job.state[markerField]).toBe(futureMarker);
+    },
+  );
+
+  it.each(["queuedAtMs", "runningAtMs"] as const)(
+    "preserves a near-future %s marker within the stale-run window",
+    (markerField) => {
+      const now = Date.now();
+      const futureMarker = now + 1_000;
+      const job = createCronSystemEventJob(now, {
+        state: {
+          nextRunAtMs: now + 60_000,
+          [markerField]: futureMarker,
+        },
+      });
+      const state = createMockCronStateForJobs({ jobs: [job], nowMs: now });
+
+      recomputeNextRunsForMaintenance(state);
+
+      expect(job.state[markerField]).toBe(futureMarker);
+    },
+  );
+
   it("isolates schedule errors while filling missing nextRunAtMs", () => {
     const now = Date.now();
     const pastDue = now - 1_000;

src/cron/service/jobs-scheduling.ts

@@ -505,7 +505,7 @@ function normalizeJobTickState(params: { state: CronServiceState; job: CronJob;
   const queuedAt = job.state.queuedAtMs;
   if (
     typeof queuedAt === "number" &&
-    nowMs - queuedAt > STUCK_RUN_MS &&
+    Math.abs(nowMs - queuedAt) > STUCK_RUN_MS &&
     !isQueuedCronRun(state, job.id, queuedAt)
   ) {
     state.deps.log.warn(
@@ -519,7 +519,7 @@ function normalizeJobTickState(params: { state: CronServiceState; job: CronJob;
   const runningAt = job.state.runningAtMs;
   if (
     typeof runningAt === "number" &&
-    nowMs - runningAt > STUCK_RUN_MS &&
+    Math.abs(nowMs - runningAt) > STUCK_RUN_MS &&
     !isQueuedCronRun(state, job.id, runningAt)
   ) {
     state.deps.log.warn(