Skip to content

Commit 372f85d

Browse files
shakkernerdshakkernerd
committed
fix: avoid cron cancel runtime cycle
1 parent 3cf9430 commit 372f85d

9 files changed

Lines changed: 87 additions & 161 deletions

‎src/cron/active-jobs.ts‎

Lines changed: 10 additions & 106 deletions
Original file line numberDiff line numberDiff line change
@@ -2,145 +2,49 @@
22
import { resolveGlobalSingleton } from "../shared/global-singleton.js";
33

44
type CronActiveJobState = {
5-
activeJobRuns: Map<string, Map<string, ActiveCronJobRun>>;
5+
activeJobIds: Set<string>;
66
};
77

88
const CRON_ACTIVE_JOB_STATE_KEY = Symbol.for("openclaw.cron.activeJobs");
9-
const DEFAULT_RUN_KEY = "__cron-job__";
10-
11-
type ActiveCronJobRun = {
12-
runId?: string;
13-
abortController?: AbortController;
14-
};
15-
16-
type MarkCronJobActiveOptions = {
17-
runId?: string;
18-
abortController?: AbortController;
19-
};
20-
21-
type CancelCronJobRunResult =
22-
| { found: false; cancelled: false; reason: string }
23-
| { found: true; cancelled: false; reason: string }
24-
| { found: true; cancelled: true };
259

2610
function getCronActiveJobState(): CronActiveJobState {
2711
// Cron runs can cross module reload boundaries in tests and dev watch; keep
28-
// the in-flight job map process-global so duplicate-run guards share state.
12+
// the in-flight job set process-global so duplicate-run guards share state.
2913
return resolveGlobalSingleton<CronActiveJobState>(CRON_ACTIVE_JOB_STATE_KEY, () => ({
30-
activeJobRuns: new Map<string, Map<string, ActiveCronJobRun>>(),
14+
activeJobIds: new Set<string>(),
3115
}));
3216
}
3317

