diff --git a/src/cron/public-job.ts b/src/cron/public-job.ts index f7e0500b62b3..d4630f3d7b56 100644 --- a/src/cron/public-job.ts +++ b/src/cron/public-job.ts @@ -4,5 +4,6 @@ import type { CronJob } from "./types.js"; export function toPublicCronJob(job: CronJob): CronJob { const state = { ...job.state }; delete state.queuedAtMs; + delete state.startupCatchupAtMs; return { ...job, state }; } diff --git a/src/cron/service.jobs.test.ts b/src/cron/service.jobs.test.ts index 007e50dffab8..495ae0af8328 100644 --- a/src/cron/service.jobs.test.ts +++ b/src/cron/service.jobs.test.ts @@ -823,7 +823,6 @@ function createMockState(now: number, opts?: { defaultAgentId?: string }): CronS nowMs: () => now, defaultAgentId: opts?.defaultAgentId, }, - pendingCatchupDeferralJobIds: new Set(), } as unknown as CronServiceState; } @@ -1349,18 +1348,16 @@ describe("recomputeNextRuns", () => { sessionTarget: "main", wakeMode: "now", payload: { kind: "systemEvent", text: "tick" }, - state: { nextRunAtMs: deferred }, + state: { nextRunAtMs: deferred, startupCatchupAtMs: deferred }, }; - const pendingCatchupDeferralJobIds = new Set([job.id]); const state = { ...createMockState(now), - pendingCatchupDeferralJobIds, store: { version: 1 as const, jobs: [job] }, } as CronServiceState; expect(recomputeNextRunsForMaintenance(state)).toBe(false); expect(job.state.nextRunAtMs).toBe(deferred); - expect(pendingCatchupDeferralJobIds.has(job.id)).toBe(true); + expect(job.state.startupCatchupAtMs).toBe(deferred); expect( recomputeNextRunsForMaintenance(state, { @@ -1368,11 +1365,11 @@ describe("recomputeNextRuns", () => { repairFutureCronNextRunAtMs: true, }), ).toBe(true); - expect(pendingCatchupDeferralJobIds.has(job.id)).toBe(false); + expect(job.state.startupCatchupAtMs).toBeUndefined(); expect(job.state.nextRunAtMs).toBe(deferred); }); - it("drops startup catch-up deferral ids for jobs no longer relevant to maintenance", () => { + it("drops startup catch-up deferrals for disabled jobs", () => { const now = Date.parse("2026-05-05T12:00:00.000Z"); const deferred = Date.parse("2026-05-05T12:02:00.000Z"); const disabledJob: CronJob = { @@ -1385,17 +1382,15 @@ describe("recomputeNextRuns", () => { sessionTarget: "main", wakeMode: "now", payload: { kind: "systemEvent", text: "tick" }, - state: { nextRunAtMs: deferred }, + state: { nextRunAtMs: deferred, startupCatchupAtMs: deferred }, }; - const pendingCatchupDeferralJobIds = new Set([disabledJob.id, "removed-deferral"]); const state = { ...createMockState(now), - pendingCatchupDeferralJobIds, store: { version: 1 as const, jobs: [disabledJob] }, } as CronServiceState; expect(recomputeNextRunsForMaintenance(state)).toBe(true); - expect([...pendingCatchupDeferralJobIds]).toEqual([]); + expect(disabledJob.state.startupCatchupAtMs).toBeUndefined(); expect(disabledJob.state.nextRunAtMs).toBeUndefined(); }); diff --git a/src/cron/service.startup-overflow-clobber.test.ts b/src/cron/service.startup-overflow-clobber.test.ts index 34eb1817a23f..24ed594e931e 100644 --- a/src/cron/service.startup-overflow-clobber.test.ts +++ b/src/cron/service.startup-overflow-clobber.test.ts @@ -42,7 +42,7 @@ describe("CronService startup catch-up repair scoping", () => { }; } - it("keeps the overflow daily-cron catch-up deferral after start()'s maintenance pass", async () => { + it("keeps the overflow daily-cron catch-up deferral across a second restart", async () => { const store = await makeStorePath(); const startNow = Date.parse("2025-12-13T17:00:00.000Z"); let now = startNow; @@ -60,15 +60,17 @@ describe("CronService startup catch-up repair scoping", () => { ], }); - const state = createCronServiceState({ - cronEnabled: true, - storePath: store.storePath, - log: noopLogger, - nowMs: () => now, - enqueueSystemEvent: vi.fn(), - requestHeartbeat: vi.fn(), - runIsolatedAgentJob: vi.fn(async () => ({ status: "ok" as const })), - }); + const createState = () => + createCronServiceState({ + cronEnabled: true, + storePath: store.storePath, + log: noopLogger, + nowMs: () => now, + enqueueSystemEvent: vi.fn(), + requestHeartbeat: vi.fn(), + runIsolatedAgentJob: vi.fn(async () => ({ status: "ok" as const })), + }); + const state = createState(); await start(state); @@ -76,20 +78,36 @@ describe("CronService startup catch-up repair scoping", () => { expect(deferred?.state.nextRunAtMs).toBe(startNow + 5_000); expect(deferred?.state.nextRunAtMs).not.toBe(tomorrowNaturalSlot); - expect(state.pendingCatchupDeferralJobIds.has("daily-overflow")).toBe(true); + expect(deferred?.state.startupCatchupAtMs).toBe(startNow + 5_000); await status(state); expect(deferred?.state.nextRunAtMs).toBe(startNow + 5_000); - now = startNow + 5_005; - await onTimer(state); + if (state.timer) { + clearTimeout(state.timer); + } + state.stopped = true; + now = startNow + 3_000; - const completed = state.store?.jobs.find((job) => job.id === "daily-overflow"); + const restartedState = createState(); + await start(restartedState); + + const restarted = restartedState.store?.jobs.find((job) => job.id === "daily-overflow"); + expect(restarted?.state.nextRunAtMs).toBe(startNow + 5_000); + expect(restarted?.state.startupCatchupAtMs).toBe(startNow + 5_000); + + now = startNow + 5_005; + await onTimer(restartedState); + + const completed = restartedState.store?.jobs.find((job) => job.id === "daily-overflow"); expect(completed?.state.lastRunStatus).toBe("ok"); expect(completed?.state.nextRunAtMs).toBe(tomorrowNaturalSlot); - expect(state.pendingCatchupDeferralJobIds.has("daily-overflow")).toBe(false); + expect(completed?.state.startupCatchupAtMs).toBeUndefined(); - state.stopped = true; + if (restartedState.timer) { + clearTimeout(restartedState.timer); + } + restartedState.stopped = true; await store.cleanup(); }); diff --git a/src/cron/service.test-harness.ts b/src/cron/service.test-harness.ts index bd0aa6eaefe8..084d17463083 100644 --- a/src/cron/service.test-harness.ts +++ b/src/cron/service.test-harness.ts @@ -239,7 +239,6 @@ export function createMockCronStateForJobs(params: { schedulingPaused: false, schedulerStarted: false, restartRecoveryPending: false, - pendingCatchupDeferralJobIds: new Set(), activeManualRunJobIds: new Set(), manualSetupTimeoutNotified: false, runAdmission: { active: 0, waiters: [] }, diff --git a/src/cron/service/jobs.ts b/src/cron/service/jobs.ts index 0e809c39236b..cf2e72a60e1c 100644 --- a/src/cron/service/jobs.ts +++ b/src/cron/service/jobs.ts @@ -617,6 +617,10 @@ function normalizeJobTickState(params: { state: CronServiceState; job: CronJob; } if (!isJobEnabled(job)) { + if (job.state.startupCatchupAtMs !== undefined) { + job.state.startupCatchupAtMs = undefined; + changed = true; + } if (job.state.nextRunAtMs !== undefined) { job.state.nextRunAtMs = undefined; changed = true; @@ -792,32 +796,23 @@ export function recomputeNextRunsForMaintenance( nowMs, deferredAutoDisableNotifications: opts?.deferredAutoDisableNotifications, }); - const deferralIds = state.pendingCatchupDeferralJobIds; - // Drop deferral markers for jobs that no longer exist in the store or - // are disabled. They will not fire, so no deferral is needed. - if (state.store && deferralIds.size > 0) { - const relevantDeferralIds = new Set( - state.store.jobs.filter((job) => isJobEnabled(job)).map((job) => job.id), - ); - for (const jobId of deferralIds) { - if (!relevantDeferralIds.has(jobId)) { - deferralIds.delete(jobId); - } - } - } return walkSchedulableJobs( state, ({ job, nowMs: now }) => { let changed = false; - // Clear stale deferral markers once the deferred staggered slot arrives. - // After the slot fires, future repair is safe for this job again. - if (deferralIds.has(job.id)) { - const nextRun = job.state.nextRunAtMs; - if (hasScheduledNextRunAtMs(nextRun) && now >= nextRun) { - deferralIds.delete(job.id); - changed = true; - } + const startupCatchupAtMs = job.state.startupCatchupAtMs; + const nextRunAtMs = job.state.nextRunAtMs; + // The persisted marker owns only its exact future slot. Schedule edits, + // malformed state, or arrival at the slot release normal repair policy. + const hasPendingStartupCatchup = + isFiniteTimestamp(startupCatchupAtMs) && + hasScheduledNextRunAtMs(nextRunAtMs) && + startupCatchupAtMs === nextRunAtMs && + now < startupCatchupAtMs; + if (startupCatchupAtMs !== undefined && !hasPendingStartupCatchup) { + job.state.startupCatchupAtMs = undefined; + changed = true; } if (!hasScheduledNextRunAtMs(job.state.nextRunAtMs)) { @@ -826,7 +821,7 @@ export function recomputeNextRunsForMaintenance( } } else if ( repairFutureCronNextRunAtMs && - !deferralIds.has(job.id) && + !hasPendingStartupCatchup && shouldRepairFutureCronNextRunAtMs({ state, job, nowMs: now }) ) { if (recomputeJob(job, now)) { diff --git a/src/cron/service/ops.test.ts b/src/cron/service/ops.test.ts index 5cf31f9769f9..1d7a846eaf00 100644 --- a/src/cron/service/ops.test.ts +++ b/src/cron/service/ops.test.ts @@ -1294,7 +1294,7 @@ describe("cron service ops persist rollback", () => { expect(loaded.jobs.map((entry) => entry.id)).toEqual([job.id]); }); - it("keeps a job's catch-up deferral marker when a remove persist fails", async () => { + it("restores a job's catch-up deferral when a remove persist fails", async () => { const { storePath } = await makeStorePath(); const now = Date.parse("2026-06-09T00:00:00.000Z"); const state = createOkIsolatedCronState({ storePath, now }); @@ -1303,13 +1303,13 @@ describe("cron service ops persist rollback", () => { if (state.timer) { clearTimeout(state.timer); } - state.pendingCatchupDeferralJobIds.add(job.id); + job.state.startupCatchupAtMs = now + 5_000; vi.spyOn(cronStoreModule, "saveCronJobsStore").mockRejectedValueOnce(new Error("disk full")); await expect(remove(state, job.id)).rejects.toThrow("disk full"); - expect(state.pendingCatchupDeferralJobIds.has(job.id)).toBe(true); + expect(state.store?.jobs[0]?.state.startupCatchupAtMs).toBe(now + 5_000); expect(state.store?.jobs.map((entry) => entry.id)).toEqual([job.id]); }); diff --git a/src/cron/service/ops.ts b/src/cron/service/ops.ts index 7d8eaa574c9b..4fcb2c4595d1 100644 --- a/src/cron/service/ops.ts +++ b/src/cron/service/ops.ts @@ -500,6 +500,7 @@ function finalizeUpdatedJob(params: { nextJob.updatedAtMs = now; if (schedulingInputsChanged) { + nextJob.state.startupCatchupAtMs = undefined; if (isJobEnabled(nextJob)) { nextJob.state.nextRunAtMs = computeJobNextRunAtMs(nextJob, now); } else { diff --git a/src/cron/service/state.ts b/src/cron/service/state.ts index 3b77df709d26..d2d27c2d650d 100644 --- a/src/cron/service/state.ts +++ b/src/cron/service/state.ts @@ -224,9 +224,6 @@ export type CronServiceState = { schedulingPaused: boolean; schedulerStarted: boolean; restartRecoveryPending: boolean; - /** Prevents maintenance reads from advancing deferred startup catch-up slots. - * Entries are removed when the deferred job runs or becomes irrelevant. */ - pendingCatchupDeferralJobIds: Set; activeManualRunJobIds: Set; manualSetupTimeoutNotified: boolean; /** Bounds scheduled, manual, and on-exit work with one shared cron limit. */ @@ -258,7 +255,6 @@ export function createCronServiceState(deps: CronServiceDeps): CronServiceState schedulingPaused: false, schedulerStarted: false, restartRecoveryPending: false, - pendingCatchupDeferralJobIds: new Set(), activeManualRunJobIds: new Set(), manualSetupTimeoutNotified: false, runAdmission: { active: 0, waiters: [] }, diff --git a/src/cron/service/store.ts b/src/cron/service/store.ts index 4562de7d1c7d..9215a1278084 100644 --- a/src/cron/service/store.ts +++ b/src/cron/service/store.ts @@ -22,7 +22,6 @@ type PersistOptions = { export type CronRollbackSnapshot = { store: CronStoreFile | null; durableNextRunAtMsByJobId: Map; - pendingCatchupDeferralJobIds: Set; }; function durableNextRunsFromJobs(jobs: readonly CronJob[]) { @@ -305,7 +304,6 @@ export function snapshotStoreForRollback(state: CronServiceState): CronRollbackS return { store: state.store ? structuredClone(state.store) : null, durableNextRunAtMsByJobId: new Map(state.durableNextRunAtMsByJobId), - pendingCatchupDeferralJobIds: new Set(state.pendingCatchupDeferralJobIds), }; } @@ -332,7 +330,6 @@ export async function persistOrRestore( } catch (err) { state.store = snapshot.store; state.durableNextRunAtMsByJobId = snapshot.durableNextRunAtMsByJobId; - state.pendingCatchupDeferralJobIds = snapshot.pendingCatchupDeferralJobIds; throw err; } for (const notify of opts.postPersistAutoDisableNotifications ?? []) { diff --git a/src/cron/service/timer.ts b/src/cron/service/timer.ts index 8682fcafaa6a..a50b81c64892 100644 --- a/src/cron/service/timer.ts +++ b/src/cron/service/timer.ts @@ -1178,13 +1178,13 @@ function applyOutcomeToStoredJob( endedAt: result.endedAt, triggerEval: result.triggerEval, }); - state.pendingCatchupDeferralJobIds.delete(job.id); + job.state.startupCatchupAtMs = undefined; return undefined; } const shouldDelete = applyJobResult(state, job, result); applyTriggerRunResult(job, result); - state.pendingCatchupDeferralJobIds.delete(job.id); + job.state.startupCatchupAtMs = undefined; emitJobFinished(state, job, result, result.startedAt); @@ -2369,13 +2369,15 @@ async function applyStartupCatchupOutcomes( continue; } if (typeof deferred.delayMs === "number") { - job.state.nextRunAtMs = baseNow + deferred.delayMs + offset - staggerMs; - state.pendingCatchupDeferralJobIds.add(jobId); + const runAtMs = baseNow + deferred.delayMs + offset - staggerMs; + job.state.nextRunAtMs = runAtMs; + job.state.startupCatchupAtMs = runAtMs; offset += staggerMs; continue; } - job.state.nextRunAtMs = baseNow + offset; - state.pendingCatchupDeferralJobIds.add(jobId); + const runAtMs = baseNow + offset; + job.state.nextRunAtMs = runAtMs; + job.state.startupCatchupAtMs = runAtMs; offset += staggerMs; } } diff --git a/src/cron/store.test.ts b/src/cron/store.test.ts index 844c25514a01..3c309c7894ea 100644 --- a/src/cron/store.test.ts +++ b/src/cron/store.test.ts @@ -501,7 +501,11 @@ describe("cron store", () => { const store = await makeStorePath(); const payload = makeStore("job-queued-phase", true); const job = expectDefined(payload.jobs[0], "payload.jobs[0] test invariant"); - job.state = { nextRunAtMs: job.createdAtMs, queuedAtMs: job.createdAtMs + 1 }; + job.state = { + nextRunAtMs: job.createdAtMs, + startupCatchupAtMs: job.createdAtMs, + queuedAtMs: job.createdAtMs + 1, + }; await saveCronStore(store.storePath, payload); @@ -509,9 +513,13 @@ describe("cron store", () => { .db.prepare("SELECT running_at_ms, state_json FROM cron_jobs WHERE job_id = ?") .get(job.id) as { running_at_ms: number | null; state_json: string }; expect(queuedRow.running_at_ms).toBeNull(); - expect(JSON.parse(queuedRow.state_json)).toMatchObject({ queuedAtMs: job.createdAtMs + 1 }); + expect(JSON.parse(queuedRow.state_json)).toMatchObject({ + queuedAtMs: job.createdAtMs + 1, + startupCatchupAtMs: job.createdAtMs, + }); expect((await loadCronStore(store.storePath)).jobs[0]?.state).toMatchObject({ queuedAtMs: job.createdAtMs + 1, + startupCatchupAtMs: job.createdAtMs, }); job.state.queuedAtMs = undefined; diff --git a/src/cron/types.ts b/src/cron/types.ts index 265eddcccde7..e4f718c736cf 100644 --- a/src/cron/types.ts +++ b/src/cron/types.ts @@ -313,6 +313,8 @@ type CronCommandPayloadPatch = { /** Mutable runtime state persisted beside the immutable cron job spec. */ export type CronJobState = { nextRunAtMs?: number; + /** Exact startup catch-up slot protected from future-slot repair across restarts. */ + startupCatchupAtMs?: number; /** Durable pre-admission reservation. Cleared on restart without recording a run. */ queuedAtMs?: number; runningAtMs?: number; diff --git a/src/gateway/server.cron.test.ts b/src/gateway/server.cron.test.ts index 82789471a969..f0aa065a721d 100644 --- a/src/gateway/server.cron.test.ts +++ b/src/gateway/server.cron.test.ts @@ -481,6 +481,7 @@ describe("gateway server cron", () => { expect(internalJob).toBeDefined(); if (internalJob) { internalJob.state.queuedAtMs = Date.now(); + internalJob.state.startupCatchupAtMs = Date.now() + 5_000; } const updateRes = await directCronReq(cronState, "cron.update", { id: String(dailyJobId), @@ -490,12 +491,18 @@ describe("gateway server cron", () => { expect( (updateRes.payload as { state?: Record } | null)?.state, ).not.toHaveProperty("queuedAtMs"); + expect( + (updateRes.payload as { state?: Record } | null)?.state, + ).not.toHaveProperty("startupCatchupAtMs"); const updateEvent = await cronEvents.wait( (payload) => payload.jobId === dailyJobId && payload.action === "updated", ); expect( (updateEvent.job as { state?: Record } | undefined)?.state, ).not.toHaveProperty("queuedAtMs"); + expect( + (updateEvent.job as { state?: Record } | undefined)?.state, + ).not.toHaveProperty("startupCatchupAtMs"); const listRes = await directCronReq(cronState, "cron.list", { includeDisabled: true, @@ -507,6 +514,9 @@ describe("gateway server cron", () => { expect((jobs as Array<{ state?: Record }>)[0]?.state).not.toHaveProperty( "queuedAtMs", ); + expect((jobs as Array<{ state?: Record }>)[0]?.state).not.toHaveProperty( + "startupCatchupAtMs", + ); expect(((jobs as Array<{ name?: unknown }>)[0]?.name as string) ?? "").toBe("daily"); expect( ((jobs as Array<{ delivery?: { mode?: unknown } }>)[0]?.delivery?.mode as string) ?? "", @@ -561,6 +571,9 @@ describe("gateway server cron", () => { expect( (getRes.payload as { state?: Record } | null)?.state, ).not.toHaveProperty("queuedAtMs"); + expect( + (getRes.payload as { state?: Record } | null)?.state, + ).not.toHaveProperty("startupCatchupAtMs"); const missingGetRes = await directCronReq(cronState, "cron.get", { id: "missing-job-id" }); expect(missingGetRes.ok).toBe(false);