diff --git a/docs/plugins/hooks.md b/docs/plugins/hooks.md index b4247303d671..8065a744170f 100644 --- a/docs/plugins/hooks.md +++ b/docs/plugins/hooks.md @@ -587,8 +587,8 @@ and `scheduled` reasons. The event carries a `PluginHookGatewayCronJob` snapshot (including `state.nextRunAtMs`, `state.lastRunStatus`, and `state.lastError` when present) plus a `PluginHookGatewayCronDeliveryStatus` of `not-requested` | `delivered` | `not-delivered` | `unknown`. Removed events -still carry the deleted job snapshot so external schedulers can reconcile -state. +are post-commit: they fire only after durable deletion succeeds and still carry +the deleted job snapshot so external schedulers can reconcile state. A `scheduled` event is post-commit: it fires only after a successful durable write changes an existing job's effective `nextRunAtMs`, excluding that job's diff --git a/docs/plugins/sdk-overview.md b/docs/plugins/sdk-overview.md index 6cb983641a44..a53a62dd3a69 100644 --- a/docs/plugins/sdk-overview.md +++ b/docs/plugins/sdk-overview.md @@ -493,7 +493,7 @@ cover CLI and Gateway-backed install or update paths. - `message_received`: use the typed `threadId` field when you need inbound thread/topic routing. Keep `metadata` for channel-specific extras. - `message_sending`: use typed `replyToId` / `threadId` routing fields before falling back to channel-specific `metadata`. - `gateway_start`: use `ctx.config`, `ctx.workspaceDir`, and `ctx.getCron?.()` for gateway-owned startup state instead of relying on internal `gateway:startup` hooks. -- `cron_changed`: observe gateway-owned cron lifecycle changes. A `scheduled` event is a post-commit reconciliation hint for an existing job's durable next wake, not an ordered delta log. Its `event.nextRunAtMs` is absent when the job has no next wake. +- `cron_changed`: observe gateway-owned cron lifecycle changes. `scheduled` and `removed` events are post-commit reconciliation hints, not an ordered delta log. A scheduled event's `event.nextRunAtMs` is absent when the job has no next wake; a removed event still carries the deleted job snapshot. External wake schedulers should debounce or coalesce `cron_changed` events by `jobId`, then reconcile the full durable view: diff --git a/src/cron/service.removal-postcommit.test.ts b/src/cron/service.removal-postcommit.test.ts new file mode 100644 index 000000000000..4800c787072b --- /dev/null +++ b/src/cron/service.removal-postcommit.test.ts @@ -0,0 +1,235 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import { resetGatewayWorkAdmission } from "../process/gateway-work-admission.js"; +import { setupCronServiceSuite } from "./service.test-harness.js"; +import { list, run } from "./service/ops.js"; +import { createCronServiceState, type CronEvent, type CronServiceState } from "./service/state.js"; +import { ensureLoaded } from "./service/store.js"; +import { onTimer, runMissedJobs } from "./service/timer.js"; +import * as cronStoreModule from "./store.js"; +import { loadCronStore, saveCronStore } from "./store.js"; +import type { CronJob } from "./types.js"; + +const { logger, makeStorePath } = setupCronServiceSuite({ + prefix: "cron-removal-postcommit-", + baseTimeIso: "2026-07-10T12:00:00.000Z", +}); + +type RemovalPath = "manual" | "timer" | "startup catch-up"; + +const removalPaths: RemovalPath[] = ["manual", "timer", "startup catch-up"]; + +function createDueOneShot(id: string, nowMs: number): CronJob { + const runAtMs = nowMs - 60_000; + return { + id, + name: `delete ${id}`, + enabled: true, + deleteAfterRun: true, + createdAtMs: runAtMs - 60_000, + updatedAtMs: runAtMs - 60_000, + schedule: { kind: "at", at: new Date(runAtMs).toISOString() }, + sessionTarget: "isolated", + wakeMode: "next-heartbeat", + payload: { kind: "agentTurn", message: "do work" }, + state: { nextRunAtMs: runAtMs }, + }; +} + +function createState(params: { + storePath: string; + nowMs: number; + onEvent: (event: CronEvent) => void; +}): CronServiceState { + return createCronServiceState({ + storePath: params.storePath, + cronEnabled: true, + log: logger, + nowMs: () => params.nowMs, + enqueueSystemEvent: vi.fn(), + requestHeartbeat: vi.fn(), + runIsolatedAgentJob: vi.fn(async () => ({ status: "ok" as const, summary: "done" })), + onEvent: params.onEvent, + }); +} + +async function executeRemovalPath( + path: RemovalPath, + state: CronServiceState, + jobId: string, +): Promise { + if (path === "manual") { + await run(state, jobId, "force"); + return; + } + if (path === "timer") { + await onTimer(state); + return; + } + await runMissedJobs(state); +} + +function clearStateTimer(state: CronServiceState): void { + if (state.timer) { + clearTimeout(state.timer); + state.timer = null; + } +} + +beforeEach(resetGatewayWorkAdmission); + +afterEach(() => { + resetGatewayWorkAdmission(); + vi.restoreAllMocks(); +}); + +describe.each(removalPaths)("cron one-shot removal via %s", (path) => { + it("emits the full removal snapshot only after the deletion is durable", async () => { + const { storePath } = await makeStorePath(); + const nowMs = Date.parse("2026-07-10T12:00:00.000Z"); + const job = createDueOneShot(`postcommit-${path.replaceAll(" ", "-")}`, nowMs); + await saveCronStore(storePath, { version: 1, jobs: [job] }); + + const events: CronEvent[] = []; + const durableJobsAtRemoval: Array> = []; + const state = createState({ + storePath, + nowMs, + onEvent: (event) => { + events.push(structuredClone(event)); + if (event.action === "removed") { + durableJobsAtRemoval.push(loadCronStore(storePath).then((store) => store.jobs)); + } + }, + }); + + try { + await executeRemovalPath(path, state, job.id); + + const relevantEvents = events.filter((event) => event.jobId === job.id); + expect(relevantEvents.map((event) => event.action)).toEqual([ + "started", + "finished", + "removed", + ]); + const removed = relevantEvents.at(-1); + expect(removed).toMatchObject({ + action: "removed", + jobId: job.id, + job: { + id: job.id, + name: job.name, + deleteAfterRun: true, + state: { + lastRunStatus: "ok", + lastStatus: "ok", + }, + }, + }); + expect(durableJobsAtRemoval).toHaveLength(1); + await expect(Promise.all(durableJobsAtRemoval)).resolves.toEqual([[]]); + expect(state.store?.jobs).toEqual([]); + } finally { + clearStateTimer(state); + } + }); + + it("restores live and durable wake state when the final deletion write fails", async () => { + const { storePath } = await makeStorePath(); + const nowMs = Date.parse("2026-07-10T12:00:00.000Z"); + const job = createDueOneShot(`rollback-${path.replaceAll(" ", "-")}`, nowMs); + await saveCronStore(storePath, { version: 1, jobs: [job] }); + + const events: CronEvent[] = []; + let listedAfterFinished: Promise | undefined; + const state = createState({ + storePath, + nowMs, + onEvent: (event) => { + events.push(structuredClone(event)); + if (event.action === "finished") { + // Models a detached hook: its read waits for the finalization lock, + // then must see the rolled-back durable topology after write failure. + listedAfterFinished = list(state, { includeDisabled: true }); + } + }, + }); + + const realSave = cronStoreModule.saveCronJobsStore; + let saveCount = 0; + vi.spyOn(cronStoreModule, "saveCronJobsStore").mockImplementation(async (...args) => { + saveCount += 1; + if (saveCount === 2) { + throw new Error("final persist failed"); + } + await realSave(...args); + }); + + try { + await expect(executeRemovalPath(path, state, job.id)).rejects.toThrow("final persist failed"); + + expect(saveCount).toBe(2); + expect(events.some((event) => event.action === "removed")).toBe(false); + if (!listedAfterFinished) { + throw new Error("missing detached finished-hook read"); + } + await expect(listedAfterFinished).resolves.toEqual([expect.objectContaining({ id: job.id })]); + + const durableStore = await loadCronStore(storePath); + expect(state.store).toEqual(durableStore); + const durableJob = durableStore.jobs[0]; + expect(durableJob?.id).toBe(job.id); + expect(state.durableNextRunAtMsByJobId).toEqual( + new Map([[job.id, durableJob?.state.nextRunAtMs]]), + ); + } finally { + clearStateTimer(state); + } + }); + + it("suppresses removal when quarantine prevents the durable write", async () => { + const { storePath } = await makeStorePath(); + const nowMs = Date.parse("2026-07-10T12:00:00.000Z"); + const job = createDueOneShot(`quarantine-${path.replaceAll(" ", "-")}`, nowMs); + await saveCronStore(storePath, { version: 1, jobs: [job] }); + const durableBefore = await loadCronStore(storePath); + + const events: CronEvent[] = []; + const state = createState({ + storePath, + nowMs, + onEvent: (event) => events.push(structuredClone(event)), + }); + await ensureLoaded(state, { skipRecompute: true }); + state.pendingQuarantineConfigJobs = [ + { sourceIndex: 0, reason: "invalid-schedule", job: { id: "quarantined-job" } }, + ]; + vi.spyOn(cronStoreModule, "saveCronQuarantineFile").mockRejectedValue( + new Error("quarantine unavailable"), + ); + const saveStore = vi.spyOn(cronStoreModule, "saveCronJobsStore"); + + try { + await expect(executeRemovalPath(path, state, job.id)).rejects.toThrow( + "cron: durable store write did not complete", + ); + + expect(saveStore).not.toHaveBeenCalled(); + expect(events.some((event) => event.action === "removed")).toBe(false); + const durableStore = await loadCronStore(storePath); + expect(durableStore).toEqual(durableBefore); + expect(state.store?.jobs).toEqual([ + expect.objectContaining({ + id: job.id, + state: expect.objectContaining({ + nextRunAtMs: durableBefore.jobs[0]?.state.nextRunAtMs, + }), + }), + ]); + expect(state.durableNextRunAtMsByJobId).toEqual( + new Map([[job.id, durableBefore.jobs[0]?.state.nextRunAtMs]]), + ); + } finally { + clearStateTimer(state); + } + }); +}); diff --git a/src/cron/service/ops.ts b/src/cron/service/ops.ts index 211bd5291dc2..15cf0efa3cae 100644 --- a/src/cron/service/ops.ts +++ b/src/cron/service/ops.ts @@ -24,7 +24,7 @@ import { createCronRunDiagnosticsFromError } from "../run-diagnostics.js"; import { createCronExecutionId } from "../run-id.js"; import { normalizeCronRunLogJobId } from "../run-log.js"; import { cronSchedulingInputsEqual } from "../schedule-identity.js"; -import type { CronJob, CronJobCreate, CronJobPatch, CronPayload, CronStoreFile } from "../types.js"; +import type { CronJob, CronJobCreate, CronJobPatch, CronPayload } from "../types.js"; import { normalizeCronRunErrorText } from "./execution-errors.js"; import { failureNotificationDeliveryFromJobState } from "./failure-alerts.js"; import { @@ -59,7 +59,14 @@ import type { CronWakeMode, } from "./state.js"; import { emit } from "./state.js"; -import { ensureLoaded, persist, warnIfDisabled } from "./store.js"; +import { + ensureLoaded, + persist, + persistOrRestore, + snapshotStoreForRollback, + type CronRollbackSnapshot, + warnIfDisabled, +} from "./store.js"; import { CRON_TASK_RUNNING_PROGRESS_SUMMARY } from "./task-ledger.js"; import { applyJobResult, @@ -513,49 +520,6 @@ 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, - opts: { - postPersistAutoDisableNotifications?: Array<() => void>; - suppressScheduledJobId?: string; - } = {}, -) { - try { - await persist( - state, - opts.suppressScheduledJobId === undefined - ? undefined - : { suppressScheduledJobId: opts.suppressScheduledJobId }, - ); - } catch (err) { - state.store = snapshot.store; - state.pendingCatchupDeferralJobIds = snapshot.pendingCatchupDeferralJobIds; - throw err; - } - for (const notify of opts.postPersistAutoDisableNotifications ?? []) { - notify(); - } -} - -function snapshotStoreForRollback(state: CronServiceState): CronRollbackSnapshot { - return { - store: state.store ? structuredClone(state.store) : null, - pendingCatchupDeferralJobIds: new Set(state.pendingCatchupDeferralJobIds), - }; -} - function finalizeUpdatedJob(params: { job: CronJob; nextJob: CronJob; @@ -897,7 +861,7 @@ async function skipInvalidPersistedManualRun(params: { severity: "warn", nowMs: params.state.deps.nowMs, }); - const shouldDelete = applyJobResult( + applyJobResult( params.state, params.job, { @@ -929,11 +893,6 @@ async function skipInvalidPersistedManualRun(params: { params.terminalTracker, ); - if (shouldDelete && params.state.store) { - params.state.store.jobs = params.state.store.jobs.filter((entry) => entry.id !== params.job.id); - emit(params.state, { jobId: params.job.id, action: "removed" }); - } - recomputeNextRunsForMaintenance(params.state, { recomputeExpired: true }); await persist(params.state); armTimer(params.state); @@ -1300,11 +1259,6 @@ async function finishPreparedManualRun( ); } - if (shouldDelete && state.store) { - state.store.jobs = state.store.jobs.filter((entry) => entry.id !== job.id); - emit(state, { jobId: job.id, action: "removed", job }); - } - // Manual runs should not advance other due jobs without executing them. // Use maintenance-only recompute to repair missing values while // preserving existing past-due nextRunAtMs entries for future timer ticks. @@ -1316,6 +1270,7 @@ async function finishPreparedManualRun( state: structuredClone(job.state), }; const postRunRemoved = shouldDelete; + const removedJob = shouldDelete ? structuredClone(job) : undefined; // Isolated Telegram send can persist target writeback directly to disk. // Reload before final persist so manual `cron run` keeps those changes. await ensureLoaded(state, { forceReload: true, skipRecompute: true }); @@ -1323,6 +1278,7 @@ async function finishPreparedManualRun( notifySetupTimeout = false; return; } + const rollbackSnapshot = snapshotStoreForRollback(state); mergeManualRunSnapshotAfterReload({ state, jobId, @@ -1330,7 +1286,10 @@ async function finishPreparedManualRun( removed: postRunRemoved, }); recomputeNextRunsForMaintenance(state, { recomputeExpired: true }); - await persist(state); + await persistOrRestore(state, rollbackSnapshot); + if (removedJob) { + emit(state, { jobId: removedJob.id, action: "removed", job: removedJob }); + } finalized = true; }); if (notifySetupTimeout && isCronActiveJobMarkerCurrent(prepared.activeJobMarker)) { diff --git a/src/cron/service/store.ts b/src/cron/service/store.ts index c5141a9f2240..4562de7d1c7d 100644 --- a/src/cron/service/store.ts +++ b/src/cron/service/store.ts @@ -10,7 +10,7 @@ import { saveCronJobsStore, type QuarantinedCronConfigJob, } from "../store.js"; -import type { CronJob } from "../types.js"; +import type { CronJob, CronStoreFile } from "../types.js"; import { recomputeNextRuns } from "./jobs.js"; import { emit, type CronServiceState } from "./state.js"; @@ -19,6 +19,12 @@ type PersistOptions = { suppressScheduledJobId?: string; }; +export type CronRollbackSnapshot = { + store: CronStoreFile | null; + durableNextRunAtMsByJobId: Map; + pendingCatchupDeferralJobIds: Set; +}; + function durableNextRunsFromJobs(jobs: readonly CronJob[]) { return new Map(jobs.map((job) => [job.id, job.state.nextRunAtMs] as const)); } @@ -273,13 +279,13 @@ export function warnIfDisabled(state: CronServiceState, action: string) { export async function persist(state: CronServiceState, opts?: PersistOptions) { const store = state.store; if (!store) { - return; + return false; } let flushedPendingQuarantine = false; if (state.pendingQuarantineConfigJobs.length > 0) { const quarantinePath = await flushPendingQuarantine(state, state.deps.nowMs()); if (!quarantinePath) { - return; + return false; } flushedPendingQuarantine = true; } @@ -291,4 +297,45 @@ export async function persist(state: CronServiceState, opts?: PersistOptions) { stateOnly, suppressScheduledJobId: opts?.suppressScheduledJobId, }); + return true; +} + +/** Captures the live cron state that must stay aligned with the durable store. */ +export function snapshotStoreForRollback(state: CronServiceState): CronRollbackSnapshot { + return { + store: state.store ? structuredClone(state.store) : null, + durableNextRunAtMsByJobId: new Map(state.durableNextRunAtMsByJobId), + pendingCatchupDeferralJobIds: new Set(state.pendingCatchupDeferralJobIds), + }; +} + +// A failed durable write must not leave readers observing speculative job +// topology, wake times, or catch-up ownership after the store lock releases. +export async function persistOrRestore( + state: CronServiceState, + snapshot: CronRollbackSnapshot, + opts: { + postPersistAutoDisableNotifications?: Array<() => void>; + suppressScheduledJobId?: string; + } = {}, +) { + try { + const persisted = await persist( + state, + opts.suppressScheduledJobId === undefined + ? undefined + : { suppressScheduledJobId: opts.suppressScheduledJobId }, + ); + if (!persisted) { + throw new Error("cron: durable store write did not complete"); + } + } catch (err) { + state.store = snapshot.store; + state.durableNextRunAtMsByJobId = snapshot.durableNextRunAtMsByJobId; + state.pendingCatchupDeferralJobIds = snapshot.pendingCatchupDeferralJobIds; + throw err; + } + for (const notify of opts.postPersistAutoDisableNotifications ?? []) { + notify(); + } } diff --git a/src/cron/service/timer.ts b/src/cron/service/timer.ts index d4c9ef52bb33..d48c4dfd0219 100644 --- a/src/cron/service/timer.ts +++ b/src/cron/service/timer.ts @@ -84,7 +84,7 @@ import { } from "./jobs.js"; import { locked } from "./locked.js"; import { emit, type CronServiceState, type CronSystemEventEnqueueResult } from "./state.js"; -import { ensureLoaded, persist } from "./store.js"; +import { ensureLoaded, persist, persistOrRestore, snapshotStoreForRollback } from "./store.js"; import { resolveMainSessionCronRunSessionKey, tryCreateCronTaskRun, @@ -1097,17 +1097,20 @@ export function applyTriggerNoFireResult( } } -function applyOutcomeToStoredJob(state: CronServiceState, result: TimedCronRunOutcome): void { +function applyOutcomeToStoredJob( + state: CronServiceState, + result: TimedCronRunOutcome, +): CronJob | undefined { tryFinishCronTaskRun(state, result); const store = state.store; if (!store) { - return; + return undefined; } const jobs = store.jobs; const job = jobs.find((entry) => entry.id === result.jobId); if (!job) { if (result.status === "ok" && result.triggerEval?.fired === false) { - return; + return undefined; } if (result.status === "ok") { // A manual/queued run may finish after the job was removed. Preserve the @@ -1126,13 +1129,13 @@ function applyOutcomeToStoredJob(state: CronServiceState, result: TimedCronRunOu { jobId: result.jobId }, "cron: finalized successful run after job was removed during execution", ); - return; + return undefined; } state.deps.log.warn( { jobId: result.jobId }, "cron: applyOutcomeToStoredJob — job not found after forceReload, result discarded", ); - return; + return undefined; } if (result.status === "ok" && result.triggerEval && !result.triggerEval.fired) { @@ -1144,7 +1147,7 @@ function applyOutcomeToStoredJob(state: CronServiceState, result: TimedCronRunOu triggerEval: result.triggerEval, }); state.pendingCatchupDeferralJobIds.delete(job.id); - return; + return undefined; } const shouldDelete = applyJobResult(state, job, { @@ -1163,8 +1166,9 @@ function applyOutcomeToStoredJob(state: CronServiceState, result: TimedCronRunOu if (shouldDelete) { store.jobs = jobs.filter((entry) => entry.id !== job.id); - emit(state, { jobId: job.id, action: "removed", job }); + return job; } + return undefined; } function clearActiveMarkersForOutcomes(outcomes: readonly TimedCronRunOutcome[]): void { @@ -1459,8 +1463,13 @@ async function onAdmittedTimer(state: CronServiceState) { await ensureLoaded(state, { forceReload: true, skipRecompute: true }); finalizedResults = filterCurrentCronRunOutcomes(currentResults); finishRetiredCronTaskRuns(state, completedResults, finalizedResults); + const rollbackSnapshot = snapshotStoreForRollback(state); + const removedJobs: CronJob[] = []; for (const result of finalizedResults) { - applyOutcomeToStoredJob(state, result); + const removedJob = applyOutcomeToStoredJob(state, result); + if (removedJob) { + removedJobs.push(removedJob); + } } if (finalizedResults.length === 0) { return; @@ -1472,7 +1481,10 @@ async function onAdmittedTimer(state: CronServiceState) { // those jobs (advancing nextRunAtMs without execution), causing // daily cron schedules to jump 48 h instead of 24 h (#17852). recomputeNextRunsForMaintenance(state); - await persist(state); + await persistOrRestore(state, rollbackSnapshot); + for (const removedJob of removedJobs) { + emit(state, { jobId: removedJob.id, action: "removed", job: removedJob }); + } }); finalizationSucceeded = finalizedResults.length > 0; return finalizedResults; @@ -1979,6 +1991,7 @@ async function applyStartupCatchupOutcomes( return; } if (state.stopped) { + const rollbackSnapshot = snapshotStoreForRollback(state); finishRetiredCronTaskRuns(state, outcomes, []); const releasedReservations = releaseUnstartedStartupCatchupReservations( state, @@ -1987,25 +2000,30 @@ async function applyStartupCatchupOutcomes( ); if (releasedReservations) { recomputeNextRunsForMaintenance(state, { repairFutureCronNextRunAtMs: false }); - await persist(state); + await persistOrRestore(state, rollbackSnapshot); } return; } finalizedOutcomes = filterCurrentCronRunOutcomes(currentOutcomes); finishRetiredCronTaskRuns(state, outcomes, finalizedOutcomes); + const rollbackSnapshot = snapshotStoreForRollback(state); const releasedReservations = releaseUnstartedStartupCatchupReservations( state, plan, outcomes, ); + const removedJobs: CronJob[] = []; for (const result of finalizedOutcomes) { - applyOutcomeToStoredJob(state, result); + const removedJob = applyOutcomeToStoredJob(state, result); + if (removedJob) { + removedJobs.push(removedJob); + } } if (finalizedOutcomes.length === 0 && plan.deferredJobs.length === 0) { if (releasedReservations) { recomputeNextRunsForMaintenance(state, { repairFutureCronNextRunAtMs: false }); - await persist(state); + await persistOrRestore(state, rollbackSnapshot); } return; } @@ -2036,7 +2054,10 @@ async function applyStartupCatchupOutcomes( // instead of being silently advanced. Future repair is disabled here so // startup overflow deferrals survive until their staggered catch-up tick. recomputeNextRunsForMaintenance(state, { repairFutureCronNextRunAtMs: false }); - await persist(state); + await persistOrRestore(state, rollbackSnapshot); + for (const removedJob of removedJobs) { + emit(state, { jobId: removedJob.id, action: "removed", job: removedJob }); + } }); return finalizedOutcomes; } finally {