From fe3215cbb0f412aff961c8e3370e6cfcca49a7ff Mon Sep 17 00:00:00 2001 From: Masato Hoshino Date: Sun, 5 Jul 2026 04:34:02 +0900 Subject: [PATCH] fix(cron): roll back live scheduler state when persisting add/update/remove fails (#99960) * fix(cron): roll back live store when persist fails during add/update/remove * fix(cron): restore catch-up deferral markers on failed persists * fix(cron): defer rollback-sensitive notifications --------- Co-authored-by: Vincent Koc --- src/cron/service/jobs.ts | 64 ++++++++++---- src/cron/service/ops.test.ts | 167 ++++++++++++++++++++++++++++++++++- src/cron/service/ops.ts | 58 ++++++++++-- 3 files changed, 263 insertions(+), 26 deletions(-) diff --git a/src/cron/service/jobs.ts b/src/cron/service/jobs.ts index 9a9be4929f9..3c59bf5e91c 100644 --- a/src/cron/service/jobs.ts +++ b/src/cron/service/jobs.ts @@ -492,6 +492,7 @@ export function recordScheduleComputeError(params: { state: CronServiceState; job: CronJob; err: unknown; + deferredAutoDisableNotifications?: Array<() => void>; }): boolean { const { state, job, err } = params; const errorCount = (job.state.scheduleErrorCount ?? 0) + 1; @@ -508,20 +509,27 @@ export function recordScheduleComputeError(params: { "cron: auto-disabled job after repeated schedule errors", ); - // Notify the user so the auto-disable is not silent (#28861). const notifyText = `⚠️ Cron job "${job.name}" has been auto-disabled after ${errorCount} consecutive schedule errors. Last error: ${errText}`; - state.deps.enqueueSystemEvent(notifyText, { - agentId: job.agentId, - sessionKey: job.sessionKey, - contextKey: `cron:${job.id}:auto-disabled`, - }); - state.deps.requestHeartbeat({ - source: "cron", - intent: "event", - reason: `cron:${job.id}:auto-disabled`, - agentId: job.agentId, - sessionKey: job.sessionKey, - }); + const notify = () => { + state.deps.enqueueSystemEvent(notifyText, { + agentId: job.agentId, + sessionKey: job.sessionKey, + contextKey: `cron:${job.id}:auto-disabled`, + }); + state.deps.requestHeartbeat({ + source: "cron", + intent: "event", + reason: `cron:${job.id}:auto-disabled`, + agentId: job.agentId, + sessionKey: job.sessionKey, + }); + }; + if (params.deferredAutoDisableNotifications) { + params.deferredAutoDisableNotifications.push(notify); + } else { + // Notify the user so the auto-disable is not silent (#28861). + notify(); + } } else { state.deps.log.warn( { jobId: job.id, name: job.name, errorCount, err: errText }, @@ -617,7 +625,12 @@ function walkSchedulableJobs( return changed; } -function recomputeJobNextRunAtMs(params: { state: CronServiceState; job: CronJob; nowMs: number }) { +function recomputeJobNextRunAtMs(params: { + state: CronServiceState; + job: CronJob; + nowMs: number; + deferredAutoDisableNotifications?: Array<() => void>; +}) { let changed = false; try { let newNext = computeJobNextRunAtMs(params.job, params.nowMs); @@ -644,7 +657,14 @@ function recomputeJobNextRunAtMs(params: { state: CronServiceState; job: CronJob changed = true; } } catch (err) { - if (recordScheduleComputeError({ state: params.state, job: params.job, err })) { + if ( + recordScheduleComputeError({ + state: params.state, + job: params.job, + err, + deferredAutoDisableNotifications: params.deferredAutoDisableNotifications, + }) + ) { changed = true; } } @@ -682,10 +702,18 @@ export function recomputeNextRunsForMaintenance( recomputeExpired?: boolean; nowMs?: number; repairFutureCronNextRunAtMs?: boolean; + deferredAutoDisableNotifications?: Array<() => void>; }, ): boolean { const recomputeExpired = opts?.recomputeExpired ?? false; const repairFutureCronNextRunAtMs = opts?.repairFutureCronNextRunAtMs ?? true; + const recomputeJob = (job: CronJob, nowMs: number) => + recomputeJobNextRunAtMs({ + state, + job, + 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. @@ -715,7 +743,7 @@ export function recomputeNextRunsForMaintenance( } if (!hasScheduledNextRunAtMs(job.state.nextRunAtMs)) { - if (recomputeJobNextRunAtMs({ state, job, nowMs: now })) { + if (recomputeJob(job, now)) { changed = true; } } else if ( @@ -723,7 +751,7 @@ export function recomputeNextRunsForMaintenance( !deferralIds.has(job.id) && shouldRepairFutureCronNextRunAtMs({ state, job, nowMs: now }) ) { - if (recomputeJobNextRunAtMs({ state, job, nowMs: now })) { + if (recomputeJob(job, now)) { changed = true; } } else if ( @@ -745,7 +773,7 @@ export function recomputeNextRunsForMaintenance( now < backoffUntilMs && job.state.nextRunAtMs < backoffUntilMs; if (alreadyExecutedSlot || isStaleBackoffSlot) { - if (recomputeJobNextRunAtMs({ state, job, nowMs: now })) { + if (recomputeJob(job, now)) { changed = true; } } diff --git a/src/cron/service/ops.test.ts b/src/cron/service/ops.test.ts index b6bf2c13b42..ecf38e621b9 100644 --- a/src/cron/service/ops.test.ts +++ b/src/cron/service/ops.test.ts @@ -1,16 +1,18 @@ // Cron service ops tests cover high-level service operations and state transitions. import fs from "node:fs/promises"; import path from "node:path"; -import { describe, expect, it, vi } from "vitest"; +import { afterEach, describe, expect, it, vi } from "vitest"; import { runOpenClawStateWriteTransaction } from "../../state/openclaw-state-db.js"; import * as detachedTaskRuntime from "../../tasks/detached-task-runtime.js"; import { findTaskByRunId, resetTaskRegistryForTests } from "../../tasks/task-registry.js"; import { formatTaskStatusDetail } from "../../tasks/task-status.js"; import { withEnvAsync } from "../../test-utils/env.js"; +import * as cronSchedule from "../schedule.js"; import { setupCronServiceSuite, writeCronStoreSnapshot } from "../service.test-harness.js"; +import * as cronStoreModule from "../store.js"; import { loadCronJobsStoreWithConfigJobs, loadCronStore } from "../store.js"; import type { CronJob } from "../types.js"; -import { add, run, start, stop, update } from "./ops.js"; +import { add, list, remove, run, start, stop, update } from "./ops.js"; import { createCronServiceState } from "./state.js"; import { runMissedJobs } from "./timer.js"; @@ -827,3 +829,164 @@ describe("cron service ops seam coverage", () => { expect(job.state.nextRunAtMs).toBeGreaterThan(finalBaseRunAtMs); }); }); + +describe("cron service ops persist rollback", () => { + afterEach(() => { + vi.restoreAllMocks(); + }); + + function makeCreateInput(name: string) { + return { + name, + enabled: true, + schedule: { kind: "cron", expr: "0 0 * * *" }, + sessionTarget: "isolated", + wakeMode: "next-heartbeat", + payload: { kind: "agentTurn", message: "do work" }, + } as const; + } + + it("rolls back an added job from the live store when persist fails", async () => { + const { storePath } = await makeStorePath(); + const now = Date.parse("2026-06-09T00:00:00.000Z"); + const state = createOkIsolatedCronState({ storePath, now }); + + vi.spyOn(cronStoreModule, "saveCronJobsStore").mockRejectedValueOnce(new Error("disk full")); + + await expect(add(state, makeCreateInput("daily cleanup"))).rejects.toThrow("disk full"); + + expect(state.timer).toBeNull(); + expect(state.store?.jobs ?? []).toEqual([]); + const listed = await list(state, { includeDisabled: true }); + if (state.timer) { + clearTimeout(state.timer); + } + expect(listed).toEqual([]); + const loaded = await loadCronStore(storePath); + expect(loaded.jobs).toEqual([]); + }); + + it("keeps the pre-update job in the live store when persist fails", async () => { + const { storePath } = await makeStorePath(); + const now = Date.parse("2026-06-09T00:00:00.000Z"); + const state = createOkIsolatedCronState({ storePath, now }); + + const job = await add(state, makeCreateInput("daily cleanup")); + if (state.timer) { + clearTimeout(state.timer); + } + + vi.spyOn(cronStoreModule, "saveCronJobsStore").mockRejectedValueOnce(new Error("disk full")); + + await expect(update(state, job.id, { name: "renamed cleanup" })).rejects.toThrow("disk full"); + + const inMemory = state.store?.jobs.find((entry) => entry.id === job.id); + expect(inMemory?.name).toBe("daily cleanup"); + const loaded = await loadCronStore(storePath); + const stored = loaded.jobs.find((entry) => entry.id === job.id); + expect(stored?.name).toBe("daily cleanup"); + }); + + it("keeps a removed job in the live store when persist fails", async () => { + const { storePath } = await makeStorePath(); + const now = Date.parse("2026-06-09T00:00:00.000Z"); + const state = createOkIsolatedCronState({ storePath, now }); + + const job = await add(state, makeCreateInput("daily cleanup")); + if (state.timer) { + clearTimeout(state.timer); + } + + vi.spyOn(cronStoreModule, "saveCronJobsStore").mockRejectedValueOnce(new Error("disk full")); + + await expect(remove(state, job.id)).rejects.toThrow("disk full"); + + expect(state.store?.jobs.map((entry) => entry.id)).toEqual([job.id]); + const loaded = await loadCronStore(storePath); + 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 () => { + const { storePath } = await makeStorePath(); + const now = Date.parse("2026-06-09T00:00:00.000Z"); + const state = createOkIsolatedCronState({ storePath, now }); + + const job = await add(state, makeCreateInput("daily cleanup")); + if (state.timer) { + clearTimeout(state.timer); + } + state.pendingCatchupDeferralJobIds.add(job.id); + + 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.map((entry) => entry.id)).toEqual([job.id]); + }); + + it("recovers after a failed persist so the next mutation succeeds", async () => { + const { storePath } = await makeStorePath(); + const now = Date.parse("2026-06-09T00:00:00.000Z"); + const state = createOkIsolatedCronState({ storePath, now }); + + vi.spyOn(cronStoreModule, "saveCronJobsStore").mockRejectedValueOnce(new Error("disk full")); + await expect(add(state, makeCreateInput("daily cleanup"))).rejects.toThrow("disk full"); + + const job = await add(state, makeCreateInput("daily cleanup")); + if (state.timer) { + clearTimeout(state.timer); + } + + const listed = await list(state, { includeDisabled: true }); + if (state.timer) { + clearTimeout(state.timer); + } + expect(listed.map((entry) => entry.id)).toEqual([job.id]); + const loaded = await loadCronStore(storePath); + expect(loaded.jobs.map((entry) => entry.id)).toEqual([job.id]); + }); + + it("notifies about schedule auto-disable only after the mutation persists", async () => { + const { storePath } = await makeStorePath(); + const now = Date.parse("2026-06-09T00:00:00.000Z"); + const state = createOkIsolatedCronState({ storePath, now }); + + const malformed = await add(state, { + ...makeCreateInput("malformed sibling"), + schedule: { kind: "cron", expr: "0 1 * * *" }, + }); + if (state.timer) { + clearTimeout(state.timer); + } + malformed.state.nextRunAtMs = undefined; + malformed.state.scheduleErrorCount = 2; + const enqueueSystemEvent = vi.mocked(state.deps.enqueueSystemEvent); + const requestHeartbeat = vi.mocked(state.deps.requestHeartbeat); + enqueueSystemEvent.mockClear(); + requestHeartbeat.mockClear(); + const computeNextRunAtMs = cronSchedule.computeNextRunAtMs; + vi.spyOn(cronSchedule, "computeNextRunAtMs").mockImplementation((schedule, nowMs) => { + if (schedule.kind === "cron" && schedule.expr === "0 1 * * *") { + throw new Error("simulated schedule failure"); + } + return computeNextRunAtMs(schedule, nowMs); + }); + + vi.spyOn(cronStoreModule, "saveCronJobsStore").mockRejectedValueOnce(new Error("disk full")); + await expect(add(state, makeCreateInput("failed mutation"))).rejects.toThrow("disk full"); + + expect(state.store?.jobs.find((job) => job.id === malformed.id)?.enabled).toBe(true); + expect(enqueueSystemEvent).not.toHaveBeenCalled(); + expect(requestHeartbeat).not.toHaveBeenCalled(); + + await add(state, makeCreateInput("successful mutation")); + if (state.timer) { + clearTimeout(state.timer); + } + + expect(state.store?.jobs.find((job) => job.id === malformed.id)?.enabled).toBe(false); + expect(enqueueSystemEvent).toHaveBeenCalledTimes(1); + expect(requestHeartbeat).toHaveBeenCalledTimes(1); + }); +}); diff --git a/src/cron/service/ops.ts b/src/cron/service/ops.ts index 52ffaca8302..c58ea26e08c 100644 --- a/src/cron/service/ops.ts +++ b/src/cron/service/ops.ts @@ -19,7 +19,7 @@ import { resolveCronDeliveryPlan, resolveFailureDestination } from "../delivery- import { createCronRunDiagnosticsFromError } from "../run-diagnostics.js"; import { createCronExecutionId } from "../run-id.js"; import { cronSchedulingInputsEqual } from "../schedule-identity.js"; -import type { CronJob, CronJobCreate, CronJobPatch, CronPayload } from "../types.js"; +import type { CronJob, CronJobCreate, CronJobPatch, CronPayload, CronStoreFile } from "../types.js"; import { normalizeCronRunErrorText } from "./execution-errors.js"; import { failureNotificationDeliveryFromJobState } from "./failure-alerts.js"; import { @@ -473,17 +473,58 @@ export async function listPage(state: CronServiceState, opts?: CronListPageOptio }); } +type CronRollbackSnapshot = { + store: CronStoreFile | null; + pendingCatchupDeferralJobIds: Set; +}; + +// Rolls the live scheduler state back to its pre-mutation snapshot when the +// durable write fails. recomputeNextRunsForMaintenance mutates schedule state +// across all jobs and drops catch-up deferral markers for removed/disabled +// jobs, so restoring only the touched job would leave siblings and deferrals +// ahead of disk; without the rollback a failed add/update/remove keeps +// running, reverting, or resurrecting jobs the caller was told did not apply. +async function persistOrRestore( + state: CronServiceState, + snapshot: CronRollbackSnapshot, + postPersistAutoDisableNotifications: Array<() => void> = [], +) { + try { + await persist(state); + } catch (err) { + state.store = snapshot.store; + state.pendingCatchupDeferralJobIds = snapshot.pendingCatchupDeferralJobIds; + throw err; + } + for (const notify of postPersistAutoDisableNotifications) { + notify(); + } +} + +function snapshotStoreForRollback(state: CronServiceState): CronRollbackSnapshot { + return { + store: state.store ? structuredClone(state.store) : null, + pendingCatchupDeferralJobIds: new Set(state.pendingCatchupDeferralJobIds), + }; +} + /** Adds a cron job, recomputes scheduler state, persists, and re-arms the timer. */ export async function add(state: CronServiceState, input: CronJobCreate) { return await locked(state, async () => { warnIfDisabled(state, "add"); await ensureLoaded(state, { skipRecompute: true }); + const snapshot = snapshotStoreForRollback(state); const job = createJob(state, input); state.store?.jobs.push(job); - recomputeNextRunsForMaintenance(state); + // Auto-disable notifications describe durable state, so publish them only + // after the write succeeds instead of leaking a rolled-back transition. + const postPersistAutoDisableNotifications: Array<() => void> = []; + recomputeNextRunsForMaintenance(state, { + deferredAutoDisableNotifications: postPersistAutoDisableNotifications, + }); - await persist(state); + await persistOrRestore(state, snapshot, postPersistAutoDisableNotifications); armTimer(state); state.deps.log.info( @@ -513,6 +554,7 @@ export async function update(state: CronServiceState, id: string, patch: CronJob return await locked(state, async () => { warnIfDisabled(state, "update"); await ensureLoaded(state, { skipRecompute: true }); + const snapshot = snapshotStoreForRollback(state); const job = findJobOrThrow(state, id); const now = state.deps.nowMs(); const nextJob = structuredClone(job); @@ -581,7 +623,7 @@ export async function update(state: CronServiceState, id: string, patch: CronJob } } - await persist(state); + await persistOrRestore(state, snapshot); armTimer(state); emit(state, { jobId: id, @@ -602,13 +644,17 @@ export async function remove(state: CronServiceState, id: string) { if (!state.store) { return { ok: false, removed: false } as const; } + const snapshot = snapshotStoreForRollback(state); const removedJob = state.store.jobs.find((j) => j.id === id); state.store.jobs = state.store.jobs.filter((j) => j.id !== id); const removed = (state.store.jobs.length ?? 0) !== before; - recomputeNextRunsForMaintenance(state); + const postPersistAutoDisableNotifications: Array<() => void> = []; + recomputeNextRunsForMaintenance(state, { + deferredAutoDisableNotifications: postPersistAutoDisableNotifications, + }); - await persist(state); + await persistOrRestore(state, snapshot, postPersistAutoDisableNotifications); armTimer(state); if (removed) { emit(state, { jobId: id, action: "removed", job: removedJob });