Skip to content

Commit be94853

Browse files
liaoandialtaywtf
liaoandi
and
altaywtf
authored
fix(tasks): recover terminal lost cron rows (#86088)
* fix(tasks): recover terminal lost cron rows * fix(tasks): require backing-session lost recovery signal * fix(tasks): satisfy has-own lint rule --------- Co-authored-by: Altay <[email protected]>
1 parent 355c43f commit be94853

4 files changed

Lines changed: 196 additions & 7 deletions

File tree

‎src/tasks/task-registry.maintenance.issue-60299.test.ts‎

Lines changed: 135 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -152,11 +152,17 @@ function createTaskRegistryMaintenanceHarness(params: {
152152
status: patch.status,
153153
endedAt: patch.endedAt,
154154
lastEventAt: patch.lastEventAt ?? patch.endedAt,
155-
...(patch.error !== undefined ? { error: patch.error } : {}),
156155
...(patch.terminalSummary !== undefined
157156
? { terminalSummary: patch.terminalSummary ?? undefined }
158157
: {}),
159158
} satisfies TaskRecord;
159+
if (Object.hasOwn(patch, "error")) {
160+
if (patch.error === undefined) {
161+
delete next.error;
162+
} else {
163+
next.error = patch.error;
164+
}
165+
}
160166
currentTasks.set(patch.taskId, next);
161167
return next;
162168
},
@@ -532,6 +538,134 @@ describe("task-registry maintenance issue #60299", () => {
532538
expectTaskStatus(currentTasks, task.taskId, "lost");
533539
});
534540

541+
it("recovers terminal lost cron tasks from durable run logs", async () => {
542+
const startedAt = Date.now() - GRACE_EXPIRED_MS;
543+
const task = makeStaleTask({
544+
runtime: "cron",
545+
sourceId: "cron-job-terminal-lost-ok",
546+
runId: `cron:cron-job-terminal-lost-ok:${startedAt}`,
547+
status: "lost",
548+
error: "backing session missing",
549+
startedAt,
550+
endedAt: startedAt + 60_000,
551+
lastEventAt: startedAt + 60_000,
552+
cleanupAfter: Date.now() + 60_000,
553+
});
554+
555+
const { currentTasks } = createTaskRegistryMaintenanceHarness({
556+
tasks: [task],
557+
cronRunLogEntries: {
558+
"cron-job-terminal-lost-ok": [
559+
{
560+
ts: startedAt + 1250,
561+
jobId: "cron-job-terminal-lost-ok",
562+
action: "finished",
563+
status: "ok",
564+
summary: "done",
565+
runAtMs: startedAt,
566+
durationMs: 1250,
567+
},
568+
],
569+
},
570+
});
571+
572+
const reconciled = reconcileInspectableTasks();
573+
expect(reconciled).toEqual([
574+
expect.objectContaining({
575+
taskId: task.taskId,
576+
status: "succeeded",
577+
endedAt: startedAt + 1250,
578+
terminalSummary: "done",
579+
}),
580+
]);
581+
expect(reconciled[0]).not.toHaveProperty("error");
582+
expect(previewTaskRegistryMaintenance()).toMatchObject({ reconciled: 0, recovered: 1 });
583+
expect(await runTaskRegistryMaintenance()).toMatchObject({ reconciled: 0, recovered: 1 });
584+
const recoveredTask = currentTasks.get(task.taskId);
585+
expect(recoveredTask).toMatchObject({
586+
status: "succeeded",
587+
endedAt: startedAt + 1250,
588+
terminalSummary: "done",
589+
});
590+
expect(recoveredTask).not.toHaveProperty("error");
591+
});
592+
593+
it("does not recover terminal lost cron tasks without a backing-session error", async () => {
594+
const startedAt = Date.now() - GRACE_EXPIRED_MS;
595+
const task = makeStaleTask({
596+
runtime: "cron",
597+
sourceId: "cron-job-terminal-lost-no-error",
598+
runId: `cron:cron-job-terminal-lost-no-error:${startedAt}`,
599+
status: "lost",
600+
startedAt,
601+
endedAt: startedAt + 60_000,
602+
lastEventAt: startedAt + 60_000,
603+
cleanupAfter: Date.now() + 60_000,
604+
});
605+
606+
const { currentTasks } = createTaskRegistryMaintenanceHarness({
607+
tasks: [task],
608+
cronRunLogEntries: {
609+
"cron-job-terminal-lost-no-error": [
610+
{
611+
ts: startedAt + 1250,
612+
jobId: "cron-job-terminal-lost-no-error",
613+
action: "finished",
614+
status: "ok",
615+
summary: "done",
616+
runAtMs: startedAt,
617+
durationMs: 1250,
618+
},
619+
],
620+
},
621+
});
622+
623+
expect(previewTaskRegistryMaintenance()).toMatchObject({ recovered: 0 });
624+
expect(await runTaskRegistryMaintenance()).toMatchObject({ recovered: 0 });
625+
expect(currentTasks.get(task.taskId)).toMatchObject({
626+
status: "lost",
627+
});
628+
});
629+
630+
it("does not recover terminal lost cron tasks with non-backing-session errors", async () => {
631+
const startedAt = Date.now() - GRACE_EXPIRED_MS;
632+
const task = makeStaleTask({
633+
runtime: "cron",
634+
sourceId: "cron-job-terminal-lost-other-error",
635+
runId: `cron:cron-job-terminal-lost-other-error:${startedAt}`,
636+
status: "lost",
637+
error: "operator marked lost",
638+
startedAt,
639+
endedAt: startedAt + 60_000,
640+
lastEventAt: startedAt + 60_000,
641+
cleanupAfter: Date.now() + 60_000,
642+
});
643+
644+
const { currentTasks } = createTaskRegistryMaintenanceHarness({
645+
tasks: [task],
646+
cronRunLogEntries: {
647+
"cron-job-terminal-lost-other-error": [
648+
{
649+
ts: startedAt + 1250,
650+
jobId: "cron-job-terminal-lost-other-error",
651+
action: "finished",
652+
status: "ok",
653+
summary: "done",
654+
runAtMs: startedAt,
655+
durationMs: 1250,
656+
},
657+
],
658+
},
659+
});
660+
661+
expect(previewTaskRegistryMaintenance()).toMatchObject({ recovered: 0 });
662+
expect(await runTaskRegistryMaintenance()).toMatchObject({ recovered: 0 });
663+
expect(currentTasks.get(task.taskId)).toMatchObject({
664+
status: "lost",
665+
error: "operator marked lost",
666+
});
667+
});
668+
535669
it("recovers interrupted cron tasks from durable cron job state when run logs are absent", async () => {
536670
const startedAt = Date.now() - GRACE_EXPIRED_MS;
537671
const task = makeStaleTask({

‎src/tasks/task-registry.maintenance.ts‎

Lines changed: 14 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -357,6 +357,14 @@ function isTimeoutCronError(error: string | undefined): boolean {
357357
return error === "cron: job execution timed out";
358358
}
359359

360+
function isRecoverableLostCronTask(task: TaskRecord): boolean {
361+
if (task.status !== "lost") {
362+
return false;
363+
}
364+
const error = task.error?.trim().toLowerCase();
365+
return Boolean(error?.includes("backing session missing"));
366+
}
367+
360368
function mapCronTerminalStatus(status: unknown, error?: string): CronTerminalRecovery["status"] {
361369
if (status === "ok" || status === "skipped") {
362370
return "succeeded";
@@ -454,7 +462,7 @@ function resolveDurableCronTaskRecovery(
454462
task: TaskRecord,
455463
context: CronRecoveryContext,
456464
): CronTerminalRecovery | undefined {
457-
if (task.runtime !== "cron" || !isActiveTask(task)) {
465+
if (task.runtime !== "cron" || (!isActiveTask(task) && !isRecoverableLostCronTask(task))) {
458466
return undefined;
459467
}
460468
const execution = parseCronExecutionId(task);
@@ -808,7 +816,7 @@ function markTaskRecovered(task: TaskRecord, recovery: CronTerminalRecovery): Ta
808816
status: recovery.status,
809817
endedAt: recovery.endedAt,
810818
lastEventAt: recovery.lastEventAt,
811-
...(recovery.error !== undefined ? { error: recovery.error } : {}),
819+
error: recovery.error,
812820
...(recovery.terminalSummary !== undefined
813821
? { terminalSummary: recovery.terminalSummary }
814822
: {}),
@@ -823,11 +831,14 @@ function projectTaskRecovered(task: TaskRecord, recovery: CronTerminalRecovery):
823831
status: recovery.status,
824832
endedAt: recovery.endedAt,
825833
lastEventAt: recovery.lastEventAt,
826-
...(recovery.error !== undefined ? { error: recovery.error } : {}),
834+
error: recovery.error,
827835
...(recovery.terminalSummary !== undefined
828836
? { terminalSummary: recovery.terminalSummary }
829837
: {}),
830838
};
839+
if (recovery.error === undefined) {
840+
delete projected.error;
841+
}
831842
return {
832843
...projected,
833844
...(typeof projected.cleanupAfter === "number"

‎src/tasks/task-registry.test.ts‎

Lines changed: 38 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -616,6 +616,44 @@ describe("task-registry", () => {
616616
});
617617
});
618618

619+
it("clears terminal errors when explicitly updated without an error", async () => {
620+
await withTaskRegistryTempDir(async (root) => {
621+
process.env.OPENCLAW_STATE_DIR = root;
622+
resetTaskRegistryForTests();
623+
624+
const task = createTaskRecord({
625+
runtime: "cron",
626+
ownerKey: "system:cron:test",
627+
scopeKind: "system",
628+
runId: "run-terminal-error-clear",
629+
task: "Recover cron task",
630+
status: "running",
631+
deliveryStatus: "not_applicable",
632+
startedAt: 100,
633+
});
634+
635+
markTaskTerminalById({
636+
taskId: task.taskId,
637+
status: "failed",
638+
endedAt: 200,
639+
error: "backing session missing",
640+
});
641+
markTaskTerminalById({
642+
taskId: task.taskId,
643+
status: "succeeded",
644+
endedAt: 250,
645+
error: undefined,
646+
});
647+
648+
const recoveredTask = getTaskById(task.taskId);
649+
expect(recoveredTask).toMatchObject({
650+
status: "succeeded",
651+
endedAt: 250,
652+
});
653+
expect(recoveredTask).not.toHaveProperty("error");
654+
});
655+
});
656+
619657
it("keeps stronger run-scoped terminal states when a late success arrives", async () => {
620658
await withTaskRegistryTempDir(async () => {
621659
resetTaskRegistryMemoryForTest();

‎src/tasks/task-registry.ts‎

Lines changed: 9 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -1180,6 +1180,9 @@ function updateTask(taskId: string, patch: Partial<TaskRecord>): TaskRecord | nu
11801180
return null;
11811181
}
11821182
const next = normalizeTaskTimestamps({ ...current, ...patch });
1183+
if (Object.hasOwn(patch, "error") && patch.error === undefined) {
1184+
delete next.error;
1185+
}
11831186
if (isTerminalTaskStatus(next.status) && typeof next.cleanupAfter !== "number") {
11841187
next.cleanupAfter = resolveTaskCleanupAfter({
11851188
...next,
@@ -1553,11 +1556,10 @@ export function markTaskTerminalById(params: {
15531556
terminalOutcome?: TaskTerminalOutcome | null;
15541557
}): TaskRecord | null {
15551558
ensureTaskRegistryReady();
1556-
return updateTask(params.taskId, {
1559+
const patch: Partial<TaskRecord> = {
15571560
status: params.status,
15581561
endedAt: params.endedAt,
15591562
lastEventAt: params.lastEventAt ?? params.endedAt,
1560-
...(params.error !== undefined ? { error: params.error } : {}),
15611563
...(params.terminalSummary !== undefined
15621564
? { terminalSummary: normalizeTaskSummary(params.terminalSummary) }
15631565
: {}),
@@ -1569,7 +1571,11 @@ export function markTaskTerminalById(params: {
15691571
}),
15701572
}
15711573
: {}),
1572-
});
1574+
};
1575+
if (Object.hasOwn(params, "error")) {
1576+
patch.error = params.error;
1577+
}
1578+
return updateTask(params.taskId, patch);
15731579
}
15741580

15751581
export function markTaskLostById(params: {

0 commit comments

Comments
 (0)