fix(cron): preserve exact restart boundaries

steipete · NianJiuZst · steipete · commit 992b50a9b9b7 · 2026-07-18T01:52:29.000+01:00
Co-authored-by: 李兰 0668001394 &lt;li.lan3@xydigit.com&gt;

Co-authored-by: NianJiuZst &lt;180004567+NianJiuZst@users.noreply.github.com&gt;
diff --git a/src/cron/service.jobs.test.ts b/src/cron/service.jobs.test.ts
@@ -2,6 +2,8 @@
 import { describe, expect, it } from "vitest";
 import {
   applyJobPatch,
+  computeJobNextRunAtMs,
+  computeJobPreviousRunAtOrBeforeMs,
   createJob,
   nextWakeAtMs,
   recomputeNextRuns,
@@ -1094,6 +1096,45 @@ describe("cron stagger defaults", () => {
   });
 });
 
+describe("computeJobPreviousRunAtOrBeforeMs", () => {
+  function createCronJob(schedule: Extract<CronJob["schedule"], { kind: "cron" }>): CronJob {
+    return {
+      id: "inclusive-previous-run",
+      name: "inclusive previous run",
+      enabled: true,
+      createdAtMs: 0,
+      updatedAtMs: 0,
+      schedule,
+      sessionTarget: "main",
+      wakeMode: "now",
+      payload: { kind: "systemEvent", text: "tick" },
+      state: {},
+    };
+  }
+
+  it("includes an exact boundary and keeps the prior slot between boundaries", () => {
+    const job = createCronJob({ kind: "cron", expr: "* * * * * *", tz: "UTC", staggerMs: 0 });
+    const boundary = Date.parse("2025-12-13T04:02:00.000Z");
+
+    expect(computeJobPreviousRunAtOrBeforeMs(job, boundary)).toBe(boundary);
+    expect(computeJobPreviousRunAtOrBeforeMs(job, boundary + 500)).toBe(boundary);
+  });
+
+  it("includes an exact effective boundary after per-job staggering", () => {
+    const job = createCronJob({
+      kind: "cron",
+      expr: "0 * * * * *",
+      tz: "UTC",
+      staggerMs: 30_000,
+    });
+    const cursor = Date.parse("2025-12-13T04:02:00.000Z");
+    const effectiveBoundary = computeJobNextRunAtMs(job, cursor);
+
+    expect(effectiveBoundary).toBeTypeOf("number");
+    expect(computeJobPreviousRunAtOrBeforeMs(job, effectiveBoundary!)).toBe(effectiveBoundary);
+  });
+});
+
 describe("createJob delivery defaults", () => {
   const now = Date.parse("2026-02-28T12:00:00.000Z");
 
diff --git a/src/cron/service.restart-catchup.test.ts b/src/cron/service.restart-catchup.test.ts
@@ -270,6 +270,116 @@ describe("CronService restart catch-up", () => {
     }
   });
 
+  it("does not defer an isolated cron agent-turn whose persisted due slot already succeeded", async () => {
+    const store = await makeStorePath();
+    const startNow = Date.parse("2025-12-13T11:00:00.000Z");
+    const dueAt = Date.parse("2025-12-13T09:10:00.000Z");
+    const completedAt = Date.parse("2025-12-13T09:10:30.000Z");
+    const runIsolatedAgentJob = vi.fn(async () => ({ status: "ok" as const }));
+    const enqueueSystemEvent = vi.fn();
+    const requestHeartbeat = vi.fn();
+
+    await writeStoreJobs(store.storePath, [
+      {
+        id: "startup-isolated-agent-already-ok",
+        name: "startup isolated agent already ok",
+        enabled: true,
+        createdAtMs: Date.parse("2025-12-10T12:00:00.000Z"),
+        updatedAtMs: completedAt,
+        schedule: { kind: "cron", expr: "10 9 * * *", tz: "UTC" },
+        sessionTarget: "isolated",
+        wakeMode: "next-heartbeat",
+        payload: { kind: "agentTurn", message: "daily reminder" },
+        state: {
+          nextRunAtMs: dueAt,
+          lastRunAtMs: completedAt,
+          lastRunStatus: "ok",
+        },
+      },
+    ]);
+
+    const cron = createRestartCronService({
+      storePath: store.storePath,
+      enqueueSystemEvent,
+      requestHeartbeat,
+      runIsolatedAgentJob,
+      nowMs: () => startNow,
+      startupDeferredMissedAgentJobDelayMs: 120_000,
+    });
+
+    try {
+      await cron.start();
+
+      expect(runIsolatedAgentJob).not.toHaveBeenCalled();
+      expect(enqueueSystemEvent).not.toHaveBeenCalled();
+      expect(requestHeartbeat).not.toHaveBeenCalled();
+
+      const listedJobs = await cron.list({ includeDisabled: true });
+      const updated = listedJobs.find((job) => job.id === "startup-isolated-agent-already-ok");
+      expect(updated?.state.lastRunStatus).toBe("ok");
+      expect(updated?.state.nextRunAtMs).toBe(Date.parse("2025-12-14T09:10:00.000Z"));
+    } finally {
+      cron.stop();
+      await store.cleanup();
+    }
+  });
+
+  it("replays a newer missed cron slot behind a completed persisted slot", async () => {
+    vi.setSystemTime(new Date("2025-12-13T04:10:00.000Z"));
+    await withRestartedCron(
+      [
+        {
+          id: "restart-completed-slot-newer-miss",
+          name: "completed slot with newer miss",
+          enabled: true,
+          createdAtMs: Date.parse("2025-12-10T12:00:00.000Z"),
+          updatedAtMs: Date.parse("2025-12-13T04:01:30.000Z"),
+          schedule: { kind: "cron", expr: "* * * * *", tz: "UTC" },
+          sessionTarget: "main",
+          wakeMode: "next-heartbeat",
+          payload: { kind: "systemEvent", text: "newer slot missed" },
+          state: {
+            nextRunAtMs: Date.parse("2025-12-13T04:01:00.000Z"),
+            lastRunAtMs: Date.parse("2025-12-13T04:01:00.000Z"),
+            lastRunStatus: "ok",
+          },
+        },
+      ],
+      async ({ enqueueSystemEvent, requestHeartbeat }) => {
+        expectQueuedSystemEvent(enqueueSystemEvent, "newer slot missed");
+        expect(requestHeartbeat).toHaveBeenCalled();
+      },
+    );
+  });
+
+  it("replays a cron slot due exactly at restart behind a completed persisted slot", async () => {
+    vi.setSystemTime(new Date("2025-12-13T04:02:00.000Z"));
+    await withRestartedCron(
+      [
+        {
+          id: "restart-completed-slot-boundary-miss",
+          name: "completed slot with boundary miss",
+          enabled: true,
+          createdAtMs: Date.parse("2025-12-10T12:00:00.000Z"),
+          updatedAtMs: Date.parse("2025-12-13T04:01:30.000Z"),
+          schedule: { kind: "cron", expr: "* * * * *", tz: "UTC" },
+          sessionTarget: "main",
+          wakeMode: "next-heartbeat",
+          payload: { kind: "systemEvent", text: "boundary slot missed" },
+          state: {
+            nextRunAtMs: Date.parse("2025-12-13T04:01:00.000Z"),
+            lastRunAtMs: Date.parse("2025-12-13T04:01:00.000Z"),
+            lastRunStatus: "ok",
+          },
+        },
+      ],
+      async ({ enqueueSystemEvent, requestHeartbeat }) => {
+        expectQueuedSystemEvent(enqueueSystemEvent, "boundary slot missed");
+        expect(requestHeartbeat).toHaveBeenCalled();
+      },
+    );
+  });
+
   it("marks interrupted recurring jobs failed instead of replaying them on startup", async () => {
     const dueAt = Date.parse("2025-12-13T16:00:00.000Z");
     const staleRunningAt = Date.parse("2025-12-13T16:30:00.000Z");
diff --git a/src/cron/service/jobs.ts b/src/cron/service/jobs.ts
@@ -181,17 +181,32 @@ function computeStaggeredCronPreviousRunAtMs(job: CronJob, nowMs: number) {
   return undefined;
 }
 
+function computeStaggeredCronPreviousRunAtOrBeforeMs(job: CronJob, nowMs: number) {
+  const previous = computeStaggeredCronPreviousRunAtMs(job, nowMs);
+  const probeMs = nowMs + 1_000;
+  if (!Number.isFinite(probeMs)) {
+    return previous;
+  }
+
+  // Croner previous-run queries are strict-before and second-granular. Keep
+  // the strict result, then probe past the current second to include a slot
+  // exactly at now without losing the prior slot between boundaries.
+  const boundary = computeStaggeredCronPreviousRunAtMs(job, probeMs);
+  if (
+    isFiniteTimestamp(boundary) &&
+    boundary <= nowMs &&
+    (!isFiniteTimestamp(previous) || boundary > previous)
+  ) {
+    return boundary;
+  }
+  return previous;
+}
+
 function isStaggeredCronRunAtMs(job: CronJob, runAtMs: number): boolean {
   if (job.schedule.kind !== "cron" || !isFiniteTimestamp(runAtMs)) {
     return false;
   }
-  // Probe past the candidate second. Croner-style second-granular schedules
-  // normalize a 1ms probe back to the candidate's second, so
-  // `previousRuns(1, runAtMs + 1)` returns the slot before the candidate
-  // rather than the candidate itself and exact-second slots get misclassified
-  // as stale. A 1s probe lands past the candidate second, matching the cursor
-  // step used elsewhere in this file (cf. #81691).
-  const previous = computeStaggeredCronPreviousRunAtMs(job, runAtMs + 1_000);
+  const previous = computeStaggeredCronPreviousRunAtOrBeforeMs(job, runAtMs);
   return previous === runAtMs;
 }
 
@@ -510,6 +525,15 @@ export function computeJobPreviousRunAtMs(job: CronJob, nowMs: number): number |
   return isFiniteTimestamp(previous) ? previous : undefined;
 }
 
+/** Computes the latest effective cron timestamp at or before the supplied time. */
+export function computeJobPreviousRunAtOrBeforeMs(job: CronJob, nowMs: number): number | undefined {
+  if (!isJobEnabled(job) || job.schedule.kind !== "cron") {
+    return undefined;
+  }
+  const previous = computeStaggeredCronPreviousRunAtOrBeforeMs(job, nowMs);
+  return isFiniteTimestamp(previous) ? previous : undefined;
+}
+
 /** Maximum consecutive schedule errors before auto-disabling a job. */
 const MAX_SCHEDULE_ERRORS = 3;
 
diff --git a/src/cron/service/timer.ts b/src/cron/service/timer.ts
@@ -69,6 +69,7 @@ import {
 } from "./failure-alerts.js";
 import {
   DEFAULT_ERROR_BACKOFF_SCHEDULE_MS,
+  computeJobPreviousRunAtOrBeforeMs,
   computeJobPreviousRunAtMs,
   computeJobNextRunAtMs,
   errorBackoffMs,
@@ -1833,7 +1834,26 @@ function isRunnableJob(params: {
     return false;
   }
   if (hasScheduledNextRunAtMs(next) && nowMs >= next) {
-    return true;
+    const lastRunAtMs = job.state.lastRunAtMs;
+    // Startup loads persisted state before maintenance recompute. Suppress a
+    // completed stale slot, but still replay a newer slot due by restart time.
+    const alreadyCompletedDueCronSlot =
+      params.allowCronMissedRunByLastRun &&
+      job.schedule.kind === "cron" &&
+      lastRunStatus === "ok" &&
+      typeof lastRunAtMs === "number" &&
+      Number.isFinite(lastRunAtMs) &&
+      lastRunAtMs >= next;
+    if (!alreadyCompletedDueCronSlot) {
+      return true;
+    }
+    let latestRunAtMs: number | undefined;
+    try {
+      latestRunAtMs = computeJobPreviousRunAtOrBeforeMs(job, nowMs);
+    } catch {
+      return false;
+    }
+    return typeof latestRunAtMs === "number" && latestRunAtMs > lastRunAtMs;
   }
   if (!params.allowCronMissedRunByLastRun || job.schedule.kind !== "cron") {
     return false;
