From 2c7e989686bac28bfe0f7a27d34a56158c65674b Mon Sep 17 00:00:00 2001 From: Masato Hoshino Date: Wed, 1 Jul 2026 14:39:11 +0900 Subject: [PATCH] fix(doctor): recover legacy cron archive across devices (#98217) * fix(doctor): recover legacy cron archive across devices * fix(doctor): harden legacy cron archive retries * fix(doctor): canonicalize shipped cron retry shape * fix(doctor): persist legacy cron migration receipts --------- Co-authored-by: Peter Steinberger --- src/commands/doctor/cron/index.test.ts | 556 ++++++++++++++++++ src/commands/doctor/cron/index.ts | 86 ++- .../doctor/cron/legacy-store-migration.ts | 421 ++++++++++++- src/commands/doctor/cron/migration-ledger.ts | 115 ++++ src/commands/doctor/cron/repair-plan.ts | 9 +- src/commands/doctor/cron/store-migration.ts | 16 +- src/cron/store.ts | 19 + 7 files changed, 1170 insertions(+), 52 deletions(-) create mode 100644 src/commands/doctor/cron/migration-ledger.ts diff --git a/src/commands/doctor/cron/index.test.ts b/src/commands/doctor/cron/index.test.ts index 55d69967b4f..cba70a7a6f7 100644 --- a/src/commands/doctor/cron/index.test.ts +++ b/src/commands/doctor/cron/index.test.ts @@ -1,4 +1,5 @@ // Doctor cron index tests cover cron doctor checks and repair entrypoints. +import fsSync from "node:fs"; import fs from "node:fs/promises"; import os from "node:os"; import path from "node:path"; @@ -13,6 +14,7 @@ import { saveCronStore, } from "../../../cron/store.js"; import { runOpenClawStateWriteTransaction } from "../../../state/openclaw-state-db.js"; +import { withRestoredMocks } from "../../../test-utils/vitest-spies.js"; import { collectLegacyWhatsAppCrontabHealthWarning, maybeRepairLegacyCronStore, @@ -193,6 +195,20 @@ function expectNoNoteContaining(message: string, title: string): void { ).toBe(false); } +function createFsError(code: string, message: string): NodeJS.ErrnoException { + return Object.assign(new Error(`${code}: ${message}`), { code }); +} + +function mockExdevRename(filePath: string) { + const realRename = fs.rename.bind(fs); + return vi.spyOn(fs, "rename").mockImplementation(async (oldPath, newPath) => { + if (oldPath === filePath) { + throw createFsError("EXDEV", "cross-device link not permitted, rename"); + } + return await realRename(oldPath, newPath); + }); +} + describe("maybeRepairLegacyCronStore", () => { it("reports quarantined cron rows even when the active store is already sanitized", async () => { const storePath = await makeTempStorePath(); @@ -444,6 +460,517 @@ describe("maybeRepairLegacyCronStore", () => { expectNoteContaining("Cron store migrated to SQLite", "Doctor changes"); }); + it("falls back to copy+unlink when renaming the legacy cron store fails with EXDEV", async () => { + const storePath = await makeTempStorePath(); + const archivePath = `${storePath}.migrated`; + const sourceMtime = new Date("2026-01-02T03:04:05.000Z"); + await writeCronStore(storePath, [createLegacyCronJob()]); + await fs.chmod(storePath, 0o640); + await fs.utimes(storePath, sourceMtime, sourceMtime); + + const renameSpy = mockExdevRename(storePath); + const realOpen = fs.open.bind(fs); + let archiveFileSynced = false; + const openSpy = vi.spyOn(fs, "open").mockImplementation(async (...args) => { + const handle = await realOpen(...args); + if (args[0] === archivePath && args[1] === "r+") { + const realSync = handle.sync.bind(handle); + vi.spyOn(handle, "sync").mockImplementation(async () => { + archiveFileSynced = true; + await realSync(); + }); + } + return handle; + }); + + await withRestoredMocks([openSpy, renameSpy], async () => { + await maybeRepairLegacyCronStore({ + cfg: createCronConfig(storePath), + options: {}, + prompter: makePrompter(true), + }); + + expect(renameSpy).toHaveBeenCalled(); + expect(archiveFileSynced).toBe(true); + await expect(fs.stat(storePath)).rejects.toMatchObject({ code: "ENOENT" }); + await expect(fs.readFile(archivePath, "utf-8")).resolves.toContain("legacy-job"); + const archiveStat = await fs.stat(archivePath); + expect(archiveStat.mode & 0o777).toBe(0o640); + expect(archiveStat.mtimeMs).toBe(sourceMtime.getTime()); + expectNoteContaining("Cron store migrated to SQLite", "Doctor changes"); + expectNoNoteContaining("could not archive the legacy cron file", "Doctor warnings"); + }); + + // A second doctor pass must not re-detect (and re-warn about) the archived store. + noteMock.mockClear(); + await maybeRepairLegacyCronStore({ + cfg: createCronConfig(storePath), + options: {}, + prompter: makePrompter(true), + }); + expectNoNoteContaining("Legacy cron job storage detected", "Cron"); + }); + + it("refuses a migration plan when the legacy source changes during confirmation", async () => { + const storePath = await makeTempStorePath(); + await writeCronStore(storePath, [createLegacyCronJob()]); + const changedJob = createLegacyCronJob({ jobId: "changed-job", name: "Changed job" }); + const prompter = { + confirm: vi.fn(async () => { + await writeCronStore(storePath, [changedJob]); + return true; + }), + }; + + await maybeRepairLegacyCronStore({ + cfg: createCronConfig(storePath), + options: {}, + prompter, + }); + + expect(await readPersistedJobs(storePath)).toHaveLength(0); + await expect(fs.readFile(storePath, "utf-8")).resolves.toContain("changed-job"); + await expect(fs.stat(`${storePath}.migrated`)).rejects.toMatchObject({ code: "ENOENT" }); + expectNoteContaining("changed while doctor was preparing", "Doctor warnings"); + expectNoNoteContaining("Cron store migrated to SQLite", "Doctor changes"); + + noteMock.mockClear(); + await maybeRepairLegacyCronStore({ + cfg: createCronConfig(storePath), + options: {}, + prompter: makePrompter(true), + }); + expect((await readPersistedJobs(storePath)).map((job) => job.id)).toEqual(["changed-job"]); + await expect(fs.stat(storePath)).rejects.toMatchObject({ code: "ENOENT" }); + expectNoteContaining("Cron store migrated to SQLite", "Doctor changes"); + }); + + it("keeps a source that changes during an EXDEV copy and imports it on retry", async () => { + const storePath = await makeTempStorePath(); + const archivePath = `${storePath}.migrated`; + await writeCronStore(storePath, [createLegacyCronJob()]); + + const renameSpy = mockExdevRename(storePath); + const realCopyFile = fs.copyFile.bind(fs); + const copyFileSpy = vi.spyOn(fs, "copyFile").mockImplementation(async (src, dest, mode) => { + await realCopyFile(src, dest, mode); + if (src === storePath) { + await writeCronStore(storePath, [ + createLegacyCronJob({ jobId: "late-job", name: "Late job" }), + ]); + } + }); + + await withRestoredMocks([copyFileSpy, renameSpy], async () => { + await maybeRepairLegacyCronStore({ + cfg: createCronConfig(storePath), + options: {}, + prompter: makePrompter(true), + }); + }); + + expect((await readPersistedJobs(storePath)).map((job) => job.id)).toEqual(["legacy-job"]); + await expect(fs.readFile(storePath, "utf-8")).resolves.toContain("late-job"); + await expect(fs.stat(archivePath)).rejects.toMatchObject({ code: "ENOENT" }); + expectNoteContaining("changed during archival", "Doctor warnings"); + + noteMock.mockClear(); + await maybeRepairLegacyCronStore({ + cfg: createCronConfig(storePath), + options: {}, + prompter: makePrompter(true), + }); + expect((await readPersistedJobs(storePath)).map((job) => job.id)).toEqual([ + "legacy-job", + "late-job", + ]); + await expect(fs.stat(storePath)).rejects.toMatchObject({ code: "ENOENT" }); + await expect(fs.stat(archivePath)).resolves.toBeTruthy(); + }); + + it("restores an archived state sidecar when the primary archive fails", async () => { + const storePath = await makeTempStorePath(); + const statePath = storePath.replace(/\.json$/, "-state.json"); + await writeCronStore(storePath, [createLegacyCronJob()]); + await fs.writeFile(statePath, JSON.stringify({ version: 1, jobs: {} }), "utf-8"); + + const realRename = fs.rename.bind(fs); + const renameSpy = vi.spyOn(fs, "rename").mockImplementation(async (oldPath, newPath) => { + if (oldPath === storePath) { + throw createFsError("EIO", "primary archive failed"); + } + return await realRename(oldPath, newPath); + }); + + await withRestoredMocks([renameSpy], async () => { + await maybeRepairLegacyCronStore({ + cfg: createCronConfig(storePath), + options: {}, + prompter: makePrompter(true), + }); + }); + + await expect(fs.stat(storePath)).resolves.toBeTruthy(); + await expect(fs.stat(statePath)).resolves.toBeTruthy(); + await expect(fs.stat(`${statePath}.migrated`)).rejects.toMatchObject({ code: "ENOENT" }); + expectNoteContaining("EIO", "Doctor warnings"); + expectNoNoteContaining("Cron store migrated to SQLite", "Doctor changes"); + + noteMock.mockClear(); + await maybeRepairLegacyCronStore({ + cfg: createCronConfig(storePath), + options: {}, + prompter: makePrompter(true), + }); + await expect(fs.stat(storePath)).rejects.toMatchObject({ code: "ENOENT" }); + await expect(fs.stat(statePath)).rejects.toMatchObject({ code: "ENOENT" }); + await expect(fs.stat(`${statePath}.migrated`)).resolves.toBeTruthy(); + expectNoteContaining("Cron store migrated to SQLite", "Doctor changes"); + }); + + it("restores the primary source when a state sidecar is recreated during archival", async () => { + const storePath = await makeTempStorePath(); + const statePath = storePath.replace(/\.json$/, "-state.json"); + await writeCronStore(storePath, [createLegacyCronJob()]); + await fs.writeFile(statePath, JSON.stringify({ version: 1, jobs: {} }), "utf-8"); + + const realRename = fs.rename.bind(fs); + const renameSpy = vi.spyOn(fs, "rename").mockImplementation(async (oldPath, newPath) => { + if (oldPath === storePath) { + await fs.writeFile( + statePath, + JSON.stringify({ version: 1, jobs: { "legacy-job": { state: { lastRunAtMs: 2 } } } }), + "utf-8", + ); + } + return await realRename(oldPath, newPath); + }); + + await withRestoredMocks([renameSpy], async () => { + await maybeRepairLegacyCronStore({ + cfg: createCronConfig(storePath), + options: {}, + prompter: makePrompter(true), + }); + }); + + await expect(fs.stat(storePath)).resolves.toBeTruthy(); + await expect(fs.readFile(statePath, "utf-8")).resolves.toContain("lastRunAtMs"); + await expect(fs.stat(`${statePath}.migrated`)).resolves.toBeTruthy(); + expectNoteContaining("state appeared after", "Doctor warnings"); + expectNoteContaining("archive rollback failed", "Doctor warnings"); + expectNoNoteContaining("Cron store migrated to SQLite", "Doctor changes"); + }); + + it("reports a late state access failure without rejecting doctor", async () => { + const storePath = await makeTempStorePath(); + const statePath = storePath.replace(/\.json$/, "-state.json"); + await writeCronStore(storePath, [createLegacyCronJob()]); + + const realAccess = fs.access.bind(fs); + let stateAccesses = 0; + const accessSpy = vi.spyOn(fs, "access").mockImplementation(async (...args) => { + if (args[0] === statePath && ++stateAccesses === 2) { + throw createFsError("EIO", "state access failed"); + } + return await realAccess(...args); + }); + + await withRestoredMocks([accessSpy], async () => { + await expect( + maybeRepairLegacyCronStore({ + cfg: createCronConfig(storePath), + options: {}, + prompter: makePrompter(true), + }), + ).resolves.toBeUndefined(); + }); + + await expect(fs.stat(storePath)).resolves.toBeTruthy(); + expectNoteContaining("state access failed", "Doctor warnings"); + expectNoNoteContaining("Cron store migrated to SQLite", "Doctor changes"); + }); + + it("removes a partial copy and warns honestly when archiving fails", async () => { + const storePath = await makeTempStorePath(); + const archivePath = `${storePath}.migrated`; + await writeCronStore(storePath, [createLegacyCronJob()]); + + const renameSpy = mockExdevRename(storePath); + const realCopyFile = fs.copyFile; + const copyFileSpy = vi.spyOn(fs, "copyFile").mockImplementation(async (src, dest, mode) => { + if (src === storePath) { + await fs.writeFile(dest, "partial", "utf-8"); + throw createFsError("ENOSPC", "no space left, copyfile"); + } + return realCopyFile(src, dest, mode); + }); + + await withRestoredMocks([copyFileSpy, renameSpy], async () => { + await maybeRepairLegacyCronStore({ + cfg: createCronConfig(storePath), + options: {}, + prompter: makePrompter(true), + }); + + // Both rename and the copy+unlink fallback failed, so the legacy file must remain + // and doctor must surface a warning instead of claiming a finished migration. + await expect(fs.stat(storePath)).resolves.toBeTruthy(); + await expect(fs.stat(archivePath)).rejects.toMatchObject({ code: "ENOENT" }); + expectNoteContaining("could not archive the legacy cron file", "Doctor warnings"); + expectNoteContaining("ENOSPC", "Doctor warnings"); + expectNoNoteContaining("Cron store migrated to SQLite", "Doctor changes"); + }); + }); + + it("accepts a failed copy that already removed its destination", async () => { + const storePath = await makeTempStorePath(); + const archivePath = `${storePath}.migrated`; + await writeCronStore(storePath, [createLegacyCronJob()]); + + const renameSpy = mockExdevRename(storePath); + const realCopyFile = fs.copyFile.bind(fs); + const copyFileSpy = vi.spyOn(fs, "copyFile").mockImplementation(async (src, dest) => { + if (src === storePath) { + await fs.unlink(dest); + throw createFsError("EIO", "copyfile failed after destination cleanup"); + } + return await realCopyFile(src, dest); + }); + + await withRestoredMocks([copyFileSpy, renameSpy], async () => { + await maybeRepairLegacyCronStore({ + cfg: createCronConfig(storePath), + options: {}, + prompter: makePrompter(true), + }); + + await expect(fs.stat(storePath)).resolves.toBeTruthy(); + await expect(fs.stat(archivePath)).rejects.toMatchObject({ code: "ENOENT" }); + expectNoteContaining("EIO", "Doctor warnings"); + expectNoNoteContaining("partial archive remains", "Doctor warnings"); + expectNoNoteContaining("Cron store migrated to SQLite", "Doctor changes"); + }); + }); + + it("reports a source stat failure without aborting doctor", async () => { + const storePath = await makeTempStorePath(); + await writeCronStore(storePath, [createLegacyCronJob()]); + + const renameSpy = mockExdevRename(storePath); + const realStat = fs.stat.bind(fs); + const statSpy = vi.spyOn(fs, "stat").mockImplementation(async (...args) => { + if (args[0] === storePath) { + throw createFsError("EIO", "stat failed"); + } + return await realStat(...args); + }); + + await withRestoredMocks([statSpy, renameSpy], async () => { + await expect( + maybeRepairLegacyCronStore({ + cfg: createCronConfig(storePath), + options: {}, + prompter: makePrompter(true), + }), + ).resolves.toBeUndefined(); + expectNoteContaining("EIO", "Doctor warnings"); + expectNoNoteContaining("Cron store migrated to SQLite", "Doctor changes"); + }); + await expect(fs.stat(storePath)).resolves.toBeTruthy(); + }); + + it("reports an archive access failure instead of treating the source as missing", async () => { + const storePath = await makeTempStorePath(); + await writeCronStore(storePath, [createLegacyCronJob()]); + + const realAccess = fs.access.bind(fs); + let sourceAccesses = 0; + const accessSpy = vi.spyOn(fs, "access").mockImplementation(async (...args) => { + if (args[0] === storePath && ++sourceAccesses === 2) { + throw createFsError("EIO", "access failed"); + } + return await realAccess(...args); + }); + + await withRestoredMocks([accessSpy], async () => { + await maybeRepairLegacyCronStore({ + cfg: createCronConfig(storePath), + options: {}, + prompter: makePrompter(true), + }); + expectNoteContaining("EIO", "Doctor warnings"); + expectNoNoteContaining("Cron store migrated to SQLite", "Doctor changes"); + }); + await expect(fs.stat(storePath)).resolves.toBeTruthy(); + }); + + it("keeps the source and removes the partial archive when durability sync fails", async () => { + const storePath = await makeTempStorePath(); + const archivePath = `${storePath}.migrated`; + await writeCronStore(storePath, [createLegacyCronJob()]); + + const renameSpy = mockExdevRename(storePath); + const realOpen = fs.open.bind(fs); + const openSpy = vi.spyOn(fs, "open").mockImplementation(async (...args) => { + const handle = await realOpen(...args); + if (args[0] === archivePath && args[1] === "r+") { + vi.spyOn(handle, "sync").mockRejectedValueOnce(createFsError("EIO", "fsync failed")); + } + return handle; + }); + + await withRestoredMocks([openSpy, renameSpy], async () => { + await maybeRepairLegacyCronStore({ + cfg: createCronConfig(storePath), + options: {}, + prompter: makePrompter(true), + }); + + await expect(fs.stat(storePath)).resolves.toBeTruthy(); + await expect(fs.stat(archivePath)).rejects.toMatchObject({ code: "ENOENT" }); + expectNoteContaining("EIO", "Doctor warnings"); + expectNoNoteContaining("Cron store migrated to SQLite", "Doctor changes"); + }); + }); + + it("keeps the source when syncing the archive directory fails", async () => { + const storePath = await makeTempStorePath(); + const archivePath = `${storePath}.migrated`; + await writeCronStore(storePath, [createLegacyCronJob()]); + + const renameSpy = mockExdevRename(storePath); + const realOpen = fs.open.bind(fs); + let injectedFailure = false; + const openSpy = vi.spyOn(fs, "open").mockImplementation(async (...args) => { + const handle = await realOpen(...args); + if (args[0] === path.dirname(storePath) && args[1] === "r" && !injectedFailure) { + injectedFailure = true; + vi.spyOn(handle, "sync").mockRejectedValueOnce( + createFsError("EIO", "directory fsync failed"), + ); + } + return handle; + }); + + await withRestoredMocks([openSpy, renameSpy], async () => { + await maybeRepairLegacyCronStore({ + cfg: createCronConfig(storePath), + options: {}, + prompter: makePrompter(true), + }); + + await expect(fs.stat(storePath)).resolves.toBeTruthy(); + await expect(fs.stat(archivePath)).rejects.toMatchObject({ code: "ENOENT" }); + expectNoteContaining("EIO", "Doctor warnings"); + expectNoNoteContaining("Cron store migrated to SQLite", "Doctor changes"); + }); + }); + + it.each([ + { label: "string id", jobId: "legacy-job", expectedId: "legacy-job", jobCount: 1 }, + { label: "numeric id", jobId: 7, expectedId: "7", jobCount: 1 }, + { label: "duplicate missing ids", jobId: undefined, expectedId: undefined, jobCount: 2 }, + ])( + "rolls back a $label archive and retries without duplicates", + async ({ jobId, expectedId, jobCount }) => { + const storePath = await makeTempStorePath(); + const archivePath = `${storePath}.migrated`; + await writeCronStore( + storePath, + Array.from({ length: jobCount }, () => createLegacyCronJob({ id: undefined, jobId })), + ); + + const renameSpy = mockExdevRename(storePath); + const realUnlink = fs.unlink.bind(fs); + const unlinkSpy = vi.spyOn(fs, "unlink").mockImplementation(async (target) => { + if (target === storePath) { + throw createFsError("EBUSY", "resource busy, unlink"); + } + return await realUnlink(target); + }); + + await withRestoredMocks([unlinkSpy, renameSpy], async () => { + await maybeRepairLegacyCronStore({ + cfg: createCronConfig(storePath), + options: {}, + prompter: makePrompter(true), + }); + + await expect(fs.stat(storePath)).resolves.toBeTruthy(); + await expect(fs.stat(archivePath)).rejects.toMatchObject({ code: "ENOENT" }); + expectNoteContaining("EBUSY", "Doctor warnings"); + expectNoNoteContaining("Cron store migrated to SQLite", "Doctor changes"); + }); + + const firstJobs = await readPersistedJobs(storePath); + expect(firstJobs).toHaveLength(jobCount); + if (expectedId) { + expect(firstJobs[0]?.id).toBe(expectedId); + } else { + const ids = firstJobs.map((job) => job.id); + expect(ids).toHaveLength(new Set(ids).size); + for (const id of ids) { + expect(id).toMatch(/^cron-migrated-\d+-[a-f0-9]{64}$/); + } + } + + noteMock.mockClear(); + await maybeRepairLegacyCronStore({ + cfg: createCronConfig(storePath), + options: {}, + prompter: makePrompter(true), + }); + + await expect(fs.stat(storePath)).rejects.toMatchObject({ code: "ENOENT" }); + await expect(fs.stat(archivePath)).resolves.toBeTruthy(); + await expect(fs.stat(`${archivePath}.2`)).rejects.toMatchObject({ code: "ENOENT" }); + const secondJobs = await readPersistedJobs(storePath); + expect(secondJobs).toHaveLength(jobCount); + expect(secondJobs.map((job) => job.id)).toEqual(firstJobs.map((job) => job.id)); + expectNoteContaining("Cron store migrated to SQLite", "Doctor changes"); + }, + ); + + it("does not resurrect a migrated job removed before an archive retry", async () => { + const storePath = await makeTempStorePath(); + const archivePath = `${storePath}.migrated`; + await writeCronStore(storePath, [createLegacyCronJob({ id: undefined, jobId: undefined })]); + + const renameSpy = mockExdevRename(storePath); + const realUnlink = fs.unlink.bind(fs); + const unlinkSpy = vi.spyOn(fs, "unlink").mockImplementation(async (target) => { + if (target === storePath) { + throw createFsError("EBUSY", "resource busy, unlink"); + } + return await realUnlink(target); + }); + + await withRestoredMocks([unlinkSpy, renameSpy], async () => { + await maybeRepairLegacyCronStore({ + cfg: createCronConfig(storePath), + options: {}, + prompter: makePrompter(true), + }); + expectNoteContaining("EBUSY", "Doctor warnings"); + }); + expect(await readPersistedJobs(storePath)).toHaveLength(1); + + // Simulate runtime-owned one-shot deletion after SQLite import but before cleanup retry. + await writeCurrentCronStore(storePath, []); + noteMock.mockClear(); + await maybeRepairLegacyCronStore({ + cfg: createCronConfig(storePath), + options: {}, + prompter: makePrompter(true), + }); + + expect(await readPersistedJobs(storePath)).toHaveLength(0); + await expect(fs.stat(storePath)).rejects.toMatchObject({ code: "ENOENT" }); + await expect(fs.stat(archivePath)).resolves.toBeTruthy(); + expectNoteContaining("Cron store migrated to SQLite", "Doctor changes"); + }); + it("imports legacy-only jobs when SQLite already has cron rows", async () => { const storePath = await makeTempStorePath(); await writeCurrentCronStore(storePath, [ @@ -579,6 +1106,35 @@ describe("maybeRepairLegacyCronStore", () => { expectNoteContaining("Cron run logs migrated to SQLite", "Doctor changes"); }); + it("does not report store normalization when run-log migration fails", async () => { + const storePath = await makeTempStorePath(); + await writeCurrentCronStore(storePath, [createCurrentCronJob()]); + const runLogPath = path.join(path.dirname(storePath), "runs", "sqlite-job.jsonl"); + await fs.mkdir(path.dirname(runLogPath), { recursive: true }); + await fs.writeFile(runLogPath, "{}\n", "utf-8"); + + const realReadFileSync = fsSync.readFileSync.bind(fsSync); + const readSpy = vi.spyOn(fsSync, "readFileSync").mockImplementation((filePath, options) => { + if (filePath === runLogPath) { + throw createFsError("EIO", "run-log read failed"); + } + return realReadFileSync(filePath as never, options as never) as never; + }); + + await withRestoredMocks([readSpy], async () => { + await maybeRepairLegacyCronStore({ + cfg: createCronConfig(storePath), + options: {}, + prompter: makePrompter(true), + }); + }); + + await expect(fs.stat(runLogPath)).resolves.toBeTruthy(); + expectNoteContaining("run-log read failed", "Doctor warnings"); + expectNoNoteContaining("Cron store normalized", "Doctor changes"); + expectNoNoteContaining("Cron run logs migrated", "Doctor changes"); + }); + it("does not claim legacy store detected when only non-legacy issues exist (#92683)", async () => { const storePath = await makeTempStorePath(); await writeCurrentCronStore(storePath, [ diff --git a/src/commands/doctor/cron/index.ts b/src/commands/doctor/cron/index.ts index 40bf680d649..feb9ae5ba13 100644 --- a/src/commands/doctor/cron/index.ts +++ b/src/commands/doctor/cron/index.ts @@ -10,6 +10,7 @@ import { resolveCronJobsStorePath, saveCronQuarantineFile, saveCronJobsStore, + saveCronJobsStoreWithMetadata, } from "../../../cron/store.js"; import type { CronJob } from "../../../cron/types.js"; import { shortenHomePath } from "../../../utils.js"; @@ -25,9 +26,16 @@ import { } from "./legacy-run-log-migration.js"; import { archiveLegacyCronStoreForMigration, + assertLegacyCronMigrationSourceCurrent, legacyCronStoreFilesExist, loadLegacyCronStoreForMigration, + type LegacyCronMigrationSource, } from "./legacy-store-migration.js"; +import { + acquireLegacyCronMigrationReceipt, + hasLegacyCronMigrationReceipt, + markLegacyCronMigrationSourceRemoved, +} from "./migration-ledger.js"; import { formatLegacyIssuePreview, formatUnresolvedCommandPromptAdvisory, @@ -63,6 +71,8 @@ type LegacyCronRepairState = { quarantinePath: string; legacyStoreDetected: boolean; legacyRunLogDetected: boolean; + legacyMigrationSource?: LegacyCronMigrationSource; + legacyMigrationAlreadyImported: boolean; legacyImportCount: number; sqliteProjectionBackfillCount: number; rawJobs: Array>; @@ -106,14 +116,22 @@ async function loadLegacyCronRepairState(params: { : 0; let rawJobs = currentJobs; let legacyImportCount = 0; + let legacyMigrationSource: LegacyCronMigrationSource | undefined; + let legacyMigrationAlreadyImported = false; if (legacyStoreDetected) { - const legacyStore = (await loadLegacyCronStoreForMigration(storePath)).store; - const merged = mergeLegacyCronJobs({ - currentJobs: rawJobs, - legacyJobs: legacyStore.jobs as unknown as Array>, - }); - rawJobs = merged.jobs; - legacyImportCount = merged.importedCount; + const loadedLegacy = await loadLegacyCronStoreForMigration(storePath); + legacyMigrationSource = loadedLegacy.migrationSource; + legacyMigrationAlreadyImported = legacyMigrationSource + ? hasLegacyCronMigrationReceipt(legacyMigrationSource) + : false; + if (!legacyMigrationAlreadyImported) { + const merged = mergeLegacyCronJobs({ + currentJobs: rawJobs, + legacyJobs: loadedLegacy.store.jobs as unknown as Array>, + }); + rawJobs = merged.jobs; + legacyImportCount = merged.importedCount; + } } return { @@ -121,6 +139,8 @@ async function loadLegacyCronRepairState(params: { quarantinePath, legacyStoreDetected, legacyRunLogDetected, + legacyMigrationSource, + legacyMigrationAlreadyImported, legacyImportCount, sqliteProjectionBackfillCount, rawJobs, @@ -144,18 +164,18 @@ async function applyLegacyCronStoreRepair(params: { const dreamingMigration = migrateLegacyDreamingPayloadShape(state.rawJobs); warnings.push(...notifyMigration.warnings); - const changed = - state.legacyStoreDetected || - state.legacyRunLogDetected || + const storeChanged = + (state.legacyStoreDetected && !state.legacyMigrationAlreadyImported) || state.sqliteProjectionBackfillCount > 0 || normalized.mutated || notifyMigration.changed || dreamingMigration.changed; + const changed = state.legacyStoreDetected || state.legacyRunLogDetected || storeChanged; if (!changed && warnings.length === 0) { return { changes, warnings }; } - if (changed) { + if (storeChanged) { try { if (normalized.removedJobs.length > 0) { await saveCronQuarantineFile({ @@ -168,10 +188,19 @@ async function applyLegacyCronStoreRepair(params: { })), }); } - await saveCronJobsStore(state.storePath, { + const store = { version: 1, jobs: state.rawJobs as unknown as CronJob[], - }); + } as const; + const migrationSource = state.legacyMigrationSource; + if (migrationSource && !state.legacyMigrationAlreadyImported) { + await assertLegacyCronMigrationSourceCurrent(migrationSource); + await saveCronJobsStoreWithMetadata(state.storePath, store, (db) => { + return acquireLegacyCronMigrationReceipt(db, migrationSource); + }); + } else { + await saveCronJobsStore(state.storePath, store); + } } catch (err) { return { changes, @@ -195,15 +224,38 @@ async function applyLegacyCronStoreRepair(params: { } if (state.legacyStoreDetected) { - await archiveLegacyCronStoreForMigration(state.storePath); - changes.push( - `Cron store migrated to SQLite at ${shortenHomePath(state.storePath)}.${formatRunLogMigrationNote(importedRunLogs)}`, + const archiveResult = await archiveLegacyCronStoreForMigration( + state.storePath, + state.legacyMigrationSource, ); + if (archiveResult.ok) { + if (state.legacyMigrationSource) { + try { + markLegacyCronMigrationSourceRemoved(state.legacyMigrationSource); + } catch (err) { + warnings.push( + `Cron store was archived, but its migration receipt could not be finalized: ${errorMessage(err)}`, + ); + } + } + changes.push( + `Cron store migrated to SQLite at ${shortenHomePath(state.storePath)}.${formatRunLogMigrationNote(importedRunLogs)}`, + ); + } else { + // SQLite already holds the migrated jobs, but the legacy file could not be + // archived (e.g. EXDEV copy+unlink failed), so report it honestly instead of + // claiming a finished migration; doctor re-detects the leftover and retries. + for (const failure of archiveResult.failures) { + warnings.push( + `Migrated cron jobs to SQLite but could not archive the legacy cron file at ${shortenHomePath(failure.path)}: ${failure.reason}. Remove it manually or rerun ${formatCliCommand("openclaw doctor --fix")} to retry.`, + ); + } + } } else if (state.legacyRunLogDetected && importedRunLogs > 0) { changes.push( `Cron run logs migrated to SQLite at ${shortenHomePath(state.storePath)}.${formatRunLogMigrationNote(importedRunLogs)}`, ); - } else if (changed) { + } else if (storeChanged) { changes.push(`Cron store normalized at ${shortenHomePath(state.storePath)}.`); } if (dreamingMigration.rewrittenCount > 0) { diff --git a/src/commands/doctor/cron/legacy-store-migration.ts b/src/commands/doctor/cron/legacy-store-migration.ts index 19b3c175444..354a139aed4 100644 --- a/src/commands/doctor/cron/legacy-store-migration.ts +++ b/src/commands/doctor/cron/legacy-store-migration.ts @@ -1,8 +1,12 @@ // Legacy cron JSON/state store loader and archiver for doctor migration. +import { createHash } from "node:crypto"; import fs from "node:fs/promises"; import path from "node:path"; import { isRecord } from "../../../../packages/normalization-core/src/record-coerce.js"; -import { normalizeOptionalString } from "../../../../packages/normalization-core/src/string-coerce.js"; +import { + normalizeOptionalString, + normalizeOptionalStringifiedId, +} from "../../../../packages/normalization-core/src/string-coerce.js"; import { coerceFiniteScheduleNumber } from "../../../cron/schedule-number.js"; import { normalizeCronStaggerMs } from "../../../cron/stagger.js"; import type { @@ -14,6 +18,19 @@ import type { CronStoreFile } from "../../../cron/types.js"; import { parseJsonWithJson5Fallback } from "../../../utils/parse-json-compat.js"; const LEGACY_CRON_ARCHIVE_SUFFIX = ".migrated"; +const legacyCronMigrationIds = new WeakMap, string>(); + +export function resolveLegacyCronMigrationId(job: Record): string | undefined { + return legacyCronMigrationIds.get(job); +} + +function markLegacyCronMigrationIdentity(job: Record, sourceIndex: number): void { + if (normalizeOptionalStringifiedId(job.id) ?? normalizeOptionalStringifiedId(job.jobId)) { + return; + } + const digest = createHash("sha256").update(JSON.stringify(job)).digest("hex"); + legacyCronMigrationIds.set(job, `cron-migrated-${sourceIndex}-${digest}`); +} function resolveLegacyCronStatePath(storePath: string): string { if (storePath.endsWith(".json")) { @@ -22,22 +39,295 @@ function resolveLegacyCronStatePath(storePath: string): string { return `${storePath}-state.json`; } -async function legacyCronFileExists(filePath: string): Promise { - return fs - .access(filePath) - .then(() => true) - .catch(() => false); +export type LegacyCronMigrationSource = { + sourceKey: string; + sourcePath: string; + sourceSha256: string; + statePath: string; + stateSha256?: string; + sourceSizeBytes: number; + sourceRecordCount: number; +}; + +function createLegacyCronMigrationSource(params: { + sourcePath: string; + raw: string; + statePath: string; + stateRaw?: string; + recordCount: number; +}): LegacyCronMigrationSource { + const sourceSha256 = createHash("sha256").update(params.raw).digest("hex"); + const stateSha256 = + params.stateRaw !== undefined + ? createHash("sha256").update(params.stateRaw).digest("hex") + : undefined; + const sourceKeyHash = createHash("sha256") + .update(`${params.sourcePath}\0${sourceSha256}\0${stateSha256 ?? ""}`) + .digest("hex"); + return { + sourceKey: `cron-json:${sourceKeyHash}`, + sourcePath: params.sourcePath, + sourceSha256, + statePath: params.statePath, + stateSha256, + sourceSizeBytes: Buffer.byteLength(params.raw) + Buffer.byteLength(params.stateRaw ?? ""), + sourceRecordCount: params.recordCount, + }; } -async function archiveLegacyCronFile(filePath: string): Promise { - if (!(await legacyCronFileExists(filePath))) { - return; +async function legacyCronFileExists(filePath: string): Promise { + try { + await fs.access(filePath); + return true; + } catch (err) { + if ((err as NodeJS.ErrnoException).code === "ENOENT") { + return false; + } + throw err; } +} + +type ArchiveOutcome = { ok: true; archivePath?: string } | { ok: false; reason: string }; + +function formatArchiveError(err: unknown): string { + return err instanceof Error ? err.message : String(err); +} + +function isUnsupportedDirectorySyncError(err: unknown): boolean { + const code = (err as NodeJS.ErrnoException).code; + if (code === "EINVAL" || code === "ENOTSUP" || code === "ENOSYS") { + return true; + } + return ( + process.platform === "win32" && (code === "EISDIR" || code === "EPERM" || code === "EACCES") + ); +} + +async function syncArchiveDirectory(dirPath: string): Promise { + let handle: Awaited> | undefined; + try { + handle = await fs.open(dirPath, "r"); + await handle.sync(); + } catch (err) { + if (!isUnsupportedDirectorySyncError(err)) { + throw err; + } + } finally { + await handle?.close(); + } +} + +async function sha256File(filePath: string): Promise { + return createHash("sha256") + .update(await fs.readFile(filePath)) + .digest("hex"); +} + +/** Refuse to persist a migration plan built from legacy files that changed after loading. */ +export async function assertLegacyCronMigrationSourceCurrent( + source: LegacyCronMigrationSource, +): Promise { + if ((await sha256File(source.sourcePath)) !== source.sourceSha256) { + throw new Error("legacy cron source changed while doctor was preparing its migration"); + } + if (source.stateSha256) { + if ((await sha256File(source.statePath)) !== source.stateSha256) { + throw new Error("legacy cron state changed while doctor was preparing its migration"); + } + } else if (await legacyCronFileExists(source.statePath)) { + throw new Error("legacy cron state appeared while doctor was preparing its migration"); + } +} + +async function restoreArchivedSource( + archivePath: string, + sourcePath: string, + expectedSha256?: string, +): Promise<{ ok: true } | { ok: false; reason: string }> { + try { + if (await legacyCronFileExists(sourcePath)) { + return { + ok: false, + reason: `archive remains at ${archivePath} because a new source exists at ${sourcePath}`, + }; + } + } catch (err) { + return { + ok: false, + reason: `archive remains at ${archivePath} because the source path could not be checked: ${formatArchiveError(err)}`, + }; + } + try { + await fs.rename(archivePath, sourcePath); + } catch (err) { + if ((err as NodeJS.ErrnoException).code === "EXDEV") { + const outcome = await copyLegacyCronFileAcrossDevices( + archivePath, + sourcePath, + expectedSha256, + false, + ); + return outcome.ok ? { ok: true } : outcome; + } + return { + ok: false, + reason: `archive remains at ${archivePath} because restoration failed: ${formatArchiveError(err)}`, + }; + } + try { + await syncArchiveDirectory(path.dirname(sourcePath)); + return { ok: true }; + } catch (err) { + return { + ok: false, + reason: `the source was restored, but rollback directory sync failed: ${formatArchiveError(err)}`, + }; + } +} + +async function copyLegacyCronFileAcrossDevices( + filePath: string, + initialArchivePath: string, + expectedSha256?: string, + useNumberedArchive = true, +): Promise { + let archivePath = initialArchivePath; + let archiveCreated = false; + let sourceRemoved = false; + try { + const sourceStat = await fs.stat(filePath); + if (!sourceStat.isFile()) { + throw new Error("legacy cron source is not a regular file"); + } + if (expectedSha256 && (await sha256File(filePath)) !== expectedSha256) { + throw new Error("legacy cron source changed after it was imported; refusing to archive it"); + } + const sourceMode = sourceStat.mode & 0o777; + for (let index = 2; ; index += 1) { + try { + const archiveHandle = await fs.open(archivePath, "wx", sourceMode | 0o600); + archiveCreated = true; + await archiveHandle.close(); + break; + } catch (err) { + if ((err as NodeJS.ErrnoException).code !== "EEXIST" || !useNumberedArchive) { + throw err; + } + archivePath = `${filePath}${LEGACY_CRON_ARCHIVE_SUFFIX}.${index}`; + } + } + + // Claim the destination before copyFile so any partial output from a failed, + // non-atomic copy is owned by this attempt and removed by the catch path. + await fs.copyFile(filePath, archivePath); + if (expectedSha256 && (await sha256File(archivePath)) !== expectedSha256) { + throw new Error("copied legacy cron archive does not match the imported source"); + } + await fs.chmod(archivePath, sourceMode | 0o600); + const archiveHandle = await fs.open(archivePath, "r+"); + try { + await archiveHandle.chmod(sourceMode); + await archiveHandle.utimes(sourceStat.atime, sourceStat.mtime); + await archiveHandle.sync(); + } finally { + await archiveHandle.close(); + } + await syncArchiveDirectory(path.dirname(archivePath)); + const currentSourceStat = await fs.stat(filePath); + if ( + currentSourceStat.dev !== sourceStat.dev || + currentSourceStat.ino !== sourceStat.ino || + (expectedSha256 && (await sha256File(filePath)) !== expectedSha256) + ) { + throw new Error("legacy cron source changed during archival; refusing to remove it"); + } + // Current OpenClaw runtime never writes legacy JSON. POSIX has no conditional + // unlink, so hashes close observed external edits before migration-owned removal. + await fs.unlink(filePath); + sourceRemoved = true; + await syncArchiveDirectory(path.dirname(filePath)); + return { ok: true, archivePath }; + } catch (err) { + if (sourceRemoved) { + return { + ok: false, + reason: `${formatArchiveError(err)}; the durable archive is preserved at ${archivePath} because the source was already removed`, + }; + } + const cleanupFailures: string[] = []; + if (archiveCreated) { + let archiveRemoved = false; + try { + try { + await fs.unlink(archivePath); + } catch (cleanupErr) { + if ((cleanupErr as NodeJS.ErrnoException).code !== "ENOENT") { + throw cleanupErr; + } + } + archiveRemoved = true; + await syncArchiveDirectory(path.dirname(archivePath)); + } catch (cleanupErr) { + cleanupFailures.push( + archiveRemoved + ? `the partial archive was removed, but cleanup directory sync failed: ${formatArchiveError(cleanupErr)}` + : `partial archive remains at ${archivePath} because cleanup failed: ${formatArchiveError(cleanupErr)}`, + ); + } + } + const cleanupReason = cleanupFailures.length > 0 ? `; ${cleanupFailures.join("; ")}` : ""; + return { ok: false, reason: `${formatArchiveError(err)}${cleanupReason}` }; + } +} + +async function archiveLegacyCronFile( + filePath: string, + expectedSha256?: string, +): Promise { let archivePath = `${filePath}${LEGACY_CRON_ARCHIVE_SUFFIX}`; - for (let index = 2; await legacyCronFileExists(archivePath); index += 1) { - archivePath = `${filePath}${LEGACY_CRON_ARCHIVE_SUFFIX}.${index}`; + try { + if (!(await legacyCronFileExists(filePath))) { + return { ok: true }; + } + for (let index = 2; await legacyCronFileExists(archivePath); index += 1) { + archivePath = `${filePath}${LEGACY_CRON_ARCHIVE_SUFFIX}.${index}`; + } + } catch (err) { + return { ok: false, reason: formatArchiveError(err) }; + } + + try { + await fs.rename(filePath, archivePath); + } catch (err) { + // A cross-device rename can occur when the configured store is a mounted file. + // Fsync before source removal and roll back failed cleanup so retries stay idempotent. + if ((err as { code?: unknown })?.code !== "EXDEV") { + return { ok: false, reason: formatArchiveError(err) }; + } + return await copyLegacyCronFileAcrossDevices(filePath, archivePath, expectedSha256); + } + + try { + if (expectedSha256 && (await sha256File(archivePath)) !== expectedSha256) { + throw new Error("legacy cron source changed after it was imported; refusing to archive it"); + } + await syncArchiveDirectory(path.dirname(filePath)); + if (await legacyCronFileExists(filePath)) { + return { + ok: false, + reason: `the imported source was archived, but a new legacy cron source now exists at ${filePath}`, + }; + } + return { ok: true, archivePath }; + } catch (err) { + const restoreFailure = await restoreArchivedSource(archivePath, filePath, expectedSha256); + return { + ok: false, + reason: restoreFailure.ok + ? formatArchiveError(err) + : `${formatArchiveError(err)}; ${restoreFailure.reason}`, + }; } - await fs.rename(filePath, archivePath).catch(() => undefined); } function parseCronStateFile(raw: string): { @@ -144,22 +434,23 @@ function cloneConfigJobs(jobs: Array>): Array structuredClone(job)); } -async function loadStateFile( - statePath: string, -): Promise<{ version: 1; jobs: Record } | null> { +async function loadStateFile(statePath: string): Promise<{ + state: { version: 1; jobs: Record } | null; + raw?: string; +}> { let raw: string; try { raw = await fs.readFile(statePath, "utf-8"); } catch (err) { if ((err as { code?: unknown })?.code === "ENOENT") { - return null; + return { state: null }; } throw new Error(`Failed to read cron state at ${statePath}: ${String(err)}`, { cause: err, }); } - return parseCronStateFile(raw); + return { state: parseCronStateFile(raw), raw }; } function hasInlineState(jobs: Array | null | undefined>): boolean { @@ -223,17 +514,77 @@ export async function legacyCronStoreFilesExist(storePath: string): Promise { +export type LegacyCronArchiveResult = + | { ok: true } + | { ok: false; failures: Array<{ path: string; reason: string }> }; + +/** Archive legacy cron JSON/state files after successful migration. */ +export async function archiveLegacyCronStoreForMigration( + storePath: string, + source?: LegacyCronMigrationSource, +): Promise { const resolvedStorePath = path.resolve(storePath); - await Promise.all([ - archiveLegacyCronFile(resolvedStorePath), - archiveLegacyCronFile(resolveLegacyCronStatePath(resolvedStorePath)), - ]); + const statePath = resolveLegacyCronStatePath(resolvedStorePath); + const failures: Array<{ path: string; reason: string }> = []; + const archived: Array<{ path: string; archivePath: string; sha256?: string }> = []; + const rollbackArchived = async (): Promise => { + for (const target of archived.toReversed()) { + const outcome = await restoreArchivedSource(target.archivePath, target.path, target.sha256); + if (!outcome.ok) { + failures.push({ path: target.path, reason: `archive rollback failed: ${outcome.reason}` }); + } + } + }; + const unexpectedStateReason = async (): Promise => { + try { + return (await legacyCronFileExists(statePath)) + ? "legacy cron state appeared after the store was imported; refusing to archive it" + : undefined; + } catch (err) { + return `legacy cron state path could not be checked: ${formatArchiveError(err)}`; + } + }; + + if (source && !source.stateSha256) { + const reason = await unexpectedStateReason(); + if (reason) { + return { ok: false, failures: [{ path: statePath, reason }] }; + } + } + + // State is archived first so the primary JSON source remains retryable until + // every byte already persisted in SQLite has a durable archive. + const targets: Array<{ path: string; sha256?: string }> = source + ? [ + ...(source.stateSha256 ? [{ path: statePath, sha256: source.stateSha256 }] : []), + { path: resolvedStorePath, sha256: source.sourceSha256 }, + ] + : [{ path: statePath }, { path: resolvedStorePath }]; + for (const target of targets) { + const outcome = await archiveLegacyCronFile(target.path, target.sha256); + if (!outcome.ok) { + failures.push({ path: target.path, reason: outcome.reason }); + await rollbackArchived(); + break; + } + if (outcome.archivePath) { + archived.push({ ...target, archivePath: outcome.archivePath }); + } + } + if (failures.length === 0 && source) { + const reason = await unexpectedStateReason(); + if (reason) { + failures.push({ path: statePath, reason }); + await rollbackArchived(); + } + } + return failures.length === 0 ? { ok: true } : { ok: false, failures }; } /** Load legacy cron JSON/state files into the current loaded-store shape for migration. */ -export async function loadLegacyCronStoreForMigration(storePath: string): Promise { +export async function loadLegacyCronStoreForMigration( + storePath: string, +): Promise { const resolvedStorePath = path.resolve(storePath); try { const raw = await fs.readFile(resolvedStorePath, "utf-8"); @@ -252,6 +603,9 @@ export async function loadLegacyCronStoreForMigration(storePath: string): Promis const invalidConfigRows: QuarantinedCronConfigJob[] = []; for (const [index, row] of rawJobs.entries()) { if (isRecord(row)) { + // The source position distinguishes identical id-less rows, while the raw digest + // prevents an edited retry from being mistaken for the row previously imported. + markLegacyCronMigrationIdentity(row, index); configJobIndexes.push(index); configRows.push(row); } else { @@ -269,7 +623,9 @@ export async function loadLegacyCronStoreForMigration(storePath: string): Promis const jobs = store.jobs as unknown as Array>; const configJobs = cloneConfigJobs(configRows); - const stateFile = await loadStateFile(resolveLegacyCronStatePath(resolvedStorePath)); + const statePath = resolveLegacyCronStatePath(resolvedStorePath); + const loadedStateFile = await loadStateFile(statePath); + const stateFile = loadedStateFile.state; const hasLegacyInlineState = !stateFile && hasInlineState(jobs); if (stateFile) { @@ -293,7 +649,20 @@ export async function loadLegacyCronStoreForMigration(storePath: string): Promis ensureJobStateObject(job); } - return { store, configJobs, configJobIndexes, configJobRuntimeEntries, invalidConfigRows }; + return { + store, + configJobs, + configJobIndexes, + configJobRuntimeEntries, + invalidConfigRows, + migrationSource: createLegacyCronMigrationSource({ + sourcePath: resolvedStorePath, + raw, + statePath, + stateRaw: loadedStateFile.raw, + recordCount: rawJobs.length, + }), + }; } catch (err) { if ((err as { code?: unknown })?.code === "ENOENT") { return { diff --git a/src/commands/doctor/cron/migration-ledger.ts b/src/commands/doctor/cron/migration-ledger.ts new file mode 100644 index 00000000000..4734b62f0d3 --- /dev/null +++ b/src/commands/doctor/cron/migration-ledger.ts @@ -0,0 +1,115 @@ +// Durable receipts make legacy cron migration retries independent of mutable runtime rows. +import type { DatabaseSync } from "node:sqlite"; +import { + executeSqliteQuerySync, + executeSqliteQueryTakeFirstSync, + getNodeSqliteKysely, +} from "../../../infra/kysely-sync.js"; +import type { DB as OpenClawStateDatabase } from "../../../state/openclaw-state-db.generated.js"; +import { + openOpenClawStateDatabase, + runOpenClawStateWriteTransaction, +} from "../../../state/openclaw-state-db.js"; +import type { LegacyCronMigrationSource } from "./legacy-store-migration.js"; + +type CronMigrationDatabase = Pick; + +function migrationRunId(source: LegacyCronMigrationSource): string { + return `cron-legacy:${source.sourceKey}`; +} + +function hasLegacyCronMigrationReceiptInDatabase( + db: DatabaseSync, + source: LegacyCronMigrationSource, +): boolean { + const row = executeSqliteQueryTakeFirstSync( + db, + getNodeSqliteKysely(db) + .selectFrom("migration_sources") + .select("status") + .where("source_key", "=", source.sourceKey), + ); + return row?.status === "completed"; +} + +export function hasLegacyCronMigrationReceipt(source: LegacyCronMigrationSource): boolean { + return hasLegacyCronMigrationReceiptInDatabase(openOpenClawStateDatabase().db, source); +} + +export function acquireLegacyCronMigrationReceipt( + db: DatabaseSync, + source: LegacyCronMigrationSource, +): boolean { + if (hasLegacyCronMigrationReceiptInDatabase(db, source)) { + return false; + } + const now = Date.now(); + const runId = migrationRunId(source); + const reportJson = JSON.stringify({ + source: "legacy-cron-json", + target: "cron_jobs", + statePath: source.stateSha256 ? source.statePath : undefined, + stateSha256: source.stateSha256, + }); + const kysely = getNodeSqliteKysely(db); + executeSqliteQuerySync( + db, + kysely + .insertInto("migration_runs") + .values({ + id: runId, + started_at: now, + finished_at: now, + status: "completed", + report_json: reportJson, + }) + .onConflict((conflict) => + conflict.column("id").doUpdateSet({ + finished_at: now, + status: "completed", + report_json: reportJson, + }), + ), + ); + executeSqliteQuerySync( + db, + kysely + .insertInto("migration_sources") + .values({ + source_key: source.sourceKey, + migration_kind: "legacy-cron-json", + source_path: source.sourcePath, + target_table: "cron_jobs", + source_sha256: source.sourceSha256, + source_size_bytes: source.sourceSizeBytes, + source_record_count: source.sourceRecordCount, + last_run_id: runId, + status: "completed", + imported_at: now, + removed_source: 0, + report_json: reportJson, + }) + .onConflict((conflict) => + conflict.column("source_key").doUpdateSet({ + last_run_id: runId, + status: "completed", + imported_at: now, + removed_source: 0, + report_json: reportJson, + }), + ), + ); + return true; +} + +export function markLegacyCronMigrationSourceRemoved(source: LegacyCronMigrationSource): void { + runOpenClawStateWriteTransaction(({ db }) => { + executeSqliteQuerySync( + db, + getNodeSqliteKysely(db) + .updateTable("migration_sources") + .set({ removed_source: 1 }) + .where("source_key", "=", source.sourceKey), + ); + }); +} diff --git a/src/commands/doctor/cron/repair-plan.ts b/src/commands/doctor/cron/repair-plan.ts index 75063637782..8ff7cd404e1 100644 --- a/src/commands/doctor/cron/repair-plan.ts +++ b/src/commands/doctor/cron/repair-plan.ts @@ -1,8 +1,9 @@ // Cron doctor repair planning helpers for previewing and merging legacy rows. import { isDeepStrictEqual } from "node:util"; -import { normalizeOptionalString } from "../../../../packages/normalization-core/src/string-coerce.js"; +import { normalizeOptionalStringifiedId } from "../../../../packages/normalization-core/src/string-coerce.js"; import { normalizeCronJobInput } from "../../../cron/normalize.js"; import type { CronJob } from "../../../cron/types.js"; +import { resolveLegacyCronMigrationId } from "./legacy-store-migration.js"; type CronLegacyIssueCounts = Partial>; @@ -119,7 +120,11 @@ export function formatLegacyIssuePreview(issues: CronLegacyIssueCounts): string[ } function cronJobMigrationKey(job: Record): string | undefined { - return normalizeOptionalString(job.id) ?? normalizeOptionalString(job.jobId); + return ( + normalizeOptionalStringifiedId(job.id) ?? + normalizeOptionalStringifiedId(job.jobId) ?? + resolveLegacyCronMigrationId(job) + ); } /** Merge legacy JSON jobs into current jobs without duplicating matching ids/jobIds. */ diff --git a/src/commands/doctor/cron/store-migration.ts b/src/commands/doctor/cron/store-migration.ts index a136d75a574..f6c52ff1ee1 100644 --- a/src/commands/doctor/cron/store-migration.ts +++ b/src/commands/doctor/cron/store-migration.ts @@ -13,6 +13,7 @@ import { coerceFiniteScheduleNumber } from "../../../cron/schedule.js"; import { inferCronJobName } from "../../../cron/service/normalize.js"; import { normalizeCronStaggerMs, resolveDefaultCronStaggerMs } from "../../../cron/stagger.js"; import { normalizeLegacyDeliveryInput } from "./legacy-delivery.js"; +import { resolveLegacyCronMigrationId } from "./legacy-store-migration.js"; import { classifyUnresolvedAgentTurnShellToolPrompt, hasLegacyOpenAICodexCronModelRef, @@ -62,7 +63,8 @@ function normalizeStoredCronJobIdentity(raw: Record): { const hadJobIdKey = "jobId" in raw; const id = normalizeOptionalStringifiedId(raw.id); const legacyJobId = normalizeOptionalStringifiedId(raw.jobId); - const canonicalId = id ?? legacyJobId ?? `cron-${randomUUID()}`; + const canonicalId = + id ?? legacyJobId ?? resolveLegacyCronMigrationId(raw) ?? `cron-${randomUUID()}`; const nonStringIdIssue = hadIdKey && raw.id != null && typeof raw.id !== "string"; const missingIdIssue = !id && !legacyJobId; let mutated = false; @@ -267,12 +269,6 @@ export function normalizeStoredCronJobs( incrementIssue(issues, key); }; - const state = raw.state; - if (!state || typeof state !== "object" || Array.isArray(state)) { - raw.state = {}; - mutated = true; - } - const idNorm = normalizeStoredCronJobIdentity(raw); if (idNorm.mutated) { mutated = true; @@ -287,6 +283,12 @@ export function normalizeStoredCronJobs( trackIssue("nonStringId"); } + const state = raw.state; + if (!state || typeof state !== "object" || Array.isArray(state)) { + raw.state = {}; + mutated = true; + } + if (typeof raw.schedule === "string") { const expr = raw.schedule.trim(); raw.schedule = { kind: "cron", expr }; diff --git a/src/cron/store.ts b/src/cron/store.ts index 8f37a3dc147..a00860a7896 100644 --- a/src/cron/store.ts +++ b/src/cron/store.ts @@ -1,6 +1,7 @@ /** Public cron store load/save API backed by SQLite plus quarantine sidecars. */ import fs from "node:fs"; import path from "node:path"; +import type { DatabaseSync } from "node:sqlite"; import { isRecord } from "@openclaw/normalization-core/record-coerce"; import { normalizeOptionalString } from "@openclaw/normalization-core/string-coerce"; import { expandHomePrefix } from "../infra/home-dir.js"; @@ -133,6 +134,24 @@ export async function saveCronJobsStore( }); } +/** Atomically acquire doctor migration metadata and replace cron rows only for the winner. */ +export async function saveCronJobsStoreWithMetadata( + storePath: string, + store: CronStoreFile, + acquireMetadata: (db: DatabaseSync) => boolean, +): Promise { + const resolvedStorePath = path.resolve(storePath); + const storeKey = cronStoreKey(resolvedStorePath); + assertCronStoreCanPersist(store); + return runOpenClawStateWriteTransaction(({ db }) => { + if (!acquireMetadata(db)) { + return false; + } + replaceCronRows(db, storeKey, store); + return true; + }); +} + // Public plugin SDK seam; core callers use the SQLite-backed cron-jobs names above. /** Resolves the public plugin-SDK cron store path. */ export function resolveCronStorePath(storePath?: string) {