diff --git a/src/cron/service/ops.test.ts b/src/cron/service/ops.test.ts index ab3770ebc744..a4b293fca0e1 100644 --- a/src/cron/service/ops.test.ts +++ b/src/cron/service/ops.test.ts @@ -1060,20 +1060,21 @@ describe("cron service ops seam coverage", () => { job.payload = { kind: "script", script: "return { state: { cursor: 'payload' } }" }; job.state.triggerState = { cursor: "old" }; await writeCronStoreSnapshot({ storePath, jobs: [job] }); - const preparedReceipt = runReceiptStore.prepareCronRunReceiptClaim({ - storePath, - job, - agentId: "main", - startedAtMs: startedAt, - }); - const receipt = runOpenClawStateWriteTransaction(({ db }) => - runReceiptStore.claimCronRunReceiptInDatabase({ - database: db, - prepared: preparedReceipt, - resolveAgentId: (current) => current.agentId ?? "main", - }), - ); - runReceiptStore.releaseLocalCronRunReceiptOwnership(receipt); + const receipt = + reservationOffsetMs === undefined + ? runOpenClawStateWriteTransaction(({ db }) => + runReceiptStore.claimCronRunReceiptInDatabase({ + database: db, + prepared: runReceiptStore.prepareCronRunReceiptClaim({ + storePath, + job, + agentId: "main", + startedAtMs: startedAt, + }), + resolveAgentId: (current) => current.agentId ?? "main", + }), + ) + : undefined; const events: CronEvent[] = []; const state = createCronServiceState({ storePath, @@ -1085,23 +1086,27 @@ describe("cron service ops seam coverage", () => { runIsolatedAgentJob: vi.fn(async () => ({ status: "ok" as const })), onEvent: (event) => events.push(structuredClone(event)), }); - const taskRunId = - reservationOffsetMs === undefined - ? taskRuns.tryCreateCronTaskRun({ state, job, startedAt }) - : taskExecutor.createRunningTaskRunCore({ - runtime: "cron", - sourceId: job.id, - ownerKey: "", - scopeKind: "system", - runId: `${createCronExecutionId(job.id, startedAt - reservationOffsetMs)}:legacy-upgrade`, - agentId: "main", - task: job.name, - deliveryStatus: "not_applicable", - notifyPolicy: "silent", - startedAt, - lastEventAt: startedAt, - detail: { storeKey: cronStoreKey(storePath) }, - })?.runId; + const taskRunId = receipt + ? taskRuns.tryCreateCronTaskRun({ + state, + job, + startedAt, + publicRunId: receipt.receiptId, + }) + : taskExecutor.createRunningTaskRunCore({ + runtime: "cron", + sourceId: job.id, + ownerKey: "", + scopeKind: "system", + runId: `${createCronExecutionId(job.id, startedAt - (reservationOffsetMs ?? 0))}:legacy-upgrade`, + agentId: "main", + task: job.name, + deliveryStatus: "not_applicable", + notifyPolicy: "silent", + startedAt, + lastEventAt: startedAt, + detail: { storeKey: cronStoreKey(storePath) }, + })?.runId; if (!taskRunId) { throw new Error("expected reserved cron task run"); } @@ -1125,6 +1130,9 @@ describe("cron service ops seam coverage", () => { triggerFired: true, }, }); + if (receipt) { + runReceiptStore.releaseLocalCronRunReceiptOwnership(receipt); + } await start(state); @@ -1159,14 +1167,16 @@ describe("cron service ops seam coverage", () => { expect(persisted.jobs[0]?.state.runningAtMs).toBeUndefined(); expect(persisted.jobs[0]?.state.lastError).toBeUndefined(); expect(persisted.jobs[0]?.state.nextRunAtMs).toBeUndefined(); - const receiptRow = runOpenClawStateWriteTransaction(({ db }) => - db - .prepare( - "SELECT status, finished_at_ms AS finishedAtMs, error_text AS error FROM cron_run_receipts WHERE receipt_id = ?", - ) - .get(receipt.receiptId), - ) as { status: string; finishedAtMs: number; error: string | null }; - expect(receiptRow).toEqual({ status: "ok", finishedAtMs: endedAt, error: null }); + if (receipt) { + const receiptRow = runOpenClawStateWriteTransaction(({ db }) => + db + .prepare( + "SELECT status, finished_at_ms AS finishedAtMs, error_text AS error FROM cron_run_receipts WHERE receipt_id = ?", + ) + .get(receipt.receiptId), + ) as { status: string; finishedAtMs: number; error: string | null }; + expect(receiptRow).toEqual({ status: "ok", finishedAtMs: endedAt, error: null }); + } expect(events.filter((event) => event.action === "finished")).toEqual([]); stop(state); }); @@ -1195,7 +1205,6 @@ describe("cron service ops seam coverage", () => { resolveAgentId: (current) => current.agentId ?? "main", }), ); - runReceiptStore.releaseLocalCronRunReceiptOwnership(receipt); const state = createCronServiceState({ storePath, cronEnabled: true, @@ -1205,7 +1214,12 @@ describe("cron service ops seam coverage", () => { requestHeartbeat: vi.fn(), runIsolatedAgentJob: vi.fn(async () => ({ status: "ok" as const })), }); - const taskRunId = taskRuns.tryCreateCronTaskRun({ state, job, startedAt }); + const taskRunId = taskRuns.tryCreateCronTaskRun({ + state, + job, + startedAt, + publicRunId: receipt.receiptId, + }); if (!taskRunId) { throw new Error("expected invalid finalized cron task run"); } @@ -1225,6 +1239,7 @@ describe("cron service ops seam coverage", () => { runOpenClawStateWriteTransaction(({ db }) => { db.prepare("UPDATE task_runs SET ended_at = -1 WHERE run_id = ?").run(taskRunId); }); + runReceiptStore.releaseLocalCronRunReceiptOwnership(receipt); await start(state); @@ -1573,13 +1588,16 @@ describe("cron service ops seam coverage", () => { ran: true, }); - expectTaskRun({ - runId: `cron:isolated-timeout:${now}:${manualRunId}`, + const task = findCronTaskByBaseRunId(`cron:isolated-timeout:${now}`); + expect(task).toMatchObject({ runtime: "cron", status: "succeeded", sourceId: "isolated-timeout", progressSummary: "Running automation.", }); + expect(task?.runId).toMatch( + new RegExp(`^cron:isolated-timeout:${now}:[^:]+:${manualRunId}$`), + ); expect(findTaskByRunId(manualRunId)).toBeUndefined(); }); }); @@ -1753,6 +1771,56 @@ describe("cron service ops seam coverage", () => { createTaskRecordSpy.mockRestore(); }); + it("keeps terminal fallback bound to its receipt before a same-time successor", async () => { + const { storePath } = await makeStorePath(); + const now = Date.parse("2026-03-23T12:00:00.000Z"); + + await withStateDirForStorePath(storePath, async () => { + await writeDueIsolatedJobSnapshot(storePath, now); + const state = createOkIsolatedCronState({ storePath, now, summary: "done" }); + const createTaskRecord = taskExecutor.createRunningTaskRunCore; + const createTaskRecordSpy = vi + .spyOn(taskExecutor, "createRunningTaskRunCore") + .mockImplementationOnce(() => { + throw new Error("transient ledger failure"); + }) + .mockImplementation((params) => createTaskRecord(params)); + + try { + await expect(run(state, "isolated-timeout", "force")).resolves.toEqual({ + ok: true, + ran: true, + }); + await expect(run(state, "isolated-timeout", "force")).resolves.toEqual({ + ok: true, + ran: true, + }); + + const receiptIds = openOpenClawStateDatabase() + .db.prepare( + `SELECT receipt_id AS receiptId + FROM cron_run_receipts + WHERE store_key = ? AND job_id = ? + ORDER BY receipt_id`, + ) + .all(cronStoreKey(storePath), "isolated-timeout") + .map((row) => (row as { receiptId: string }).receiptId); + const taskRunIds = listTaskRecordsUnsorted() + .filter((task) => task.runtime === "cron" && task.sourceId === "isolated-timeout") + .map((task) => task.runId); + + expect(receiptIds).toHaveLength(2); + expect(taskRunIds).toHaveLength(2); + for (const receiptId of receiptIds) { + expect(taskRunIds).toContain(`cron:isolated-timeout:${now}:${receiptId}`); + } + } finally { + createTaskRecordSpy.mockRestore(); + stop(state); + } + }); + }); + it("keeps manual cron cleanup progressing when task ledger updates fail", async () => { const { storePath } = await makeStorePath(); const now = Date.parse("2026-03-23T12:00:00.000Z"); diff --git a/src/cron/service/run-recovery.test.ts b/src/cron/service/run-recovery.test.ts index b59614bc516c..39b48359237d 100644 --- a/src/cron/service/run-recovery.test.ts +++ b/src/cron/service/run-recovery.test.ts @@ -1,7 +1,10 @@ import { describe, expect, it, vi } from "vitest"; import { runOpenClawStateWriteTransaction } from "../../state/openclaw-state-db.js"; +import { createRunningTaskRunCore } from "../../tasks/task-executor.js"; +import { createCronExecutionId } from "../run-id.js"; import { setupCronServiceSuite, writeCronStoreSnapshot } from "../service.test-harness.js"; import { loadCronStore } from "../store.js"; +import { cronStoreKey } from "../store/key.js"; import { claimCronRunReceiptInDatabase, finishCronRunReceipt, @@ -15,7 +18,11 @@ import type { CronJob } from "../types.js"; import { proposeCronRunRecovery, recoverCronRunProposal } from "./run-recovery.js"; import { createCronServiceState } from "./state.js"; import { runPostPersistCronNotifications } from "./store.js"; -import { tryCreateCronTaskRun, tryFinishCronTaskRunWithoutHistory } from "./task-runs.js"; +import { + tryCreateCronTaskRun, + tryFinishCronTaskRun, + tryFinishCronTaskRunWithoutHistory, +} from "./task-runs.js"; const { logger, makeStorePath } = setupCronServiceSuite({ prefix: "cron-run-recovery-" }); @@ -71,6 +78,32 @@ function claimReceipt(storePath: string, job: CronJob, startedAtMs: number) { ); } +function createLegacyCronTaskRun(params: { + storePath: string; + job: CronJob; + startedAtMs: number; + runId: string; +}): string { + const task = createRunningTaskRunCore({ + runtime: "cron", + sourceId: params.job.id, + ownerKey: "", + scopeKind: "system", + runId: params.runId, + agentId: params.job.agentId, + task: params.job.name, + deliveryStatus: "not_applicable", + notifyPolicy: "silent", + startedAt: params.startedAtMs, + lastEventAt: params.startedAtMs, + detail: { storeKey: cronStoreKey(params.storePath) }, + }); + if (!task) { + throw new Error("expected legacy cron task run"); + } + return params.runId; +} + async function commitCompletedJob(params: { storePath: string; jobs: CronJob[]; @@ -243,20 +276,32 @@ describe("atomic cron run recovery", () => { expect(sendCronFailureAlert).not.toHaveBeenCalled(); }); - it("restores a finalized quiet trigger with a skipped receipt", async () => { + it("restores a pre-upgrade receipt-keyed task over a prior legacy manual task", async () => { const { storePath } = await makeStorePath(); const startedAtMs = Date.parse("2026-08-13T10:45:00.000Z"); const job = makeJob("quiet-trigger-recovery", startedAtMs); job.trigger = { script: "return false" }; await writeCronStoreSnapshot({ storePath, jobs: [job] }); const state = makeState(storePath, startedAtMs + 30_000); + const priorReceipt = claimReceipt(storePath, job, startedAtMs); + createLegacyCronTaskRun({ + storePath, + job, + startedAtMs, + runId: `${createCronExecutionId(job.id, startedAtMs)}:prior-manual-run`, + }); + finishCronRunReceipt({ + handle: priorReceipt, + status: "interrupted", + finishedAtMs: startedAtMs + 1, + }); const receipt = claimReceipt(storePath, job, startedAtMs); const proposal = proposeCronRunRecovery(state, job.id, undefined, startedAtMs); - const taskRunId = tryCreateCronTaskRun({ - state, + const taskRunId = createLegacyCronTaskRun({ + storePath, job, - startedAt: startedAtMs, - publicRunId: receipt.receiptId, + startedAtMs, + runId: `${createCronExecutionId(job.id, startedAtMs)}:${receipt.receiptId}`, }); tryFinishCronTaskRunWithoutHistory(state, { taskRunId, @@ -277,6 +322,165 @@ describe("atomic cron run recovery", () => { .get(receipt.receiptId), ) as { status: string }; expect(receiptRow.status).toBe("skipped"); + const requestRow = runOpenClawStateWriteTransaction(({ db }) => + db + .prepare( + "SELECT request_run_id AS requestRunId FROM cron_run_receipts WHERE receipt_id = ?", + ) + .get(receipt.receiptId), + ) as { requestRunId: string | null }; + expect(requestRow.requestRunId).toBeNull(); + }); + + it("does not restore a prior same-millisecond task for a different receipt", async () => { + const { storePath } = await makeStorePath(); + const startedAtMs = Date.parse("2026-08-13T10:50:00.000Z"); + const job = makeJob("same-millisecond-task-recovery", startedAtMs); + await writeCronStoreSnapshot({ storePath, jobs: [job] }); + const state = makeState(storePath, startedAtMs + 30_000); + const priorReceipt = claimReceipt(storePath, job, startedAtMs); + const priorTaskRunId = tryCreateCronTaskRun({ + state, + job, + startedAt: startedAtMs, + publicRunId: priorReceipt.receiptId, + }); + if (!priorTaskRunId) { + throw new Error("expected prior cron task run"); + } + tryFinishCronTaskRun(state, { + taskRunId: priorTaskRunId, + job, + event: { + jobId: job.id, + action: "finished", + job, + status: "ok", + summary: "prior receipt completed", + runAtMs: startedAtMs, + durationMs: 1, + }, + }); + finishCronRunReceipt({ + handle: priorReceipt, + status: "ok", + finishedAtMs: startedAtMs + 1, + }); + const receipt = claimReceipt(storePath, job, startedAtMs); + const proposal = proposeCronRunRecovery(state, job.id, undefined, startedAtMs); + releaseLocalCronRunReceiptOwnership(receipt); + + expect(recoverCronRunProposal(state, proposal)).toMatchObject({ kind: "repaired" }); + expect((await loadCronStore(storePath)).jobs[0]?.state).toMatchObject({ + lastRunStatus: "error", + lastError: expect.stringContaining("interrupted by gateway restart"), + }); + const receiptRow = runOpenClawStateWriteTransaction(({ db }) => + db + .prepare("SELECT status FROM cron_run_receipts WHERE receipt_id = ?") + .get(receipt.receiptId), + ) as { status: string }; + expect(receiptRow.status).toBe("interrupted"); + }); + + it("does not restore a sole legacy caller-ID task after its prior receipt was pruned", async () => { + const { storePath } = await makeStorePath(); + const startedAtMs = Date.parse("2026-08-13T10:52:00.000Z"); + const job = makeJob("pruned-manual-task-recovery", startedAtMs); + await writeCronStoreSnapshot({ storePath, jobs: [job] }); + const state = makeState(storePath, startedAtMs + 30_000); + const priorReceipt = claimReceipt(storePath, job, startedAtMs); + const priorTaskRunId = createLegacyCronTaskRun({ + storePath, + job, + startedAtMs, + runId: `${createCronExecutionId(job.id, startedAtMs)}:prior-manual-run`, + }); + tryFinishCronTaskRun(state, { + taskRunId: priorTaskRunId, + job, + event: { + jobId: job.id, + action: "finished", + job, + status: "ok", + runId: "prior-manual-run", + runAtMs: startedAtMs, + durationMs: 1, + }, + }); + finishCronRunReceipt({ + handle: priorReceipt, + status: "ok", + finishedAtMs: startedAtMs + 1, + }); + // Receipt retention is shorter than task history retention. + runOpenClawStateWriteTransaction(({ db }) => { + db.prepare("DELETE FROM cron_run_receipts WHERE receipt_id = ?").run(priorReceipt.receiptId); + }); + const receipt = claimReceipt(storePath, job, startedAtMs); + const proposal = proposeCronRunRecovery(state, job.id, undefined, startedAtMs); + // This final shape (one active receipt plus one same-ms legacy task) is + // indistinguishable from a pre-upgrade caller-ID task for the active run. + releaseLocalCronRunReceiptOwnership(receipt); + + expect(recoverCronRunProposal(state, proposal)).toMatchObject({ kind: "repaired" }); + expect((await loadCronStore(storePath)).jobs[0]?.state).toMatchObject({ + lastRunStatus: "error", + lastError: expect.stringContaining("interrupted by gateway restart"), + }); + const receiptRow = runOpenClawStateWriteTransaction(({ db }) => + db + .prepare("SELECT status FROM cron_run_receipts WHERE receipt_id = ?") + .get(receipt.receiptId), + ) as { status: string }; + expect(receiptRow.status).toBe("interrupted"); + }); + + it.each([ + { + identity: "pre-upgrade caller-supplied manual", + reservationOffsetMs: 0, + discriminator: "legacy-manual", + }, + { identity: "reservation-keyed", reservationOffsetMs: 250, discriminator: "legacy-upgrade" }, + ])("fails closed for a $identity task without exact receipt identity", async (testCase) => { + const { storePath } = await makeStorePath(); + const startedAtMs = Date.parse("2026-08-13T10:54:00.000Z"); + const job = makeJob("reservation-task-recovery", startedAtMs); + await writeCronStoreSnapshot({ storePath, jobs: [job] }); + const state = makeState(storePath, startedAtMs + 30_000); + const receipt = claimReceipt(storePath, job, startedAtMs); + const proposal = proposeCronRunRecovery(state, job.id, undefined, startedAtMs); + const taskRunId = `${createCronExecutionId(job.id, startedAtMs - testCase.reservationOffsetMs)}:${testCase.discriminator}`; + createLegacyCronTaskRun({ storePath, job, startedAtMs, runId: taskRunId }); + tryFinishCronTaskRun(state, { + taskRunId, + job, + event: { + jobId: job.id, + action: "finished", + job, + status: "ok", + runAtMs: startedAtMs, + durationMs: 1, + }, + }); + releaseLocalCronRunReceiptOwnership(receipt); + + expect(recoverCronRunProposal(state, proposal)).toMatchObject({ kind: "repaired" }); + expect((await loadCronStore(storePath)).jobs[0]?.state).toMatchObject({ + lastRunStatus: "error", + lastError: expect.stringContaining("interrupted by gateway restart"), + }); + const receiptRow = runOpenClawStateWriteTransaction(({ db }) => + db + .prepare( + "SELECT status, request_run_id AS requestRunId FROM cron_run_receipts WHERE receipt_id = ?", + ) + .get(receipt.receiptId), + ) as { status: string; requestRunId: string | null }; + expect(receiptRow).toEqual({ status: "interrupted", requestRunId: null }); }); it("retires a dead owner receipt after timeout state already finalized", async () => { diff --git a/src/cron/service/run-recovery.ts b/src/cron/service/run-recovery.ts index 5dc291e1bb33..23ebd0f33194 100644 --- a/src/cron/service/run-recovery.ts +++ b/src/cron/service/run-recovery.ts @@ -143,6 +143,7 @@ function repairInDatabase(params: { jobId: proposal.jobId, startedAt: proposal.runningAtMs, storeKey, + ...(proposal.receipt ? { receiptId: proposal.receipt.receiptId } : {}), }); const finalized = task.finalized; const restored = finalized diff --git a/src/cron/service/task-runs.ts b/src/cron/service/task-runs.ts index 9ff0a996c60f..a07e69b65058 100644 --- a/src/cron/service/task-runs.ts +++ b/src/cron/service/task-runs.ts @@ -20,6 +20,7 @@ import { import { createCronExecutionId } from "../run-id.js"; import type { CronRunLogEntry } from "../run-log-types.js"; import { cronStoreKey } from "../store/key.js"; +import { bindCronRunReceiptTaskIdentity } from "../store/run-receipt-store.js"; import { cronRunLogEntryToTaskDetail, cronRunStatusToTaskStatus, @@ -95,19 +96,52 @@ export function tryCreateCronTaskRun(params: { startedAt: number; publicRunId?: string; }): string | undefined { - const runId = createCronTaskRunId(params.job.id, params.startedAt, params.publicRunId); - return tryCreateCronTaskRunRecord({ + let runId: string; + try { + runId = resolveCronTaskRunId({ + state: params.state, + jobId: params.job.id, + startedAt: params.startedAt, + publicRunId: params.publicRunId, + }); + } catch (error) { + params.state.deps.log.warn( + { jobId: params.job.id, error }, + "cron: failed to bind task ledger record to its run receipt", + ); + return undefined; + } + tryCreateCronTaskRunRecord({ state: params.state, job: params.job, jobId: params.job.id, startedAt: params.startedAt, runId, }); + return runId; } -function createCronTaskRunId(jobId: string, startedAt: number, publicRunId?: string): string { - const discriminator = publicRunId?.trim() || randomUUID(); - return `${createCronExecutionId(jobId, startedAt)}:${discriminator}`; +function resolveCronTaskRunId(params: { + state: CronServiceState; + jobId: string; + startedAt: number; + publicRunId?: string; +}): string { + const publicRunId = params.publicRunId?.trim() || undefined; + const receiptId = bindCronRunReceiptTaskIdentity({ + storePath: params.state.deps.storePath, + jobId: params.jobId, + startedAtMs: params.startedAt, + requestRunId: publicRunId, + }); + const executionRunId = createCronExecutionId(params.jobId, params.startedAt); + if (receiptId) { + const receiptRunId = `${executionRunId}:${receiptId}`; + return publicRunId && publicRunId !== receiptId + ? `${receiptRunId}:${publicRunId}` + : receiptRunId; + } + return `${executionRunId}:${publicRunId ?? randomUUID()}`; } function findLatestCronTaskRunForRecoveryFromRecords( @@ -115,34 +149,51 @@ function findLatestCronTaskRunForRecoveryFromRecords( jobId: string, startedAt: number, storeKey: string, + receiptId?: string, ): TaskRecord | undefined { const executionRunId = createCronExecutionId(jobId, startedAt); const prefix = `${executionRunId}:`; + const receiptRunId = receiptId ? `${prefix}${receiptId}` : undefined; return records .filter((task) => { if (task.runtime !== "cron" || task.sourceId !== jobId) { return false; } const taskStoreKey = cronTaskRecordStoreKey(task); + if (receiptRunId) { + // Sequential receipts can share the timestamp and public id, so only + // the receipt-prefixed task identity can own receipt-based recovery. + return ( + taskStoreKey === storeKey && + (task.runId === receiptRunId || task.runId?.startsWith(`${receiptRunId}:`)) + ); + } if (taskStoreKey === undefined) { // Exact match covers detail-less pre-discriminator rows from older releases. return task.runId === executionRunId; } - return ( - taskStoreKey === storeKey && - (task.runId === executionRunId || - task.runId?.startsWith(prefix) || - // Released reservation-keyed rows still record the authoritative execution start. - task.startedAt === startedAt) - ); + if (taskStoreKey !== storeKey) { + return false; + } + if (task.runId === executionRunId) { + return true; + } + if (task.runId?.startsWith(prefix)) { + return true; + } + // Released reservation-keyed rows still record the authoritative execution start. + return task.startedAt === startedAt; }) - .toSorted( - (left, right) => - Number(left.endedAt !== undefined) - Number(right.endedAt !== undefined) || + .toSorted((left, right) => { + const finalizedOrder = + Number(right.endedAt !== undefined) - Number(left.endedAt !== undefined); + return ( + (receiptRunId ? finalizedOrder : -finalizedOrder) || resolveCronTaskRecordTimestamp(right) - resolveCronTaskRecordTimestamp(left) || right.createdAt - left.createdAt || - right.taskId.localeCompare(left.taskId), - )[0]; + right.taskId.localeCompare(left.taskId) + ); + })[0]; } type FinalizedCronTaskRun = { @@ -193,12 +244,14 @@ export function findCronTaskRunRecoveryInDatabase(params: { jobId: string; startedAt: number; storeKey: string; + receiptId?: string; }): { taskRunId?: string; finalized?: FinalizedCronTaskRun } { const task = findLatestCronTaskRunForRecoveryFromRecords( listTaskRecordsByRuntimeSourceIdInDatabase(params.database, "cron", params.jobId), params.jobId, params.startedAt, params.storeKey, + params.receiptId, ); const finalized = finalizedCronTaskRun(task, params.jobId); return { @@ -337,9 +390,14 @@ export function tryFinishCronTaskRun( result.errorClassification, ); const startedAt = entry.runAtMs ?? entry.ts; - const candidateRunId = - result.taskRunId ?? createCronTaskRunId(entry.jobId, startedAt, entry.runId); + let candidateRunId = result.taskRunId; try { + candidateRunId ??= resolveCronTaskRunId({ + state, + jobId: entry.jobId, + startedAt, + publicRunId: entry.runId, + }); const existingCandidate = findTaskByRunId(candidateRunId); const taskRunId = existingCandidate?.runtime === "cron" @@ -432,7 +490,7 @@ export function tryFinishCronTaskRun( } } catch (error) { state.deps.log.warn( - { runId: candidateRunId, jobStatus: entry.status, error }, + { runId: candidateRunId ?? entry.runId, jobStatus: entry.status, error }, "cron: failed to update task ledger record", ); } diff --git a/src/cron/store/run-receipt-store.test.ts b/src/cron/store/run-receipt-store.test.ts index 0a96e197797d..69dd2d09327b 100644 --- a/src/cron/store/run-receipt-store.test.ts +++ b/src/cron/store/run-receipt-store.test.ts @@ -9,12 +9,14 @@ import type { CronJob } from "../types.js"; import { cronStoreKey } from "./key.js"; import { assertCronRunReceiptCurrent, + bindCronRunReceiptTaskIdentity, claimCronRunReceiptInDatabase, CronRunReceiptConflictError, CronRunReceiptRevisionError, findActiveCronRunReceiptInDatabase, finishCronRunReceipt, prepareCronRunReceiptClaim, + releaseLocalCronRunReceiptOwnership, } from "./run-receipt-store.js"; const { makeStorePath } = setupCronServiceSuite({ prefix: "cron-run-receipt-" }); @@ -55,7 +57,8 @@ function receipts(storePath: string, jobId: string) { return openOpenClawStateDatabase() .db.prepare( `SELECT receipt_id AS receiptId, status, agent_id AS agentId, - started_at_ms AS startedAtMs, error_text AS error + request_run_id AS requestRunId, started_at_ms AS startedAtMs, + error_text AS error FROM cron_run_receipts WHERE store_key = ? AND job_id = ? ORDER BY started_at_ms DESC, receipt_id DESC`, @@ -64,6 +67,7 @@ function receipts(storePath: string, jobId: string) { receiptId: string; status: string; agentId: string; + requestRunId: string | null; startedAtMs: number; error: string | null; }>; @@ -96,10 +100,31 @@ describe("cron run receipt store", () => { await saveCronStore(storePath, { version: 1, jobs: [job] }); const first = claim(storePath, job, 100); + expect( + bindCronRunReceiptTaskIdentity({ + storePath, + jobId: job.id, + startedAtMs: 100, + requestRunId: "manual-overlap-1", + }), + ).toBe(first.receiptId); + expect(() => + bindCronRunReceiptTaskIdentity({ + storePath, + jobId: job.id, + startedAtMs: 100, + requestRunId: "manual-overlap-2", + }), + ).toThrow(CronRunReceiptRevisionError); expect(() => claim(storePath, job, 101)).toThrow(CronRunReceiptConflictError); expect(receipts(storePath, job.id)).toMatchObject([ - { receiptId: first.receiptId, status: "running", startedAtMs: 100 }, + { + receiptId: first.receiptId, + requestRunId: "manual-overlap-1", + status: "running", + startedAtMs: 100, + }, ]); finishCronRunReceipt({ handle: first, status: "ok", finishedAtMs: 110 }); @@ -109,6 +134,35 @@ describe("cron run receipt store", () => { expect(receipts(storePath, job.id).map((receipt) => receipt.status)).toEqual(["skipped", "ok"]); }); + it.each(["process-start identity", "process-local ownership"])( + "rejects task binding without matching %s", + async (fence) => { + const { storePath } = await makeStorePath(); + const job = makeJob(`task-bind-${fence}`); + await saveCronStore(storePath, { version: 1, jobs: [job] }); + const receipt = claim(storePath, job, 150); + if (fence === "process-start identity") { + openOpenClawStateDatabase() + .db.prepare( + "UPDATE cron_run_receipts SET owner_start_time = owner_start_time + 1 WHERE receipt_id = ?", + ) + .run(receipt.receiptId); + } else { + releaseLocalCronRunReceiptOwnership(receipt); + } + + expect(() => + bindCronRunReceiptTaskIdentity({ + storePath, + jobId: job.id, + startedAtMs: 150, + requestRunId: "manual-fenced", + }), + ).toThrow(CronRunReceiptRevisionError); + releaseLocalCronRunReceiptOwnership(receipt); + }, + ); + it("retires a provably dead process claim before admitting its successor", async () => { const { storePath } = await makeStorePath(); const job = makeJob("restart"); diff --git a/src/cron/store/run-receipt-store.ts b/src/cron/store/run-receipt-store.ts index 77a5b5e37583..1dc6ee288b04 100644 --- a/src/cron/store/run-receipt-store.ts +++ b/src/cron/store/run-receipt-store.ts @@ -559,6 +559,61 @@ export function activateCronRunReceiptInDatabase(params: { return { ...params.handle, startedAtMs: params.startedAtMs }; } +/** Binds a task/public identity to the exact locally owned active receipt. */ +export function bindCronRunReceiptTaskIdentity(params: { + storePath: string; + jobId: string; + startedAtMs: number; + requestRunId?: string; +}): string | undefined { + const storeKey = cronStoreKey(params.storePath); + const requestRunId = params.requestRunId?.trim() || undefined; + return withReceiptWrite("cron.run-receipt.bind-task", {}, (database) => { + const current = activeRow(database, storeKey, params.jobId); + if (!current) { + return undefined; + } + const ownerStartTime = getFileLockProcessStartTime(process.pid); + if ( + current.started_at_ms !== params.startedAtMs || + current.owner_pid !== process.pid || + ownerStartTime === null || + current.owner_start_time !== ownerStartTime || + !locallyOwnedReceipts.has(current.receipt_id) + ) { + throw new CronRunReceiptRevisionError( + current.receipt_id, + "cron run task identity no longer belongs to the active receipt", + ); + } + if ( + requestRunId && + current.request_run_id !== null && + current.request_run_id !== requestRunId + ) { + throw new CronRunReceiptRevisionError( + current.receipt_id, + "cron run request identity changed after admission", + ); + } + if (requestRunId && current.request_run_id === null) { + executeSqliteQuerySync( + database, + query(database) + .updateTable("cron_run_receipts") + .set({ request_run_id: requestRunId }) + .where("receipt_id", "=", current.receipt_id) + .where("status", "=", "running") + .where("owner_pid", "=", current.owner_pid) + .where("owner_start_time", "=", current.owner_start_time) + .where("started_at_ms", "=", current.started_at_ms) + .where("request_run_id", "is", null), + ); + } + return current.receipt_id; + }); +} + export function assertCronRunReceiptCurrent(params: { handle: CronRunReceiptHandle; resolveAgentId: ResolveReceiptAgentId;