From 3ec5ee6bb62d64ec05006bbd53f5a66f45d0dbcd Mon Sep 17 00:00:00 2001 From: LiLan0125 Date: Sat, 18 Jul 2026 09:19:00 +0800 Subject: [PATCH] fix(cron): skip completed restart catchup slots (#101998) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * fix(cron): preserve exact restart boundaries Co-authored-by: 李兰 0668001394 Co-authored-by: NianJiuZst <180004567+NianJiuZst@users.noreply.github.com> * fix(cron): suppress skipped restart slots --------- Co-authored-by: Peter Steinberger Co-authored-by: NianJiuZst <180004567+NianJiuZst@users.noreply.github.com> --- src/cron/service.jobs.test.ts | 41 ++++++++ src/cron/service.restart-catchup.test.ts | 115 +++++++++++++++++++++++ src/cron/service/jobs.ts | 38 ++++++-- src/cron/service/timer.ts | 22 ++++- 4 files changed, 208 insertions(+), 8 deletions(-) diff --git a/src/cron/service.jobs.test.ts b/src/cron/service.jobs.test.ts index f3b13d5943ee..007e50dffab8 100644 --- 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 { + 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 index 3f4c137fa90e..ca8da8ba519f 100644 --- a/src/cron/service.restart-catchup.test.ts +++ b/src/cron/service.restart-catchup.test.ts @@ -270,6 +270,121 @@ describe("CronService restart catch-up", () => { } }); + it.each(["ok", "skipped"] as const)( + "does not defer an isolated cron agent-turn whose persisted due slot finished as %s", + async (lastRunStatus) => { + 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-${lastRunStatus}`, + name: `startup isolated agent already ${lastRunStatus}`, + 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, + }, + }, + ]); + + 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-${lastRunStatus}`, + ); + expect(updated?.state.lastRunStatus).toBe(lastRunStatus); + 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 index 43aee8eb2dd8..0e809c39236b 100644 --- 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 index c3376d4598a6..8682fcafaa6a 100644 --- 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" || lastRunStatus === "skipped") && + 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;