34-
function normalizeRunKey(runId: string | undefined): string {
35-
return runId?.trim() || DEFAULT_RUN_KEY;
36-
}
37-
3818
/** Marks a cron job id as currently executing for duplicate-run suppression. */
39-
export function markCronJobActive(jobId: string, opts?: MarkCronJobActiveOptions) {
19+
export function markCronJobActive(jobId: string) {
4020
if (!jobId) {
4121
return;
4222
}
43-
const state = getCronActiveJobState();
44-
const runs = state.activeJobRuns.get(jobId) ?? new Map<string, ActiveCronJobRun>();
45-
runs.set(normalizeRunKey(opts?.runId), {
46-
...(opts?.runId ? { runId: opts.runId } : {}),
47-
...(opts?.abortController ? { abortController: opts.abortController } : {}),
48-
});
49-
state.activeJobRuns.set(jobId, runs);
23+
getCronActiveJobState().activeJobIds.add(jobId);
5024
}
5125

5226
/** Clears the active marker when a cron run exits or is abandoned. */
53-
export function clearCronJobActive(jobId: string, runId?: string) {
27+
export function clearCronJobActive(jobId: string) {
5428
if (!jobId) {
5529
return;
5630
}
57-
const state = getCronActiveJobState();
58-
if (runId === undefined) {
59-
state.activeJobRuns.delete(jobId);
60-
return;
61-
}
62-
const runs = state.activeJobRuns.get(jobId);
63-
if (!runs) {
64-
return;
65-
}
66-
runs.delete(normalizeRunKey(runId));
67-
if (runs.size === 0) {
68-
state.activeJobRuns.delete(jobId);
69-
}
31+
getCronActiveJobState().activeJobIds.delete(jobId);
7032
}
7133

7234
/** Returns whether the given cron job id is currently executing in this process. */
7335
export function isCronJobActive(jobId: string) {
7436
if (!jobId) {
7537
return false;
7638
}
77-
return (getCronActiveJobState().activeJobRuns.get(jobId)?.size ?? 0) > 0;
39+
return getCronActiveJobState().activeJobIds.has(jobId);
7840
}
7941

8042
/** Returns whether any cron run is active in this process. */
8143
export function hasActiveCronJobs() {
82-
return getCronActiveJobState().activeJobRuns.size > 0;
83-
}
84-
85-
/** Aborts an active cron run in the current process when one owns the task row. */
86-
export function cancelCronJobRun(params: {
87-
jobId?: string;
88-
runId?: string;
89-
reason?: string;
90-
}): CancelCronJobRunResult {
91-
const jobId = params.jobId?.trim();
92-
if (!jobId) {
93-
return {
94-
found: false,
95-
cancelled: false,
96-
reason: "Cron task has no cancellable job id.",
97-
};
98-
}
99-
const runs = getCronActiveJobState().activeJobRuns.get(jobId);
100-
if (!runs || runs.size === 0) {
101-
return {
102-
found: false,
103-
cancelled: false,
104-
reason: "Cron task is not active in this gateway process.",
105-
};
106-
}
107-
let run: ActiveCronJobRun | undefined;
108-
if (params.runId) {
109-
run = runs.get(normalizeRunKey(params.runId));
110-
} else {
111-
const first = runs.values().next();
112-
run = first.done ? undefined : first.value;
113-
}
114-
if (!run) {
115-
return {
116-
found: false,
117-
cancelled: false,
118-
reason: "Cron task run is not active in this gateway process.",
119-
};
120-
}
121-
const controller = run.abortController;
122-
if (!controller) {
123-
return {
124-
found: true,
125-
cancelled: false,
126-
reason: "Cron task has no active cancellation handle.",
127-
};
128-
}
129-
if (controller.signal.aborted) {
130-
return {
131-
found: true,
132-
cancelled: false,
133-
reason: "Cron task is already cancelling.",
134-
};
135-
}
136-
controller.abort(params.reason?.trim() || "Cancelled by operator.");
137-
return {
138-
found: true,
139-
cancelled: true,
140-
};
44+
return getCronActiveJobState().activeJobIds.size > 0;
14145
}
14246

14347
/** Clears process-global cron active-job state between tests. */
14448
export function resetCronActiveJobsForTests() {
145-
getCronActiveJobState().activeJobRuns.clear();
49+
getCronActiveJobState().activeJobIds.clear();
14650
}

‎src/cron/service/timer.regression.test.ts‎

Lines changed: 7 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -11,14 +11,16 @@ import {
1111
setupCronRegressionFixtures,
1212
} from "../../../test/helpers/cron/service-regression-fixtures.js";
1313
import { HEARTBEAT_SKIP_LANES_BUSY, type HeartbeatRunResult } from "../../infra/heartbeat-wake.js";
14+
import {
15+
cancelActiveCronTaskRun,
16+
resetActiveCronTaskRunsForTests,
17+
} from "../../tasks/cron-task-cancel.js";
1418
import {
1519
cancelTaskById,
1620
listTaskRecords,
1721
resetTaskRegistryControlRuntimeForTests,
1822
resetTaskRegistryForTests,
19-
setTaskRegistryControlRuntimeForTests,
2023
} from "../../tasks/task-registry.js";
21-
import { cancelCronJobRun } from "../active-jobs.js";
2224
import * as schedule from "../schedule.js";
2325
import { loadCronStore, saveCronStore } from "../store.js";
2426
import type {
@@ -889,13 +891,12 @@ describe("cron service timer regressions", () => {
889891
const timerPromise = onTimer(state);
890892
const observedAbortSignal = await runnerStarted.promise;
891893
const runId = `cron:cancel-before-timeout:${scheduledAt}`;
892-
const cancelled = cancelCronJobRun({
893-
jobId: "cancel-before-timeout",
894+
const cancelled = cancelActiveCronTaskRun({
894895
runId,
895896
reason: "Cancelled by operator.",
896897
});
897898

898-
expect(cancelled).toEqual({ found: true, cancelled: true });
899+
expect(cancelled).toBe(true);
899900
expect(observedAbortSignal?.aborted).toBe(true);
900901

901902
await vi.advanceTimersByTimeAsync(Math.ceil(FAST_TIMEOUT_SECONDS * 1_000) + 10);
@@ -1131,13 +1132,6 @@ describe("cron service timer regressions", () => {
11311132
vi.useFakeTimers();
11321133
try {
11331134
resetTaskRegistryForTests();
1134-
setTaskRegistryControlRuntimeForTests({
1135-
getAcpSessionManager: () => ({
1136-
cancelSession: vi.fn(),
1137-
}),
1138-
killSubagentRunAdmin: vi.fn(),
1139-
cancelCronJobRun,
1140-
});
11411135

11421136
const store = timerRegressionFixtures.makeStorePath();
11431137
const scheduledAt = Date.parse("2026-02-15T13:00:00.000Z");
@@ -1227,6 +1221,7 @@ describe("cron service timer regressions", () => {
12271221
}),
12281222
);
12291223
} finally {
1224+
resetActiveCronTaskRunsForTests();
12301225
resetTaskRegistryControlRuntimeForTests();
12311226
resetTaskRegistryForTests();
12321227
vi.useRealTimers();

‎src/cron/service/timer.ts‎

Lines changed: 12 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -13,6 +13,7 @@ import {
1313
normalizeAgentId,
1414
resolveAgentIdFromSessionKey,
1515
} from "../../routing/session-key.js";
16+
import { registerActiveCronTaskRun } from "../../tasks/cron-task-cancel.js";
1617
import { deliveryContextFromSession } from "../../utils/delivery-context.shared.js";
1718
import type { DeliveryContext } from "../../utils/delivery-context.types.js";
1819
import { clearCronJobActive, markCronJobActive } from "../active-jobs.js";
@@ -124,14 +125,16 @@ export async function executeJobCoreWithTimeout(
124125
opts?: { runId?: string },
125126
): Promise<Awaited<ReturnType<typeof executeJobCore>>> {
126127
const runAbortController = new AbortController();
127-
markCronJobActive(job.id, {
128-
runId: opts?.runId,
129-
// Main-session cron jobs enqueue work into a downstream child session.
130-
// The cron wrapper does not own that queued run, so exposing its abort
131-
// signal would let task cancellation mark the ledger row cancelled while
132-
// the child session can continue running.
133-
...(job.sessionTarget !== "main" ? { abortController: runAbortController } : {}),
134-
});
128+
// Main-session cron jobs enqueue work into a downstream child session. The
129+
// cron wrapper does not own that queued run, so it must not expose a task
130+
// cancellation handle that could make the wrapper row lie about child state.
131+
const releaseCronTaskRun =
132+
job.sessionTarget !== "main"
133+
? registerActiveCronTaskRun({
134+
runId: opts?.runId,
135+
controller: runAbortController,
136+
})
137+
: undefined;
135138
const jobTimeoutMs = resolveCronJobTimeoutMs(job);
136139
try {
137140
if (typeof jobTimeoutMs !== "number") {
@@ -193,7 +196,7 @@ export async function executeJobCoreWithTimeout(
193196
watchdog.dispose();
194197
}
195198
} finally {
196-
clearCronJobActive(job.id, opts?.runId);
199+
releaseCronTaskRun?.();
197200
}
198201
}
199202

‎src/tasks/cron-task-cancel.ts‎

Lines changed: 45 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,45 @@
1+
// Process-local cancellation handles for live cron task runs.
2+
3+
type CronTaskCancelHandle = {
4+
controller: AbortController;
5+
};
6+
7+
const activeCronTaskRunsByRunId = new Map<string, CronTaskCancelHandle>();
8+
9+
export function registerActiveCronTaskRun(params: {
10+
runId: string | undefined;
11+
controller: AbortController;
12+
}): (() => void) | undefined {
13+
const runId = params.runId?.trim();
14+
if (!runId) {
15+
return undefined;
16+
}
17+
activeCronTaskRunsByRunId.set(runId, { controller: params.controller });
18+
return () => {
19+
if (activeCronTaskRunsByRunId.get(runId)?.controller === params.controller) {
20+
activeCronTaskRunsByRunId.delete(runId);
21+
}
22+
};
23+
}
24+
25+
export function cancelActiveCronTaskRun(params: {
26+
runId: string | undefined;
27+
reason?: string;
28+
}): boolean {
29+
const runId = params.runId?.trim();
30+
if (!runId) {
31+
return false;
32+
}
33+
const handle = activeCronTaskRunsByRunId.get(runId);
34+
if (!handle) {
35+
return false;
36+
}
37+
if (!handle.controller.signal.aborted) {
38+
handle.controller.abort(params.reason?.trim() || "Cancelled by operator.");
39+
}
40+
return true;
41+
}
42+
43+
export function resetActiveCronTaskRunsForTests(): void {
44+
activeCronTaskRunsByRunId.clear();
45+
}

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

Lines changed: 0 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -117,11 +117,6 @@ async function withTaskExecutorStateDir(run: (stateDir: string) => Promise<void>
117117
cancelSession: hoisted.cancelSessionMock,
118118
}),
119119
killSubagentRunAdmin: async (params) => hoisted.killSubagentRunAdminMock(params),
120-
cancelCronJobRun: () => ({
121-
found: false,
122-
cancelled: false,
123-
reason: "Cron task is not active in this gateway process.",
124-
}),
125120
});
126121
try {
127122
await run(stateDir);

‎src/tasks/task-registry-control.runtime.ts‎

Lines changed: 0 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,3 @@
11
// Runtime control seam for cancelling ACP sessions and subagent runs from task APIs.
22
export { getAcpSessionManager } from "../acp/control-plane/manager.js";
33
export { killSubagentRunAdmin } from "../agents/subagent-control.js";
4-
export { cancelCronJobRun } from "../cron/active-jobs.js";

‎src/tasks/task-registry-control.types.ts‎

Lines changed: 0 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -22,21 +22,9 @@ export type KillSubagentRunAdmin = (params: {
2222
sessionKey: string;
2323
}) => Promise<KillSubagentRunAdminResult>;
2424

25-
export type CancelCronJobRunResult =
26-
| { found: false; cancelled: false; reason: string }
27-
| { found: true; cancelled: false; reason: string }
28-
| { found: true; cancelled: true };
29-
30-
export type CancelCronJobRun = (params: {
31-
jobId?: string;
32-
runId?: string;
33-
reason?: string;
34-
}) => CancelCronJobRunResult;
35-
3625
export type TaskRegistryControlRuntime = {
3726
getAcpSessionManager: () => {
3827
cancelSession: CancelAcpSessionAdmin;
3928
};
4029
killSubagentRunAdmin: KillSubagentRunAdmin;
41-
cancelCronJobRun: CancelCronJobRun;
4230
};

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

Lines changed: 5 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -2,11 +2,7 @@
22
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
33
import type { AcpSessionStoreEntry } from "../acp/runtime/session-meta.js";
44
import { startAcpSpawnParentStreamRelay } from "../agents/acp-spawn-parent-stream.js";
5-
import {
6-
cancelCronJobRun,
7-
markCronJobActive,
8-
resetCronActiveJobsForTests,
9-
} from "../cron/active-jobs.js";
5+
import { resetCronActiveJobsForTests } from "../cron/active-jobs.js";
106
import {
117
emitAgentEvent,
128
registerAgentRunContext,
@@ -21,6 +17,7 @@ import { peekSystemEvents, resetSystemEventsForTest } from "../infra/system-even
2117
import type { ParsedAgentSessionKey } from "../routing/session-key.js";
2218
import { withTempDir } from "../test-helpers/temp-dir.js";
2319
import { withEnvAsync } from "../test-utils/env.js";
20+
import { registerActiveCronTaskRun, resetActiveCronTaskRunsForTests } from "./cron-task-cancel.js";
2421
import {
2522
createTaskFlowForTask as createTaskFlowForTaskOrNull,
2623
createManagedTaskFlow as createManagedTaskFlowOrNull,
@@ -466,7 +463,6 @@ describe("task-registry", () => {
466463
cancelSession: hoisted.cancelSessionMock,
467464
}),
468465
killSubagentRunAdmin: async (params) => hoisted.killSubagentRunAdminMock(params),
469-
cancelCronJobRun,
470466
});
471467
});
472468

@@ -476,6 +472,7 @@ describe("task-registry", () => {
476472
resetHeartbeatWakeStateForTests();
477473
resetAgentRunContextForTest();
478474
resetCronActiveJobsForTests();
475+
resetActiveCronTaskRunsForTests();
479476
resetTaskRegistryControlRuntimeForTests();
480477
resetTaskRegistryDeliveryRuntimeForTests();
481478
resetTaskRegistryMaintenanceRuntimeForTests();
@@ -3549,9 +3546,9 @@ describe("task-registry", () => {
35493546
if (!task) {
35503547
throw new Error("expected cron task");
35513548
}
3552-
markCronJobActive("nightly-gmail-sync", {
3549+
registerActiveCronTaskRun({
35533550
runId: "cron:nightly-gmail-sync:123",
3554-
abortController,
3551+
controller: abortController,
35553552
});
35563553

35573554
const result = await cancelTaskById({

0 commit comments

Comments
 (0)