From e680b3dbeb262e470c2facec401bbe594b10a5ca Mon Sep 17 00:00:00 2001 From: Vincent Koc Date: Sun, 12 Jul 2026 13:50:28 +0800 Subject: [PATCH] fix(tasks): fail closed on registry restore errors (#105031) --- src/cli/gateway-cli/run-loop.test.ts | 77 ++++++++++++++++++ src/cli/gateway-cli/run-loop.ts | 6 +- src/tasks/task-registry.store.test.ts | 102 +++++++++++++++++++++--- src/tasks/task-registry.test.ts | 110 ++++++++++++++++++++++++++ src/tasks/task-registry.ts | 57 +++++++++---- 5 files changed, 325 insertions(+), 27 deletions(-) diff --git a/src/cli/gateway-cli/run-loop.test.ts b/src/cli/gateway-cli/run-loop.test.ts index cd7a8cd78f0e..67795b7a4614 100644 --- a/src/cli/gateway-cli/run-loop.test.ts +++ b/src/cli/gateway-cli/run-loop.test.ts @@ -1365,6 +1365,83 @@ describe("runGatewayLoop", () => { }); }); + it("keeps the process alive and retries after task-registry restore fails during restart", async () => { + vi.clearAllMocks(); + reloadTaskRegistryFromStore.mockReset(); + reloadTaskRegistryFromStore.mockImplementationOnce(() => { + throw new Error("task registry restore failed"); + }); + peekGatewaySigusr1RestartReason.mockReturnValue(undefined); + respawnGatewayProcessForUpdate.mockReturnValue({ + mode: "disabled", + detail: "OPENCLAW_NO_RESPAWN", + }); + + try { + await withIsolatedSignals(async ({ captureSignal }) => { + const closeFirst = createCloseMock(); + const closeSecond = createCloseMock(); + const { runtime, exited } = createRuntimeWithExitSignal(); + let resolveSecondStart: (() => void) | null = null; + const startedSecond = new Promise((resolve) => { + resolveSecondStart = resolve; + }); + const start = vi + .fn() + .mockResolvedValueOnce({ close: closeFirst }) + .mockImplementationOnce(async () => { + resolveSecondStart?.(); + return { close: closeSecond }; + }); + + const { runGatewayLoop } = await import("./run-loop.js"); + void runGatewayLoop({ + start: start as unknown as Parameters[0]["start"], + runtime: runtime as unknown as Parameters[0]["runtime"], + }); + await waitForLoopCondition( + () => start.mock.calls.length === 1, + "expected initial gateway start", + ); + const sigusr1 = captureSignal("SIGUSR1"); + const sigterm = captureSignal("SIGTERM"); + + try { + sigusr1(); + await waitForLoopCondition( + () => + gatewayLog.error.mock.calls.some(([message]) => + String(message).includes("gateway startup failed: task registry restore failed."), + ), + "expected failed task-registry restore to be logged", + ); + + expectRestartCloseCall(closeFirst, 90_000); + expect(reloadTaskRegistryFromStore).toHaveBeenCalledTimes(1); + expect(start).toHaveBeenCalledTimes(1); + expect(runtime.exit).not.toHaveBeenCalled(); + + sigusr1(); + await startedSecond; + + expect(reloadTaskRegistryFromStore).toHaveBeenCalledTimes(2); + expect(start).toHaveBeenCalledTimes(2); + expect(runtime.exit).not.toHaveBeenCalled(); + } finally { + sigterm(); + await expect(exited).resolves.toBe(0); + } + + expect(closeSecond).toHaveBeenCalledWith({ + reason: "gateway stopping", + restartExpectedMs: null, + }); + }); + } finally { + reloadTaskRegistryFromStore.mockReset(); + } + }); + it("uses the default restart drain timeout when config omits deferralTimeoutMs", async () => { vi.clearAllMocks(); loadConfig.mockReturnValue({ gateway: { reload: {} } }); diff --git a/src/cli/gateway-cli/run-loop.ts b/src/cli/gateway-cli/run-loop.ts index 8db3f0ad67a5..e534839ebeab 100644 --- a/src/cli/gateway-cli/run-loop.ts +++ b/src/cli/gateway-cli/run-loop.ts @@ -918,11 +918,13 @@ export async function runGatewayLoop(params: { // SIGTERM/SIGINT still exit after a graceful shutdown. let isFirstStart = true; for (;;) { - await onIteration(); + // The restart hook reopens admission before reloading durable state. Clear + // its local mirror first so a failed reload cannot skip the next drain. restartDrainingMarked = false; - startupStartedAt = Date.now(); let startupFailedBeforeServerHandle = false; try { + await onIteration(); + startupStartedAt = Date.now(); await params.beginBoot?.(startupStartedAt); server = await params.start({ startupStartedAt }); startupFailedWithoutServerHandle = false; diff --git a/src/tasks/task-registry.store.test.ts b/src/tasks/task-registry.store.test.ts index ca04dd93dffe..09004db64183 100644 --- a/src/tasks/task-registry.store.test.ts +++ b/src/tasks/task-registry.store.test.ts @@ -35,6 +35,7 @@ import { listFreshTasksForOwnerKey, markTaskTerminalById, maybeDeliverTaskStateChangeUpdate, + reloadTaskRegistryFromStore, resetTaskRegistryForTests, updateTaskNotifyPolicyById, } from "./task-registry.js"; @@ -179,24 +180,28 @@ describe("task-registry store runtime", () => { expect(latestSnapshot.tasks.get("task-restored")?.task).toBe("Restored task"); }); - it("logs restore parser failures and keeps the registry empty", async () => { + it("logs restore parser failures and keeps the failure sticky", async () => { const warnLogs = createWarnLogCapture("openclaw-task-registry-restore-test"); const invalidValue = "not-requested"; + const loadSnapshot = vi.fn(() => { + throw new Error(`Invalid persisted task delivery status: ${JSON.stringify(invalidValue)}`); + }); try { configureTaskRegistryRuntime({ store: { - loadSnapshot: () => { - throw new Error( - `Invalid persisted task delivery status: ${JSON.stringify(invalidValue)}`, - ); - }, + loadSnapshot, saveSnapshot: () => {}, }, }); - expect(findTaskByRunId("run-restored")).toBeUndefined(); + expect(() => findTaskByRunId("run-restored")).toThrow( + `Task registry restore failed: Invalid persisted task delivery status: "${invalidValue}"`, + ); expect(await warnLogs.findText(invalidValue)).toContain(invalidValue); - expect(getTaskById("task-restored")).toBeUndefined(); + expect(() => getTaskById("task-restored")).toThrow( + `Task registry restore failed: Invalid persisted task delivery status: "${invalidValue}"`, + ); + expect(loadSnapshot).toHaveBeenCalledTimes(1); } finally { warnLogs.cleanup(); } @@ -223,13 +228,92 @@ describe("task-registry store runtime", () => { }, }); - expect(findTaskByRunId("run-restored")).toBeUndefined(); + expect(() => findTaskByRunId("run-restored")).toThrow( + `Task registry restore failed: Invalid persisted task delivery status: "${invalidValue}"`, + ); expect(warn).toHaveBeenCalledTimes(1); expect(String(warn.mock.calls[0]?.[0])).toContain( `Failed to restore task registry: Invalid persisted task delivery status: "${invalidValue}"`, ); }); + it("blocks writes until an explicit reload recovers the registry", () => { + const storedTask = createStoredTask(); + let restoreShouldFail = true; + const loadSnapshot = vi.fn(() => { + if (restoreShouldFail) { + throw new Error("SQLITE_CORRUPT: malformed task registry"); + } + return { + tasks: new Map([[storedTask.taskId, storedTask]]), + deliveryStates: new Map(), + }; + }); + const upsertTaskWithDeliveryState = vi.fn(); + configureTaskRegistryRuntime({ + store: { + loadSnapshot, + saveSnapshot: () => {}, + upsertTaskWithDeliveryState, + }, + }); + + expect(() => + createTaskRecord({ + runtime: "cron", + ownerKey: "agent:main:main", + scopeKind: "session", + task: "Must not overwrite hidden durable tasks", + status: "queued", + deliveryStatus: "not_applicable", + notifyPolicy: "silent", + }), + ).toThrow("Task registry restore failed: SQLITE_CORRUPT: malformed task registry"); + expect(() => getTaskById(storedTask.taskId)).toThrow( + "Task registry restore failed: SQLITE_CORRUPT: malformed task registry", + ); + expect(upsertTaskWithDeliveryState).not.toHaveBeenCalled(); + expect(loadSnapshot).toHaveBeenCalledTimes(1); + + restoreShouldFail = false; + reloadTaskRegistryFromStore(); + + expect(getTaskById(storedTask.taskId)).toMatchObject({ taskId: storedTask.taskId }); + expect(loadSnapshot).toHaveBeenCalledTimes(2); + }); + + it("clears a sticky restore failure during the test reset boundary", () => { + const failedLoad = vi.fn(() => { + throw new Error("SQLITE_IOERR: failed to read task registry"); + }); + configureTaskRegistryRuntime({ + store: { + loadSnapshot: failedLoad, + saveSnapshot: () => {}, + }, + }); + + expect(() => getTaskById("task-restored")).toThrow( + "Task registry restore failed: SQLITE_IOERR: failed to read task registry", + ); + resetTaskRegistryForTests({ persist: false }); + + const cleanLoad = vi.fn(() => ({ + tasks: new Map(), + deliveryStates: new Map(), + })); + configureTaskRegistryRuntime({ + store: { + loadSnapshot: cleanLoad, + saveSnapshot: () => {}, + }, + }); + + expect(getTaskById("task-restored")).toBeUndefined(); + expect(failedLoad).toHaveBeenCalledTimes(1); + expect(cleanLoad).toHaveBeenCalledTimes(1); + }); + it("uses scoped owner lookups for fresh owner task reads", () => { const storedTask = createStoredTask(); const loadSnapshot = vi.fn(() => ({ diff --git a/src/tasks/task-registry.test.ts b/src/tasks/task-registry.test.ts index e99ab79de42c..e07c38800e5f 100644 --- a/src/tasks/task-registry.test.ts +++ b/src/tasks/task-registry.test.ts @@ -15,7 +15,9 @@ import { import type { SessionBindingRecord } from "../infra/outbound/session-binding-service.js"; import { peekSystemEvents, resetSystemEventsForTest } from "../infra/system-events.js"; import { + beginGatewayRestartSignalAdmission, getActiveGatewayRootWorkCount, + markGatewayRestartDraining, resetGatewayWorkAdmission, tryBeginGatewaySuspendAdmission, } from "../process/gateway-work-admission.js"; @@ -3690,6 +3692,114 @@ describe("task-registry", () => { }); }); + it("reattaches the lifecycle listener after recovering from an initial restore failure", async () => { + await withTaskRegistryTempDir(async () => { + resetTaskRegistryMemoryForTest(); + const runId = "run-restore-listener"; + const storedTask: TaskRecord = { + taskId: "task-restore-listener", + runtime: "acp", + requesterSessionKey: "agent:main:main", + ownerKey: "agent:main:main", + scopeKind: "session", + runId, + task: "Resume lifecycle tracking after restore recovery", + status: "running", + deliveryStatus: "not_applicable", + notifyPolicy: "silent", + createdAt: 100, + startedAt: 100, + lastEventAt: 100, + }; + let restoreShouldFail = true; + configureTaskRegistryRuntime({ + store: { + loadSnapshot: () => { + if (restoreShouldFail) { + throw new Error("SQLITE_IOERR: initial task restore failed"); + } + return { + tasks: new Map([[storedTask.taskId, storedTask]]), + deliveryStates: new Map(), + }; + }, + saveSnapshot: () => {}, + }, + }); + + expect(() => getTaskById(storedTask.taskId)).toThrow( + "Task registry restore failed: SQLITE_IOERR: initial task restore failed", + ); + restoreShouldFail = false; + reloadTaskRegistryFromStore(); + + emitAgentEvent({ + runId, + stream: "lifecycle", + data: { + phase: "end", + endedAt: 250, + }, + }); + + expectRecordFields(requireTaskByRunId(runId), { + status: "succeeded", + endedAt: 250, + }); + }); + }); + + it("does not hide a failed reload behind the restart-draining delivery fallback", async () => { + await withTaskRegistryTempDir(async () => { + resetTaskRegistryMemoryForTest(); + const storedTask: TaskRecord = { + taskId: "task-reload-failure", + runtime: "acp", + requesterSessionKey: "agent:main:main", + ownerKey: "agent:main:main", + scopeKind: "session", + runId: "run-reload-failure", + task: "Keep restore failures visible", + status: "succeeded", + deliveryStatus: "pending", + notifyPolicy: "done_only", + createdAt: 100, + endedAt: 200, + lastEventAt: 200, + }; + let restoreError: Error | null = null; + configureTaskRegistryRuntime({ + store: { + loadSnapshot: () => { + if (restoreError) { + throw restoreError; + } + return { + tasks: new Map([[storedTask.taskId, storedTask]]), + deliveryStates: new Map(), + }; + }, + saveSnapshot: () => {}, + }, + }); + expect(getTaskById(storedTask.taskId)?.taskId).toBe(storedTask.taskId); + + beginGatewayRestartSignalAdmission(); + const pendingDelivery = maybeDeliverTaskTerminalUpdate(storedTask.taskId); + await Promise.resolve(); + + restoreError = new Error("SQLITE_CORRUPT: task reload failed"); + expect(() => reloadTaskRegistryFromStore()).toThrow( + "Task registry restore failed: SQLITE_CORRUPT: task reload failed", + ); + markGatewayRestartDraining(); + + await expect(pendingDelivery).rejects.toThrow( + "Task registry restore failed: SQLITE_CORRUPT: task reload failed", + ); + }); + }); + it("summarizes inspectable task audit findings", async () => { await withTaskRegistryTempDir(async () => { resetTaskRegistryMemoryForTest(); diff --git a/src/tasks/task-registry.ts b/src/tasks/task-registry.ts index c401b13ffe11..66b9d7159ea6 100644 --- a/src/tasks/task-registry.ts +++ b/src/tasks/task-registry.ts @@ -77,7 +77,12 @@ const taskIdsByRelatedSessionKey = taskRegistryProcessState.taskIdsByRelatedSess const tasksWithPendingDelivery = taskRegistryProcessState.tasksWithPendingDelivery; let listenerStarted = false; let listenerStop: (() => void) | null = null; -let restoreAttempted = false; +type TaskRegistryRestoreState = + | { status: "uninitialized" } + | { status: "restoring" } + | { status: "ready" } + | { status: "failed"; error: Error }; +let taskRegistryRestoreState: TaskRegistryRestoreState = { status: "uninitialized" }; const taskFlowSyncRetryTimers = new Map>(); type TaskRegistryDeliveryRuntime = Pick< typeof import("./task-registry-delivery-runtime.js"), @@ -1208,36 +1213,54 @@ function clearTaskFlowSyncRetries(): void { } function restoreTaskRegistryOnce() { - if (restoreAttempted) { - return; + switch (taskRegistryRestoreState.status) { + case "ready": + return; + case "failed": + throw taskRegistryRestoreState.error; + case "restoring": + throw new Error("Task registry restore is already in progress."); + case "uninitialized": + break; } - restoreAttempted = true; + taskRegistryRestoreState = { status: "restoring" }; try { const restored = getTaskRegistryStore().loadSnapshot(); - if (restored.tasks.size === 0 && restored.deliveryStates.size === 0) { - return; - } + const restoredTasks = new Map(); for (const [taskId, task] of restored.tasks.entries()) { - tasks.set(taskId, normalizeTaskTimestamps(task)); + restoredTasks.set(taskId, normalizeTaskTimestamps(task)); } - for (const [taskId, state] of restored.deliveryStates.entries()) { + const restoredDeliveryStates = new Map(restored.deliveryStates); + + clearTaskRegistryMemory(); + for (const [taskId, task] of restoredTasks.entries()) { + tasks.set(taskId, task); + } + for (const [taskId, state] of restoredDeliveryStates.entries()) { taskDeliveryStates.set(taskId, state); } rebuildRunIdIndex(); rebuildOwnerKeyIndex(); rebuildParentFlowIdIndex(); rebuildRelatedSessionKeyIndex(); - emitTaskRegistryObserverEvent(() => ({ - kind: "restored", - tasks: snapshotTaskRecords(tasks), - })); + taskRegistryRestoreState = { status: "ready" }; + if (restoredTasks.size > 0 || restoredDeliveryStates.size > 0) { + emitTaskRegistryObserverEvent(() => ({ + kind: "restored", + tasks: snapshotTaskRecords(tasks), + })); + } } catch (error) { + clearTaskRegistryMemory(); const message = formatErrorMessage(error); + const restoreError = new Error(`Task registry restore failed: ${message}`, { cause: error }); + taskRegistryRestoreState = { status: "failed", error: restoreError }; // Compact console logs omit structured metadata, so keep the rejected value visible there too. log.warn("Failed to restore task registry", { error: message, consoleMessage: `Failed to restore task registry: ${message}`, }); + throw restoreError; } } @@ -1248,8 +1271,8 @@ export function ensureTaskRegistryReady() { export function reloadTaskRegistryFromStore(): void { clearTaskRegistryMemory(); - restoreAttempted = false; - restoreTaskRegistryOnce(); + taskRegistryRestoreState = { status: "uninitialized" }; + ensureTaskRegistryReady(); } function updateTask(taskId: string, patch: Partial): TaskRecord | null { @@ -1428,6 +1451,7 @@ async function runTaskDeliveryWithIndependentAdmission( taskId: string, deliver: () => Promise, ): Promise { + ensureTaskRegistryReady(); let admitted = false; try { return await runWithGatewayIndependentRootWorkAdmission(async () => { @@ -1439,6 +1463,7 @@ async function runTaskDeliveryWithIndependentAdmission( // restart closes admission. An already-admitted delivery still reports its // own failures instead of hiding them behind a concurrent restart. if (!admitted && isGatewayRestartDraining()) { + ensureTaskRegistryReady(); const current = tasks.get(taskId); return current ? cloneTaskRecord(current) : null; } @@ -2675,7 +2700,7 @@ export function deleteTaskRecordById(taskId: string): boolean { export function resetTaskRegistryForTests(opts?: { persist?: boolean }) { clearTaskRegistryMemory(); - restoreAttempted = false; + taskRegistryRestoreState = { status: "uninitialized" }; resetTaskRegistryRuntimeForTests(); if (listenerStop) { listenerStop();