diff --git a/src/cron/scratch-store.ts b/src/cron/scratch-store.ts index 7d6b71819724..da2917f27d2b 100644 --- a/src/cron/scratch-store.ts +++ b/src/cron/scratch-store.ts @@ -208,8 +208,8 @@ export function writeCronJobScratch(params: { /** * Deletes scratch when its owning job is removed, or — with expectedRevision — - * atomically reverts a migration write back to the no-row state. Orphans remain - * harmless on partial failure. Returns false when the guarded revision moved. + * atomically reverts a migration write back to the no-row state. Returns false + * when the guarded revision moved. */ export function deleteCronJobScratch( storePath: string, diff --git a/src/cron/service.removal-postcommit.test.ts b/src/cron/service.removal-postcommit.test.ts index 5acae1e0331a..6d5eca9702bd 100644 --- a/src/cron/service.removal-postcommit.test.ts +++ b/src/cron/service.removal-postcommit.test.ts @@ -1,6 +1,8 @@ import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; import { resetGatewayWorkAdmission } from "../process/gateway-work-admission.js"; +import { readCronJobScratchState, writeCronJobScratch } from "./scratch-store.js"; import { setupCronServiceSuite } from "./service.test-harness.js"; +import { add } from "./service/ops-mutations.js"; import { list } from "./service/ops-read.js"; import { run } from "./service/ops-run.js"; import { createCronServiceState, type CronEvent, type CronServiceState } from "./service/state.js"; @@ -37,6 +39,18 @@ function createDueOneShot(id: string, nowMs: number): CronJob { }; } +function createReplacementInput(id: string) { + return { + id, + name: `replacement ${id}`, + enabled: true, + schedule: { kind: "every" as const, everyMs: 60_000 }, + sessionTarget: "isolated" as const, + wakeMode: "next-heartbeat" as const, + payload: { kind: "agentTurn" as const, message: "replacement work" }, + }; +} + function createState(params: { storePath: string; nowMs: number; @@ -85,21 +99,37 @@ afterEach(() => { }); describe.each(removalPaths)("cron one-shot removal via %s", (path) => { - it("emits the full removal snapshot only after the deletion is durable", async () => { + it("emits the full removal snapshot only after the job and scratch deletion are 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] }); + expect( + writeCronJobScratch({ + storePath, + jobId: job.id, + content: "original scratch", + nowMs: nowMs - 1, + }), + ).toMatchObject({ ok: true, currentRevision: 1 }); const events: CronEvent[] = []; - const durableJobsAtRemoval: Array> = []; + const durableStateAtRemoval: Array< + Promise<{ + jobs: CronJob[]; + scratch: ReturnType; + }> + > = []; const state = createState({ storePath, nowMs, onEvent: (event) => { events.push(structuredClone(event)); if (event.action === "removed") { - durableJobsAtRemoval.push(loadCronStore(storePath).then((store) => store.jobs)); + const scratch = readCronJobScratchState(storePath, job.id); + durableStateAtRemoval.push( + loadCronStore(storePath).then((store) => ({ jobs: store.jobs, scratch })), + ); } }, }); @@ -127,9 +157,15 @@ describe.each(removalPaths)("cron one-shot removal via %s", (path) => { }, }, }); - expect(durableJobsAtRemoval).toHaveLength(1); - await expect(Promise.all(durableJobsAtRemoval)).resolves.toEqual([[]]); + expect(durableStateAtRemoval).toHaveLength(1); + await expect(Promise.all(durableStateAtRemoval)).resolves.toEqual([ + { jobs: [], scratch: { currentRevision: 0 } }, + ]); expect(state.store?.jobs).toEqual([]); + + const replacement = await add(state, createReplacementInput(job.id)); + expect(replacement.id).toBe(job.id); + expect(readCronJobScratchState(storePath, job.id)).toEqual({ currentRevision: 0 }); } finally { clearStateTimer(state); } @@ -140,6 +176,14 @@ describe.each(removalPaths)("cron one-shot removal via %s", (path) => { const nowMs = Date.parse("2026-07-10T12:00:00.000Z"); const job = createDueOneShot(`rollback-${path.replaceAll(" ", "-")}`, nowMs); await saveCronStore(storePath, { version: 1, jobs: [job] }); + writeCronJobScratch({ + storePath, + jobId: job.id, + content: "scratch must survive rollback", + sourceSha256: "original-source", + nowMs: nowMs - 1, + }); + const scratchBefore = readCronJobScratchState(storePath, job.id); const events: CronEvent[] = []; let listedAfterFinished: Promise | undefined; @@ -184,6 +228,7 @@ describe.each(removalPaths)("cron one-shot removal via %s", (path) => { expect(state.durableNextRunAtMsByJobId).toEqual( new Map([[job.id, durableJob?.state.nextRunAtMs]]), ); + expect(readCronJobScratchState(storePath, job.id)).toEqual(scratchBefore); } finally { clearStateTimer(state); } diff --git a/src/cron/service/ops-lifecycle.ts b/src/cron/service/ops-lifecycle.ts index 0d3c3d8cc872..42052d3d16f0 100644 --- a/src/cron/service/ops-lifecycle.ts +++ b/src/cron/service/ops-lifecycle.ts @@ -10,7 +10,7 @@ import { STARTUP_INTERRUPTED_ERROR, } from "./startup-run-repair.js"; import type { CronServiceState, DeferredCronNotifications } from "./state.js"; -import { ensureLoaded, persist } from "./store.js"; +import { ensureLoaded, persist, pruneCronJobScratchAfterCommit } from "./store.js"; import { tryFindCronTaskRunIdForRecovery, tryFindFinalizedCronTaskRun } from "./task-runs.js"; import { armTimer, runMissedJobs, stopTimer } from "./timer.js"; @@ -92,10 +92,13 @@ export async function start(state: CronServiceState) { if (repairedAnyStartupRun || jobs.length > 0) { // Recovery notifications describe repaired durable rows, so never // publish them until the startup write has committed successfully. - await persist(state, { + const persisted = await persist(state, { ...(repairedAnyStartupRun ? {} : { stateOnly: true }), postPersistNotifications, }); + if (persisted) { + pruneCronJobScratchAfterCommit(state, completedJobIdsToDelete); + } } }); diff --git a/src/cron/service/ops-mutations.ts b/src/cron/service/ops-mutations.ts index f71daf77996c..f8be1efae67b 100644 --- a/src/cron/service/ops-mutations.ts +++ b/src/cron/service/ops-mutations.ts @@ -13,7 +13,6 @@ import { onCronJobInactive, } from "../active-jobs.js"; import { cronSchedulingInputsEqual } from "../schedule-identity.js"; -import { deleteCronJobScratch } from "../scratch-store.js"; import { removeCronJobBaseSession } from "../session-reaper.js"; import { removeStaleCronJobFamilyRows } from "../store.js"; import { createCronStreamSourceIdentity, cronStreamScheduleKey } from "../stream-schedule.js"; @@ -51,6 +50,7 @@ import { ensureLoaded, persist, persistOrRestore, + pruneCronJobScratchAfterCommit, runPostPersistCronNotifications, snapshotStoreForRollback, type CronRollbackSnapshot, @@ -533,13 +533,7 @@ export async function remove( release, }; } - try { - deleteCronJobScratch(state.deps.storePath, id); - } catch (error) { - // The job deletion is already durable. Scratch cleanup is idempotent and - // must not turn a committed removal into a retryable API failure. - state.deps.log.warn({ jobId: id, err: String(error) }, "cron: scratch cleanup failed"); - } + pruneCronJobScratchAfterCommit(state, [id]); armTimer(state); emit(state, { jobId: id, action: "removed", job: removedJob }); return { ok: true, removed: true } as const; @@ -614,6 +608,12 @@ export async function removeAgentJobsTransactional( armTimer(state); for (const job of removedJobs) { noteActiveCronJobRemoval(job.id); + } + pruneCronJobScratchAfterCommit( + state, + removedJobs.map((job) => job.id), + ); + for (const job of removedJobs) { emit(state, { jobId: job.id, action: "removed", job }); } throw error; @@ -637,15 +637,11 @@ export async function removeAgentJobsTransactional( runPostPersistCronNotifications(state, postPersistNotifications); for (const job of removedJobs) { noteActiveCronJobRemoval(job.id); - try { - deleteCronJobScratch(state.deps.storePath, job.id); - } catch (error) { - state.deps.log.warn( - { jobId: job.id, err: String(error) }, - "cron: agent scratch cleanup failed", - ); - } } + pruneCronJobScratchAfterCommit( + state, + removedJobs.map((job) => job.id), + ); armTimer(state); for (const job of removedJobs) { emit(state, { jobId: job.id, action: "removed", job }); diff --git a/src/cron/service/ops-run.ts b/src/cron/service/ops-run.ts index 45259b3f5a01..3a4252155d5c 100644 --- a/src/cron/service/ops-run.ts +++ b/src/cron/service/ops-run.ts @@ -22,7 +22,12 @@ import { releaseQueuedCronRun, runWithCronAdmission } from "./run-admission.js"; import { mergeManualRunSnapshotAfterReload } from "./startup-run-repair.js"; import type { CronServiceState, CronWakeMode, DeferredCronNotifications } from "./state.js"; import { emit } from "./state.js"; -import { ensureLoaded, persistOrRestore, snapshotStoreForRollback } from "./store.js"; +import { + ensureLoaded, + persistOrRestore, + pruneCronJobScratchAfterCommit, + snapshotStoreForRollback, +} from "./store.js"; import { tryFinishCronTaskRunWithoutHistory } from "./task-runs.js"; import { resolveCronRunScheduleOwnership, @@ -304,6 +309,7 @@ async function finishPreparedManualRun( postPersistNotifications, }); if (removedJob) { + pruneCronJobScratchAfterCommit(state, [removedJob.id]); emit(state, { jobId: removedJob.id, action: "removed", job: removedJob }); } finalized = true; diff --git a/src/cron/service/ops.test.ts b/src/cron/service/ops.test.ts index bca5cab8f730..d251dec3612a 100644 --- a/src/cron/service/ops.test.ts +++ b/src/cron/service/ops.test.ts @@ -11,6 +11,7 @@ import { formatTaskStatusDetail } from "../../tasks/task-status.js"; import { withEnvAsync } from "../../test-utils/env.js"; import { createCronExecutionId } from "../run-id.js"; import * as cronSchedule from "../schedule.js"; +import { readCronJobScratchState, writeCronJobScratch } from "../scratch-store.js"; import { setupCronServiceSuite, writeCronStoreSnapshot } from "../service.test-harness.js"; import * as cronStoreModule from "../store.js"; import { loadCronJobsStoreWithConfigJobs, loadCronStore } from "../store.js"; @@ -836,6 +837,84 @@ describe("cron service ops seam coverage", () => { }, ); + it("prunes scratch when startup deletes a finalized delete-after-run one-shot", async () => { + const { storePath } = await makeStorePath(); + const now = Date.parse("2026-03-23T12:00:00.000Z"); + const startedAt = now - 30_000; + const endedAt = startedAt + 4_000; + + await withStateDirForStorePath(storePath, async () => { + const job = createDueIsolatedJob(now); + job.id = "startup-finalized-delete-after-run"; + job.name = "startup finalized delete after run"; + job.deleteAfterRun = true; + job.schedule = { kind: "at", at: new Date(startedAt).toISOString() }; + job.state = { runningAtMs: startedAt, nextRunAtMs: startedAt }; + await writeCronStoreSnapshot({ storePath, jobs: [job] }); + expect( + writeCronJobScratch({ + storePath, + jobId: job.id, + content: "completed one-shot scratch", + nowMs: startedAt, + }), + ).toMatchObject({ ok: true, currentRevision: 1 }); + + const events: CronEvent[] = []; + const state = createCronServiceState({ + storePath, + cronEnabled: true, + log: logger, + nowMs: () => now, + enqueueSystemEvent: vi.fn(), + requestHeartbeat: vi.fn(), + runIsolatedAgentJob: vi.fn(async () => ({ status: "ok" as const })), + onEvent: (event) => events.push(structuredClone(event)), + }); + const taskRunId = tryCreateCronTaskRun({ state, job, startedAt }); + if (!taskRunId) { + throw new Error("expected cron task run"); + } + tryFinishCronTaskRun(state, { + taskRunId, + job, + event: { + jobId: job.id, + action: "finished", + job, + status: "ok", + summary: "completed before restart", + runAtMs: startedAt, + durationMs: endedAt - startedAt, + }, + }); + + try { + await start(state); + + expect((await loadCronStore(storePath)).jobs).toEqual([]); + expect(readCronJobScratchState(storePath, job.id)).toEqual({ currentRevision: 0 }); + expect( + events.filter((event) => event.action === "finished" || event.action === "removed"), + ).toEqual([]); + + const replacement = await add(state, { + id: job.id, + name: "same-id replacement", + enabled: true, + schedule: { kind: "every", everyMs: 60_000 }, + sessionTarget: "isolated", + wakeMode: "next-heartbeat", + payload: { kind: "agentTurn", message: "replacement work" }, + }); + expect(replacement.id).toBe(job.id); + expect(readCronJobScratchState(storePath, job.id)).toEqual({ currentRevision: 0 }); + } finally { + stop(state); + } + }); + }); + it("keeps a finalized one-shot disabled when startup restores its stale marker", async () => { const { storePath } = await makeStorePath(); const now = Date.parse("2026-03-23T12:00:00.000Z"); @@ -2027,6 +2106,16 @@ describe("cron service ops persist rollback", () => { ...makeCreateInput("deleted agent job"), agentId: "doomed", }); + expect( + writeCronJobScratch({ + storePath, + jobId: removed.id, + content: "deleted agent scratch", + sourceSha256: "deleted-agent-source", + nowMs: now - 1, + }), + ).toMatchObject({ ok: true, currentRevision: 1 }); + const scratchBefore = readCronJobScratchState(storePath, removed.id); const malformed = await add(state, { ...makeCreateInput("malformed surviving job"), agentId: "survivor", @@ -2085,6 +2174,21 @@ describe("cron service ops persist rollback", () => { const persisted = await loadCronStore(storePath); expect(persisted.jobs.some((job) => job.id === removed.id)).toBe(rolledBack); expect(persisted.jobs.find((job) => job.id === malformed.id)?.enabled).toBe(rolledBack); + expect(readCronJobScratchState(storePath, removed.id)).toEqual( + rolledBack ? scratchBefore : { currentRevision: 0 }, + ); + if (!rolledBack) { + const replacement = await add(state, { + ...makeCreateInput("same-id replacement"), + id: removed.id, + agentId: "survivor", + }); + expect(replacement.id).toBe(removed.id); + expect(readCronJobScratchState(storePath, removed.id)).toEqual({ currentRevision: 0 }); + if (state.timer) { + clearTimeout(state.timer); + } + } }, ); diff --git a/src/cron/service/store.ts b/src/cron/service/store.ts index f5554ecd3aff..5c307dea8138 100644 --- a/src/cron/service/store.ts +++ b/src/cron/service/store.ts @@ -3,6 +3,7 @@ import { normalizeCronJobIdentityFields } from "../normalize-job-identity.js"; import { normalizeCronJobInput } from "../normalize.js"; import { getInvalidPersistedCronJobReason } from "../persisted-shape.js"; import { cronSchedulingInputsEqual } from "../schedule-identity.js"; +import { deleteCronJobScratch } from "../scratch-store.js"; import { isInvalidCronSessionTargetIdError } from "../session-target.js"; import { getCronJobsStoreRevision, @@ -320,6 +321,23 @@ export function runPostPersistCronNotifications( } } +/** Best-effort scratch pruning after the owning job deletions are durable. */ +export function pruneCronJobScratchAfterCommit( + state: CronServiceState, + committedJobIds: Iterable, +) { + for (const jobId of committedJobIds) { + try { + deleteCronJobScratch(state.deps.storePath, jobId); + } catch (error) { + state.deps.log.warn( + { jobId, err: String(error) }, + "cron: post-commit scratch cleanup failed", + ); + } + } +} + /** Captures the live cron state that must stay aligned with the durable store. */ export function snapshotStoreForRollback(state: CronServiceState): CronRollbackSnapshot { return { diff --git a/src/cron/service/timer-outcome-finalization.ts b/src/cron/service/timer-outcome-finalization.ts index 9d7d028dba16..93f327d09f82 100644 --- a/src/cron/service/timer-outcome-finalization.ts +++ b/src/cron/service/timer-outcome-finalization.ts @@ -6,7 +6,12 @@ import { recomputeNextRunsForMaintenance } from "./jobs.js"; import { locked } from "./locked.js"; import { clearQueuedCronRunReservationMarker, releaseQueuedCronRun } from "./run-admission.js"; import { emit, type CronServiceState, type DeferredCronNotifications } from "./state.js"; -import { ensureLoaded, persistOrRestore, snapshotStoreForRollback } from "./store.js"; +import { + ensureLoaded, + persistOrRestore, + pruneCronJobScratchAfterCommit, + snapshotStoreForRollback, +} from "./store.js"; import { tryFinishCronTaskRunWithoutHistory } from "./task-runs.js"; import type { TimedCronRunOutcome } from "./timer-execution-timeout.js"; import { applyOutcomeToStoredJob } from "./timer-outcomes.js"; @@ -160,6 +165,10 @@ export async function finalizeCompletedCronRunOutcomes( await persistOrRestore(state, rollbackSnapshot, { postPersistNotifications, }); + pruneCronJobScratchAfterCommit( + state, + removedJobs.map((job) => job.id), + ); finishPersistedQuietCronTaskRuns(state, finalizedOutcomes); for (const removedJob of removedJobs) { emit(state, { jobId: removedJob.id, action: "removed", job: removedJob });