From b951bb8223f90f8d680be2eb17523cb033cde942 Mon Sep 17 00:00:00 2001 From: Peter Steinberger Date: Tue, 25 Aug 2026 20:22:30 -0700 Subject: [PATCH] fix(cron): preserve watched commands across gateway hot reload (#115761) Co-authored-by: Andy Ye <35905412+TurboTheTurtle@users.noreply.github.com> --- src/gateway/cron-exit-watchers.test.ts | 86 +++++++++++ src/gateway/cron-exit-watchers.ts | 172 ++++++++++++++------- src/gateway/server-cron-lazy.test.ts | 43 ++++++ src/gateway/server-cron-lazy.ts | 86 ++++++++--- src/gateway/server-cron.drain.test.ts | 1 + src/gateway/server-cron.test.ts | 73 +++++++++ src/gateway/server-cron.ts | 84 ++++++++-- src/gateway/server-reload-handlers.test.ts | 107 +++++++++++++ src/gateway/server-reload-hot.ts | 23 ++- 9 files changed, 576 insertions(+), 99 deletions(-) diff --git a/src/gateway/cron-exit-watchers.test.ts b/src/gateway/cron-exit-watchers.test.ts index 7fb9b4dc213f..a6da94d9547d 100644 --- a/src/gateway/cron-exit-watchers.test.ts +++ b/src/gateway/cron-exit-watchers.test.ts @@ -146,6 +146,92 @@ describe("createCronExitWatchers", () => { expect(order).toEqual(["persist", "fire"]); }); + it("rebinds live watchers but drains callbacks already owned by the previous scheduler", async () => { + const { supervisor, runs } = makeFakeSupervisor(); + let releasePersistence = () => {}; + const persistenceGate = new Promise((resolve) => { + releasePersistence = resolve; + }); + const releaseCompletion = vi.fn(); + const oldPersistCompletion = vi.fn(async () => { + await persistenceGate; + return releaseCompletion; + }); + const oldFireOnExit = vi.fn(async () => {}); + const newPersistCompletion = vi.fn(async () => {}); + const newFireOnExit = vi.fn(async () => {}); + const watchers = createCronExitWatchers({ + getProcessSupervisor: () => supervisor as never, + persistCompletion: oldPersistCompletion, + fireOnExit: oldFireOnExit, + logger: noopLogger, + }); + + watchers.reconcile([onExitJob("old-owner"), onExitJob("new-owner")]); + await flush(); + expectDefined(runs[0], "runs[0] test invariant").deferred.resolve({ + exitCode: 0, + reason: "exit", + }); + await vi.waitFor(() => expect(oldPersistCompletion).toHaveBeenCalledOnce()); + + const handoff = watchers.updateHandlers({ + getProcessSupervisor: () => supervisor as never, + persistCompletion: newPersistCompletion, + fireOnExit: newFireOnExit, + logger: noopLogger, + }); + const handoffSettled = vi.fn(); + void handoff?.then(handoffSettled); + + expectDefined(runs[1], "runs[1] test invariant").deferred.resolve({ + exitCode: 0, + reason: "exit", + }); + await vi.waitFor(() => expect(newFireOnExit).toHaveBeenCalledOnce()); + expect(newPersistCompletion).toHaveBeenCalledOnce(); + expect(oldFireOnExit).not.toHaveBeenCalled(); + expect(handoffSettled).not.toHaveBeenCalled(); + + releasePersistence(); + await handoff; + await vi.waitFor(() => expect(oldFireOnExit).toHaveBeenCalledOnce()); + expect(releaseCompletion).toHaveBeenCalledOnce(); + expect(supervisor.spawn).toHaveBeenCalledTimes(2); + expect(supervisor.cancelScope).not.toHaveBeenCalled(); + }); + + it("routes post-handoff retries through the replacement scheduler owner", async () => { + const { supervisor, runs } = makeFakeSupervisor(); + const oldUpdateWatcherState = vi.fn(async () => {}); + const newUpdateWatcherState = vi.fn(async () => {}); + const watchers = createCronExitWatchers({ + getProcessSupervisor: () => supervisor as never, + persistCompletion: vi.fn(async () => {}), + fireOnExit: vi.fn(async () => {}), + updateWatcherState: oldUpdateWatcherState, + logger: noopLogger, + retryBackoffMs: [0], + }); + + watchers.reconcile([onExitJob("job-a")]); + await flush(); + await watchers.updateHandlers({ + getProcessSupervisor: () => supervisor as never, + persistCompletion: vi.fn(async () => {}), + fireOnExit: vi.fn(async () => {}), + updateWatcherState: newUpdateWatcherState, + logger: noopLogger, + }); + expectDefined(runs[0], "runs[0] test invariant").deferred.reject(new Error("wait failed")); + + await vi.waitFor(() => expect(newUpdateWatcherState).toHaveBeenCalledOnce()); + expect(oldUpdateWatcherState).not.toHaveBeenCalled(); + await delay(5); + await flush(); + expect(supervisor.spawn).toHaveBeenCalledTimes(2); + }); + it("a fired job stays unarmed across a simulated restart (disabled in store → not re-run)", async () => { // persistCompletion disables the job; after a restart the reconcile sees a // disabled job and must NOT re-arm (which would re-run the command). diff --git a/src/gateway/cron-exit-watchers.ts b/src/gateway/cron-exit-watchers.ts index 64eb0d08a28c..3835ac589e07 100644 --- a/src/gateway/cron-exit-watchers.ts +++ b/src/gateway/cron-exit-watchers.ts @@ -29,11 +29,23 @@ export type CronExitResult = { noOutputTimedOut: boolean; }; -type CronExitWatchers = { +export type CronExitWatcherHandlers = { + getProcessSupervisor: () => ProcessSupervisor; + persistCompletion: (job: OnExitCronJob) => Promise<(() => void) | void>; + fireOnExit: (job: CronJob, exit: CronExitResult) => void | Promise; + updateWatcherState?: ( + job: OnExitCronJob, + patch: Pick, + ) => Promise; + logger: Logger; +}; + +export type CronExitWatchers = { reconcile: (jobs: CronJob[]) => void; cancel: (jobId: string) => void; cancelAll: () => Promise; activeJobIds: () => string[]; + updateHandlers: (handlers: CronExitWatcherHandlers) => Promise | void; }; const SCOPE_PREFIX = "cron-exit"; @@ -46,18 +58,26 @@ function isWatchableExitJob(job: CronJob): job is OnExitCronJob { return job.enabled && job.schedule.kind === "on-exit"; } -export function createCronExitWatchers(params: { - getProcessSupervisor: () => ProcessSupervisor; - persistCompletion: (job: OnExitCronJob) => Promise<(() => void) | void>; - fireOnExit: (job: CronJob, exit: CronExitResult) => void | Promise; - updateWatcherState?: ( - job: OnExitCronJob, - patch: Pick, - ) => Promise; - logger: Logger; - shell?: { command: string; argsFor: (command: string) => string[] }; - retryBackoffMs?: readonly number[]; -}): CronExitWatchers { +export function createCronExitWatchers( + params: CronExitWatcherHandlers & { + shell?: { command: string; argsFor: (command: string) => string[] }; + retryBackoffMs?: readonly number[]; + }, +): CronExitWatchers { + let handlers: CronExitWatcherHandlers = params; + const ownerSettlements = new Set>(); + const settleOwnerCallback = async (operation: Promise): Promise => { + const settlement = operation.then( + () => undefined, + () => undefined, + ); + ownerSettlements.add(settlement); + try { + return await operation; + } finally { + ownerSettlements.delete(settlement); + } + }; const shell = params.shell ?? resolveExitWatchShell(); const retryBackoffMs = params.retryBackoffMs && params.retryBackoffMs.length > 0 @@ -110,9 +130,9 @@ export function createCronExitWatchers(params: { // killed by the arm() ownership check once it resolves. slot.run?.cancel("manual-cancel"); try { - params.getProcessSupervisor().cancelScope(scopeKey(jobId), "manual-cancel"); + handlers.getProcessSupervisor().cancelScope(scopeKey(jobId), "manual-cancel"); } catch (err) { - params.logger.warn({ err: String(err), jobId }, "cron-exit: cancel watcher failed"); + handlers.logger.warn({ err: String(err), jobId }, "cron-exit: cancel watcher failed"); } }; @@ -141,16 +161,17 @@ export function createCronExitWatchers(params: { const persistWatcherState = async ( patch: Pick, ) => { - if (!params.updateWatcherState) { + const owner = handlers; + if (!owner.updateWatcherState) { return; } try { - const updated = await params.updateWatcherState(slot.job, patch); + const updated = await settleOwnerCallback(owner.updateWatcherState(slot.job, patch)); if (owns() && updated && isWatchableExitJob(updated)) { slot.job = updated; } } catch (err) { - params.logger.warn( + owner.logger.warn( { err: String(err), jobId: slot.job.id }, "cron-exit: failed to persist watcher state", ); @@ -180,7 +201,7 @@ export function createCronExitWatchers(params: { arm(slot.job, slot.consecutiveFailures); }, delayMs); slot.retryTimer.unref?.(); - params.logger.warn( + handlers.logger.warn( { err: String(error), jobId: slot.job.id, retryInMs: delayMs }, `cron-exit: watcher ${phase} failed; retry scheduled`, ); @@ -188,7 +209,7 @@ export function createCronExitWatchers(params: { void (async () => { let run: ManagedRun; try { - run = await params.getProcessSupervisor().spawn({ + run = await handlers.getProcessSupervisor().spawn({ sessionId: `cron-exit:${job.id}`, backendId: "cron-exit-watch", scopeKey: scopeKey(job.id), @@ -223,7 +244,10 @@ export function createCronExitWatchers(params: { return; } slot.run = run; - params.logger.info({ jobId: job.id, runId: run.runId, command }, "cron-exit: watcher armed"); + handlers.logger.info( + { jobId: job.id, runId: run.runId, command }, + "cron-exit: watcher armed", + ); let exit: Awaited>; try { exit = await run.wait(); @@ -237,7 +261,8 @@ export function createCronExitWatchers(params: { if (!owns()) { return; } - params.logger.info( + const owner = handlers; + owner.logger.info( { jobId: job.id, exitCode: exit.exitCode, reason: exit.reason }, "cron-exit: watched command exited; firing job", ); @@ -245,45 +270,52 @@ export function createCronExitWatchers(params: { // Persist the terminal one-shot state BEFORE firing. FAIL CLOSED: if the // store write fails we do NOT wake — waking without a persisted terminal // state would let a gateway restart re-arm and re-run the command. - let releaseCompletion: (() => void) | void; try { - releaseCompletion = await params.persistCompletion(slot.job); - } catch (err) { - if (owns()) { - active.delete(job.id); - } - params.logger.warn( - { err: String(err), jobId: job.id }, - "cron-exit: persistCompletion failed; NOT firing (fail closed to avoid replay)", + await settleOwnerCallback( + (async () => { + let releaseCompletion: (() => void) | void; + try { + releaseCompletion = await owner.persistCompletion(slot.job); + } catch (err) { + if (owns()) { + active.delete(job.id); + } + owner.logger.warn( + { err: String(err), jobId: job.id }, + "cron-exit: persistCompletion failed; NOT firing (fail closed to avoid replay)", + ); + return; + } + try { + if (!owns() || slot.cancelled) { + if (active.get(job.id) === slot) { + active.delete(job.id); + } + return; + } + slot.fired = true; + try { + await owner.fireOnExit(slot.job, { + exitCode: exit.exitCode, + reason: exit.reason, + stdout: exit.stdout, + stderr: exit.stderr, + timedOut: exit.timedOut, + noOutputTimedOut: exit.noOutputTimedOut, + }); + } catch (err) { + owner.logger.warn( + { err: String(err), jobId: job.id }, + "cron-exit: fireOnExit after exit failed", + ); + } + } finally { + releaseCompletion?.(); + } + })(), ); - return; - } - slot.terminalPersisting = false; - try { - if (!owns() || slot.cancelled) { - if (active.get(job.id) === slot) { - active.delete(job.id); - } - return; - } - slot.fired = true; - try { - await params.fireOnExit(slot.job, { - exitCode: exit.exitCode, - reason: exit.reason, - stdout: exit.stdout, - stderr: exit.stderr, - timedOut: exit.timedOut, - noOutputTimedOut: exit.noOutputTimedOut, - }); - } catch (err) { - params.logger.warn( - { err: String(err), jobId: job.id }, - "cron-exit: fireOnExit after exit failed", - ); - } } finally { - releaseCompletion?.(); + slot.terminalPersisting = false; } })().finally(() => { slot.lifecycleSettled = true; @@ -296,10 +328,23 @@ export function createCronExitWatchers(params: { }; const reconcile = (jobs: CronJob[]) => { + const jobsById = new Map(jobs.map((job) => [job.id, job] as const)); const want = new Map(jobs.filter(isWatchableExitJob).map((j) => [j.id, j] as const)); // Cancel watchers whose job is gone or no longer watchable. - for (const jobId of Array.from(active.keys())) { + for (const [jobId, slot] of Array.from(active.entries())) { if (!want.has(jobId)) { + const storedJob = jobsById.get(jobId); + // A replacement can observe the terminal disable before its previous + // owner's completion callback has settled. + if ( + slot.terminalPersisting && + storedJob?.schedule.kind === "on-exit" && + !storedJob.enabled && + slot.command === storedJob.schedule.command && + slot.cwd === storedJob.schedule.cwd + ) { + continue; + } cancel(jobId); } } @@ -342,5 +387,14 @@ export function createCronExitWatchers(params: { ...Array.from(settlingCancelledSlots, (slot) => slot.job.id), ]), ), + updateHandlers: (nextHandlers) => { + handlers = nextHandlers; + if (ownerSettlements.size > 0) { + // Finish callbacks that already captured the old scheduler, but keep + // live watched children running under their newly adopted owner. + return Promise.all(ownerSettlements).then(() => undefined); + } + return undefined; + }, }; } diff --git a/src/gateway/server-cron-lazy.test.ts b/src/gateway/server-cron-lazy.test.ts index a5fa7638309e..eda9b3a4ae2a 100644 --- a/src/gateway/server-cron-lazy.test.ts +++ b/src/gateway/server-cron-lazy.test.ts @@ -87,6 +87,49 @@ describe("createLazyGatewayCronState", () => { expect(cron["readJob"]).toHaveBeenCalledWith("demo"); }); + it("does not load cron solely to prepare a watcher handoff", async () => { + const lazy = createLazyGatewayCronState(createParams()); + + await expect(lazy.prepareExitWatcherHandoff?.()).resolves.toBeUndefined(); + expect(hoisted.buildGatewayCronService).not.toHaveBeenCalled(); + }); + + it("preserves a watcher owner when hot reload overtakes lazy startup", async () => { + const finishStart = deferred(); + const cron = createCronService(); + cron.start = vi.fn(async () => await finishStart.promise); + const watchers = { + reconcile: vi.fn(), + cancel: vi.fn(), + cancelAll: vi.fn(async () => {}), + activeJobIds: vi.fn(() => ["watched-job"]), + updateHandlers: vi.fn(), + }; + const stopOwner = vi.fn(async () => {}); + hoisted.setState({ + ...createCronState(cron), + prepareExitWatcherHandoff: vi.fn(async () => ({ + current: () => watchers, + adopt: vi.fn(), + stopOwner, + })), + }); + const lazy = createLazyGatewayCronState(createParams()); + + const start = lazy.cron.start(); + await vi.waitFor(() => expect(cron["start"]).toHaveBeenCalledOnce()); + const handoff = await lazy.prepareExitWatcherHandoff?.(); + expect(handoff?.current()).toBe(watchers); + + await handoff?.stopOwner(); + finishStart.resolve(); + await start; + + expect(stopOwner).toHaveBeenCalledOnce(); + expect(watchers.cancelAll).not.toHaveBeenCalled(); + expect(cron["stop"]).not.toHaveBeenCalled(); + }); + it("forwards run payload overrides to the loaded cron service", async () => { const cron = createCronService(); hoisted.setState(createCronState(cron)); diff --git a/src/gateway/server-cron-lazy.ts b/src/gateway/server-cron-lazy.ts index 8e39fcc7c8ae..ec9a6475d717 100644 --- a/src/gateway/server-cron-lazy.ts +++ b/src/gateway/server-cron-lazy.ts @@ -5,7 +5,7 @@ import type { OpenClawConfig } from "../config/types.openclaw.js"; import { resolveCronJobsStorePathFromConfig } from "../cron/store.js"; import { createLazyPromiseLoader } from "../shared/lazy-runtime.js"; import type { GatewayCronServiceContract } from "./server-cron-contract.js"; -import type { GatewayCronState } from "./server-cron.js"; +import type { GatewayCronExitWatcherHandoff, GatewayCronState } from "./server-cron.js"; import type { GatewayRequestContext } from "./server-methods/types.js"; type LazyGatewayCronParams = { @@ -38,6 +38,8 @@ export function createLazyGatewayCronState(params: LazyGatewayCronParams): Gatew const cronEnabled = env.OPENCLAW_SKIP_CRON !== "1" && params.cfg.cron?.enabled !== false; let loaded: LoadedGatewayCronState | null = null; let stopped = false; + let preserveExitWatchersOnStop = false; + let exitWatcherHandoffStop: Promise | undefined; let lifecycleGeneration = 0; let schedulingPaused = false; const schedulingResumeWaiters = new Set<() => void>(); @@ -86,6 +88,48 @@ export function createLazyGatewayCronState(params: LazyGatewayCronParams): Gatew return await cronStateLoader.load(); }; + const stopResolvedCron = async ( + resolved: LoadedGatewayCronState, + preserveExitWatchers: boolean, + ): Promise => { + resolved.phase = "stopped"; + resolved.underlyingStarted = false; + const handoff = preserveExitWatchers + ? await resolved.state.prepareExitWatcherHandoff?.() + : undefined; + if (handoff) { + await handoff.stopOwner(); + } else if (resolved.state.cron.stopAndDrain) { + await resolved.state.cron.stopAndDrain(); + } else { + resolved.state.cron.stop(); + await resolved.state.stopStreamWatchers(); + } + }; + + const stopResolvedCronOnce = ( + resolved: LoadedGatewayCronState, + preserveExitWatchers: boolean, + ): Promise => { + if (!preserveExitWatchers) { + return stopResolvedCron(resolved, false); + } + exitWatcherHandoffStop ??= stopResolvedCron(resolved, true); + return exitWatcherHandoffStop; + }; + + const stopLoadedCronAndDrain = async (preserveExitWatchers = false): Promise => { + stopped = true; + preserveExitWatchersOnStop ||= preserveExitWatchers; + lifecycleGeneration += 1; + releaseSchedulingResumeWaiters(); + const loading = cronStateLoader.peek(); + const resolved = loaded ?? (loading ? await loading : null); + if (resolved) { + await stopResolvedCronOnce(resolved, preserveExitWatchersOnStop); + } + }; + const cron: GatewayCronServiceContract = { async start() { stopped = false; @@ -140,10 +184,7 @@ export function createLazyGatewayCronState(params: LazyGatewayCronParams): Gatew resolved.underlyingStartInFlight = false; } if (startCancelled()) { - resolved.phase = "stopped"; - resolved.underlyingStarted = false; - resolved.state.cron.stop(); - await resolved.state.stopStreamWatchers(); + await stopResolvedCronOnce(resolved, preserveExitWatchersOnStop); return; } if (schedulingPaused) { @@ -164,10 +205,7 @@ export function createLazyGatewayCronState(params: LazyGatewayCronParams): Gatew throw err; } if (startCancelled()) { - resolved.phase = "stopped"; - resolved.underlyingStarted = false; - resolved.state.cron.stop(); - await resolved.state.stopStreamWatchers(); + await stopResolvedCronOnce(resolved, preserveExitWatchersOnStop); return; } resolved.phase = "started"; @@ -209,21 +247,7 @@ export function createLazyGatewayCronState(params: LazyGatewayCronParams): Gatew } }, async stopAndDrain() { - stopped = true; - lifecycleGeneration += 1; - releaseSchedulingResumeWaiters(); - const resolved = loaded ?? (cronStateLoader.peek() ? await cronStateLoader.peek() : null); - if (!resolved) { - return; - } - resolved.phase = "stopped"; - resolved.underlyingStarted = false; - if (resolved.state.cron.stopAndDrain) { - await resolved.state.cron.stopAndDrain(); - } else { - resolved.state.cron.stop(); - await resolved.state.stopStreamWatchers(); - } + await stopLoadedCronAndDrain(); }, pauseScheduling() { schedulingPaused = true; @@ -320,6 +344,20 @@ export function createLazyGatewayCronState(params: LazyGatewayCronParams): Gatew cron, storePath, cronEnabled, + prepareExitWatcherHandoff: async (): Promise => { + const loading = cronStateLoader.peek(); + const resolved = loaded ?? (loading ? await loading : null); + const handoff = await resolved?.state.prepareExitWatcherHandoff?.(); + if (!handoff) { + return undefined; + } + return { + ...handoff, + stopOwner: async () => { + await stopLoadedCronAndDrain(true); + }, + }; + }, // Reload rules invoke these hooks on whatever cronState is live; the lazy // proxy must forward every GatewayCronState member or hot reloads silently // no-op until a gateway restart (heartbeat cadence changes never applied). diff --git a/src/gateway/server-cron.drain.test.ts b/src/gateway/server-cron.drain.test.ts index 3d5f1e502f9d..282996454e1e 100644 --- a/src/gateway/server-cron.drain.test.ts +++ b/src/gateway/server-cron.drain.test.ts @@ -24,6 +24,7 @@ vi.mock("./cron-exit-watchers.js", async (importOriginal) => ({ cancel: vi.fn(), cancelAll: cancelAllMock, activeJobIds: () => [], + updateHandlers: vi.fn(), }), })); diff --git a/src/gateway/server-cron.test.ts b/src/gateway/server-cron.test.ts index a04fe0a508aa..3d21d9c62c4c 100644 --- a/src/gateway/server-cron.test.ts +++ b/src/gateway/server-cron.test.ts @@ -903,6 +903,79 @@ describe("buildGatewayCronService", () => { } }); + it.each(["update", "updateWithPrecondition", "add"] as const)( + "honors an explicit %s disable while terminal persistence is settling", + async (mutation) => { + const commandExit = createDeferred(); + const completionPersistCommitted = createDeferred(); + const allowCompletionPersist = createDeferred(); + const cancel = vi.fn(); + const cancelScope = vi.fn(); + const spawn = vi.fn(async () => ({ + runId: "run-on-exit-explicit-disable", + startedAtMs: Date.now(), + cancel, + wait: () => commandExit.promise, + })); + getProcessSupervisorMock.mockReturnValue({ spawn, cancelScope }); + const state = loadCronService( + createCronConfig(`server-cron-on-exit-explicit-disable-${mutation}`), + ); + const originalUpdateWithPrecondition = state.cron.updateWithPrecondition.bind(state.cron); + let gateTerminalCompletion = true; + vi.spyOn(state.cron, "updateWithPrecondition").mockImplementation(async (...args) => { + if (!gateTerminalCompletion) { + return await originalUpdateWithPrecondition(...args); + } + gateTerminalCompletion = false; + const result = await originalUpdateWithPrecondition(...args); + completionPersistCommitted.resolve(); + await allowCompletionPersist.promise; + return result; + }); + const run = vi.spyOn(state.cron, "run"); + + try { + const input = { + name: "watch and honor explicit disable", + declarationKey: "agent:main:watch-and-honor-explicit-disable", + enabled: true, + schedule: { kind: "on-exit" as const, command: "true" }, + payload: { kind: "systemEvent" as const, text: "must not fire" }, + sessionTarget: "main" as const, + wakeMode: "now" as const, + }; + const job = await state.cron.add(input); + await state.reconcileExitWatchers(); + await vi.waitFor(() => expect(spawn).toHaveBeenCalledOnce()); + + commandExit.resolve(runExit()); + await completionPersistCommitted.promise; + + if (mutation === "updateWithPrecondition") { + await state.cron.updateWithPrecondition(job.id, { enabled: false }, () => {}); + } else if (mutation === "add") { + await state.cron.add({ ...input, enabled: false }, { enabledExplicit: true }); + } else { + await state.cron.update(job.id, { enabled: false }); + } + expect(cancel).toHaveBeenCalledWith("manual-cancel"); + expect(cancelScope).toHaveBeenCalledWith(`cron-exit:${job.id}`, "manual-cancel"); + allowCompletionPersist.resolve(); + + await vi.waitFor(async () => { + const handoff = await state.prepareExitWatcherHandoff?.(); + expect(handoff?.current().activeJobIds()).toEqual([]); + }); + expect(run).not.toHaveBeenCalled(); + } finally { + allowCompletionPersist.resolve(); + commandExit.resolve(runExit()); + await state.cron.stopAndDrain?.(); + } + }, + ); + it("aborts and drains active cron runs during shutdown", async () => { const controller = new AbortController(); const coreRun = new Promise((resolve) => { diff --git a/src/gateway/server-cron.ts b/src/gateway/server-cron.ts index b1a168f8c96c..2f78a74fb7bc 100644 --- a/src/gateway/server-cron.ts +++ b/src/gateway/server-cron.ts @@ -81,7 +81,12 @@ import { } from "../routing/session-key.js"; import { defaultRuntime } from "../runtime.js"; import { parseAgentSessionKey } from "../sessions/session-key-utils.js"; -import { createCronExitWatchers, type CronExitResult } from "./cron-exit-watchers.js"; +import { + createCronExitWatchers, + type CronExitResult, + type CronExitWatcherHandlers, + type CronExitWatchers, +} from "./cron-exit-watchers.js"; import { createCronStreamWatchers, type CronStreamFireDisposition, @@ -112,6 +117,7 @@ export type GatewayCronState = { cron: GatewayCronServiceContract; storePath: string; cronEnabled: boolean; + prepareExitWatcherHandoff?: () => Promise; // Required, not optional: reload rules call these hooks directly on whatever // cronState is live (including the lazy proxy). An optional member here let // the proxy silently omit reconcileHeartbeatJobs, turning every @@ -123,6 +129,12 @@ export type GatewayCronState = { reconcileHeartbeatJobs: (cfg?: OpenClawConfig) => Promise; }; +export type GatewayCronExitWatcherHandoff = { + current: () => CronExitWatchers; + adopt: (watchers: CronExitWatchers) => Promise | void; + stopOwner: () => Promise; +}; + function formatOnExitRunSummary(exit: CronExitResult): string { const lines = [ "Watched command finished.", @@ -562,7 +574,10 @@ export function buildGatewayCronService(params: { } = { current: undefined }; let exitWatcherReconciliations = 0; let streamWatcherReconciliations = 0; - const terminalExitCompletionTokens = new Map(); + const terminalExitCompletionTokens = new Map< + string, + Parameters[2] + >(); let exitWatcherGeneration = 0; let exitWatcherMutationRevision = 0; let exitWatchersStopped = false; @@ -1136,10 +1151,14 @@ export function buildGatewayCronService(params: { }, }); - exitWatchersRef.current = createCronExitWatchers({ + const exitWatcherHandlers = { getProcessSupervisor, persistCompletion: async (job) => { - const completionToken = {}; + const completionToken: Parameters[2] = (current) => { + if (!current.enabled || current.updatedAtMs !== job.updatedAtMs) { + throw new Error("cron on-exit job changed before completion"); + } + }; terminalExitCompletionTokens.set(job.id, completionToken); const releaseCompletionToken = () => { if (terminalExitCompletionTokens.get(job.id) === completionToken) { @@ -1148,11 +1167,7 @@ export function buildGatewayCronService(params: { }; try { await runWithGatewayIndependentRootWorkAdmission(async () => { - await cron.updateWithPrecondition(job.id, { enabled: false }, (current) => { - if (!current.enabled || current.updatedAtMs !== job.updatedAtMs) { - throw new Error("cron on-exit job changed before completion"); - } - }); + await cron.updateWithPrecondition(job.id, { enabled: false }, completionToken); }); return () => { releaseCompletionToken(); @@ -1191,7 +1206,8 @@ export function buildGatewayCronService(params: { } }), logger: cronLogger, - }); + } satisfies CronExitWatcherHandlers; + exitWatchersRef.current = createCronExitWatchers(exitWatcherHandlers); const updateCron = cron.update.bind(cron); streamWatchersRef.current = createCronStreamWatchers({ getProcessSupervisor, @@ -1270,10 +1286,21 @@ export function buildGatewayCronService(params: { patch.schedule !== undefined ? "schedule-update" : "disabled", ); }; + const cancelDisabledExitWatcher = (job: CronJob) => { + if (job.enabled || job.schedule.kind !== "on-exit") { + return; + } + // An operator disable wins over a completion retained during owner handoff. + exitWatcherMutationRevision += 1; + exitWatchersRef.current?.cancel(job.id); + }; const addCron = cron.add.bind(cron); cron.add = async (input, options) => { const result = await addCron(input, options); const addedJob = "job" in result ? result.job : result; + if (options?.enabledExplicit && !input.enabled) { + cancelDisabledExitWatcher(addedJob); + } await routeCurrentStreamJob(addedJob.id, addedJob, "added"); return result; }; @@ -1314,6 +1341,9 @@ export function buildGatewayCronService(params: { }; try { const result = await updateCronWithPrecondition(jobId, patch, routeAfterValidation, opts); + if (patch.enabled === false) { + cancelDisabledExitWatcher(result); + } await settleStopAfterCommittedUpdate(jobId, lifecycleStop); await routeLiveStreamJobLogged(jobId); return result; @@ -1333,6 +1363,9 @@ export function buildGatewayCronService(params: { }; try { const result = await updateCronWithPrecondition(jobId, patch, routeAfterPrecondition, opts); + if (patch.enabled === false && terminalExitCompletionTokens.get(jobId) !== precondition) { + cancelDisabledExitWatcher(result); + } await settleStopAfterCommittedUpdate(jobId, lifecycleStop); await routeLiveStreamJobLogged(jobId); return result; @@ -1408,10 +1441,17 @@ export function buildGatewayCronService(params: { }; const automationEpoch = claimSessionAutomationEpoch(); const stopCron = cron.stop.bind(cron); - cron.stop = () => { + const stopCronLifecycle = (preserveExitWatchers = false) => { try { stopCron(); - stopExitWatchers(); + if (preserveExitWatchers) { + // A committed replacement owns these children; fence this scheduler + // without terminating the adopted manager. + exitWatchersStopped = true; + exitWatcherGeneration += 1; + } else { + stopExitWatchers(); + } stopHeartbeatReconcileRetry(); void stopStreamWatchers().catch((err: unknown) => { cronLogger.warn( @@ -1425,8 +1465,11 @@ export function buildGatewayCronService(params: { unregisterSessionAutomationSource(automationSource); } }; - cron.stopAndDrain = async () => { - cron.stop(); + cron.stop = () => { + stopCronLifecycle(); + }; + const stopAndDrainCron = async (preserveExitWatchers = false) => { + stopCronLifecycle(preserveExitWatchers); const exitWatchersStop = exitWatchersStopPromise ?? Promise.resolve(); const streamWatchersStop = stopStreamWatchers().then( () => ({ ok: true as const }), @@ -1448,6 +1491,9 @@ export function buildGatewayCronService(params: { throw streamWatchersResult.error; } }; + cron.stopAndDrain = async () => { + await stopAndDrainCron(); + }; // Reconciliations serialize on one tail and only the latest requested epoch // executes, so an older reload's convergence can never clobber a newer one. // A failed pass schedules one bounded retry; a newer request supersedes it. @@ -1536,6 +1582,16 @@ export function buildGatewayCronService(params: { cron, storePath, cronEnabled, + prepareExitWatcherHandoff: async () => ({ + current: () => exitWatchersRef.current!, + adopt: (watchers) => { + exitWatchersRef.current = watchers; + return watchers.updateHandlers(exitWatcherHandlers); + }, + stopOwner: async () => { + await stopAndDrainCron(true); + }, + }), reconcileExitWatchers, stopExitWatchers, reconcileStreamWatchers, diff --git a/src/gateway/server-reload-handlers.test.ts b/src/gateway/server-reload-handlers.test.ts index bc3539a8e0c2..dce75cafa49f 100644 --- a/src/gateway/server-reload-handlers.test.ts +++ b/src/gateway/server-reload-handlers.test.ts @@ -2,6 +2,7 @@ * Gateway config reload handler tests. */ import fs from "node:fs"; +import { readFile, writeFile } from "node:fs/promises"; import path from "node:path"; import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; import { createDeferred } from "../../test/helpers/promise.js"; @@ -58,6 +59,8 @@ import { tryBeginGatewaySuspendAdmission, } from "../process/gateway-work-admission.js"; import { CommandLane } from "../process/lanes.js"; +import { getProcessSupervisor } from "../process/supervisor/index.js"; +import { buildWindowsCmdExeCommandLine } from "../process/windows-command.js"; import { resolveAuthProfileSecretOwnerId } from "../secrets/runtime-auth-profile-owner.js"; import { listActiveDegradedSecretOwners } from "../secrets/runtime-degraded-state.js"; import { createEmptyRuntimeWebToolsMetadata } from "../secrets/runtime-fast-path.js"; @@ -1270,6 +1273,110 @@ describe("provider auth hot reload path ownership", () => { }); describe("gateway hot reload model state", () => { + it("keeps a supervised on-exit child alive exactly once across same-store cron reload", async () => { + const fixtureDir = autoCleanupTempDirs.make("openclaw-cron-exit-reload-"); + const childScriptPath = path.join(fixtureDir, "watcher.cjs"); + const markerPath = path.join(fixtureDir, "watcher-runs.txt"); + const releasePath = path.join(fixtureDir, "release-watcher"); + const config = { + session: { mainKey: "main", store: path.join(fixtureDir, "sessions.json") }, + cron: { enabled: true, store: path.join(fixtureDir, "jobs.json") }, + } as OpenClawConfig; + await writeFile( + childScriptPath, + "const fs=require('node:fs');" + + "fs.appendFileSync(process.argv[2],'run\\n');" + + "const timer=setInterval(()=>{if(fs.existsSync(process.argv[3]))clearInterval(timer)},10)", + "utf8", + ); + const childArgs = [childScriptPath, markerPath, releasePath]; + const command = + process.platform === "win32" + ? buildWindowsCmdExeCommandLine(process.execPath, childArgs) + : [process.execPath, ...childArgs].map((argument) => JSON.stringify(argument)).join(" "); + const supervisor = getProcessSupervisor(); + const spawn = vi.spyOn(supervisor, "spawn"); + let state: ReturnType | undefined; + + vi.stubEnv("OPENCLAW_STATE_DIR", fixtureDir); + vi.stubEnv("OPENCLAW_SKIP_CRON", "0"); + hoisted.runtimeConfig.value = config; + setRuntimeConfigSnapshot(config, config); + + try { + const actualCron = + await vi.importActual("./server-cron.js"); + const initialCronState = actualCron.buildGatewayCronService({ + cfg: config, + deps: {} as never, + broadcast: vi.fn(), + }); + state = { + ...createDefaultGatewayReloadState(), + cronState: initialCronState, + }; + await initialCronState.cron.start(); + const job = await initialCronState.cron.add({ + name: "preserve the real watched child", + enabled: true, + schedule: { kind: "on-exit", command }, + sessionTarget: "main", + wakeMode: "next-heartbeat", + payload: { kind: "systemEvent", text: "watched child finished" }, + }); + await initialCronState.reconcileExitWatchers(); + await waitForFast(async () => expect(await readFile(markerPath, "utf8")).toBe("run\n"), { + timeout: 10_000, + }); + expect(spawn).toHaveBeenCalledOnce(); + const watchedRun = await spawn.mock.results[0]?.value; + if (!watchedRun) { + throw new Error("expected the supervised cron exit watcher to start"); + } + + hoisted.buildGatewayCronService.mockImplementationOnce( + (params) => + actualCron.buildGatewayCronService( + params as Parameters[0], + ) as unknown as ReturnType, + ); + const handlers = createGatewayReloadHandlers({ + getState: () => { + if (!state) { + throw new Error("expected gateway state"); + } + return state; + }, + setState: (nextState) => { + state = nextState; + }, + }); + + await withGatewayRestartSignal(async () => { + await handlers.applyHotReload(createCronRestartPlan(), config); + }); + + expect(supervisor.getRecord(watchedRun.runId)).toMatchObject({ + pid: watchedRun.pid, + state: "running", + }); + expect(await readFile(markerPath, "utf8")).toBe("run\n"); + expect(spawn).toHaveBeenCalledOnce(); + + await writeFile(releasePath, "release"); + await waitForFast(() => expect(state?.cronState.cron.getJob(job.id)?.enabled).toBe(false), { + timeout: 10_000, + }); + expect(await readFile(markerPath, "utf8")).toBe("run\n"); + expect(spawn).toHaveBeenCalledOnce(); + } finally { + await writeFile(releasePath, "release").catch(() => {}); + await state?.cronState.cron.stopAndDrain?.(); + spawn.mockRestore(); + vi.unstubAllEnvs(); + } + }); + it.each([ "agents.defaults.compaction.model", "agents.defaults.compaction.maxActiveTranscriptBytes", diff --git a/src/gateway/server-reload-hot.ts b/src/gateway/server-reload-hot.ts index a12022f38e2d..567be4a7bc35 100644 --- a/src/gateway/server-reload-hot.ts +++ b/src/gateway/server-reload-hot.ts @@ -20,7 +20,7 @@ import { } from "./config-reload-recovery.js"; import type { GatewayReloadPlan } from "./config-reload.js"; import { commitHooksConfigReload, resolveHooksConfig } from "./hooks.js"; -import { buildGatewayCronService } from "./server-cron.js"; +import { buildGatewayCronService, type GatewayCronExitWatcherHandoff } from "./server-cron.js"; import { applyGatewayLaneConcurrency, resolveGatewayLaneConcurrency } from "./server-lanes.js"; import { createGatewayActiveWorkTracker } from "./server-reload-active-work.js"; import { @@ -114,6 +114,9 @@ export function createGatewayReloadHandlers(params: GatewayReloadHandlerParams) } nextState.hookClientIpConfig = resolveHookClientIpConfig(nextConfig); + let cronExitWatcherHandoff: + | { previous: GatewayCronExitWatcherHandoff; next: GatewayCronExitWatcherHandoff } + | undefined; if (plan.restartCron) { nextState.cronState = buildGatewayCronService({ cfg: nextConfig, @@ -126,6 +129,19 @@ export function createGatewayReloadHandlers(params: GatewayReloadHandlerParams) ? { resolveGatewayContext: params.resolveGatewayContext } : {}), }); + if ( + state.cronState.cronEnabled && + nextState.cronState.cronEnabled && + state.cronState.storePath === nextState.cronState.storePath + ) { + const [previous, next] = await Promise.all([ + state.cronState.prepareExitWatcherHandoff?.(), + nextState.cronState.prepareExitWatcherHandoff?.(), + ]); + if (previous && next) { + cronExitWatcherHandoff = { previous, next }; + } + } } resetDirectoryCache(); @@ -201,7 +217,10 @@ export function createGatewayReloadHandlers(params: GatewayReloadHandlerParams) if (plan.restartCron) { params.cronReconciliation.invalidate(); params.onCronRestart?.(); - if (state.cronState.cron.stopAndDrain) { + if (cronExitWatcherHandoff) { + await cronExitWatcherHandoff.next.adopt(cronExitWatcherHandoff.previous.current()); + await cronExitWatcherHandoff.previous.stopOwner(); + } else if (state.cronState.cron.stopAndDrain) { await state.cronState.cron.stopAndDrain(); } else { state.cronState.cron.stop();