mirror of
https://github.com/openclaw/openclaw.git
synced 2026-07-23 14:01:12 +00:00
fix(cron): skip completed restart catchup slots (#101998)
* fix(cron): preserve exact restart boundaries Co-authored-by: 李兰 0668001394 <li.lan3@xydigit.com> Co-authored-by: NianJiuZst <180004567+NianJiuZst@users.noreply.github.com> * fix(cron): suppress skipped restart slots --------- Co-authored-by: Peter Steinberger <steipete@gmail.com> Co-authored-by: NianJiuZst <180004567+NianJiuZst@users.noreply.github.com>
This commit is contained in:
@@ -2,6 +2,8 @@
|
||||
import { describe, expect, it } from "vitest";
|
||||
import {
|
||||
applyJobPatch,
|
||||
computeJobNextRunAtMs,
|
||||
computeJobPreviousRunAtOrBeforeMs,
|
||||
createJob,
|
||||
nextWakeAtMs,
|
||||
recomputeNextRuns,
|
||||
@@ -1094,6 +1096,45 @@ describe("cron stagger defaults", () => {
|
||||
});
|
||||
});
|
||||
|
||||
describe("computeJobPreviousRunAtOrBeforeMs", () => {
|
||||
function createCronJob(schedule: Extract<CronJob["schedule"], { kind: "cron" }>): CronJob {
|
||||
return {
|
||||
id: "inclusive-previous-run",
|
||||
name: "inclusive previous run",
|
||||
enabled: true,
|
||||
createdAtMs: 0,
|
||||
updatedAtMs: 0,
|
||||
schedule,
|
||||
sessionTarget: "main",
|
||||
wakeMode: "now",
|
||||
payload: { kind: "systemEvent", text: "tick" },
|
||||
state: {},
|
||||
};
|
||||
}
|
||||
|
||||
it("includes an exact boundary and keeps the prior slot between boundaries", () => {
|
||||
const job = createCronJob({ kind: "cron", expr: "* * * * * *", tz: "UTC", staggerMs: 0 });
|
||||
const boundary = Date.parse("2025-12-13T04:02:00.000Z");
|
||||
|
||||
expect(computeJobPreviousRunAtOrBeforeMs(job, boundary)).toBe(boundary);
|
||||
expect(computeJobPreviousRunAtOrBeforeMs(job, boundary + 500)).toBe(boundary);
|
||||
});
|
||||
|
||||
it("includes an exact effective boundary after per-job staggering", () => {
|
||||
const job = createCronJob({
|
||||
kind: "cron",
|
||||
expr: "0 * * * * *",
|
||||
tz: "UTC",
|
||||
staggerMs: 30_000,
|
||||
});
|
||||
const cursor = Date.parse("2025-12-13T04:02:00.000Z");
|
||||
const effectiveBoundary = computeJobNextRunAtMs(job, cursor);
|
||||
|
||||
expect(effectiveBoundary).toBeTypeOf("number");
|
||||
expect(computeJobPreviousRunAtOrBeforeMs(job, effectiveBoundary!)).toBe(effectiveBoundary);
|
||||
});
|
||||
});
|
||||
|
||||
describe("createJob delivery defaults", () => {
|
||||
const now = Date.parse("2026-02-28T12:00:00.000Z");
|
||||
|
||||
|
||||
@@ -270,6 +270,121 @@ describe("CronService restart catch-up", () => {
|
||||
}
|
||||
});
|
||||
|
||||
it.each(["ok", "skipped"] as const)(
|
||||
"does not defer an isolated cron agent-turn whose persisted due slot finished as %s",
|
||||
async (lastRunStatus) => {
|
||||
const store = await makeStorePath();
|
||||
const startNow = Date.parse("2025-12-13T11:00:00.000Z");
|
||||
const dueAt = Date.parse("2025-12-13T09:10:00.000Z");
|
||||
const completedAt = Date.parse("2025-12-13T09:10:30.000Z");
|
||||
const runIsolatedAgentJob = vi.fn(async () => ({ status: "ok" as const }));
|
||||
const enqueueSystemEvent = vi.fn();
|
||||
const requestHeartbeat = vi.fn();
|
||||
|
||||
await writeStoreJobs(store.storePath, [
|
||||
{
|
||||
id: `startup-isolated-agent-already-${lastRunStatus}`,
|
||||
name: `startup isolated agent already ${lastRunStatus}`,
|
||||
enabled: true,
|
||||
createdAtMs: Date.parse("2025-12-10T12:00:00.000Z"),
|
||||
updatedAtMs: completedAt,
|
||||
schedule: { kind: "cron", expr: "10 9 * * *", tz: "UTC" },
|
||||
sessionTarget: "isolated",
|
||||
wakeMode: "next-heartbeat",
|
||||
payload: { kind: "agentTurn", message: "daily reminder" },
|
||||
state: {
|
||||
nextRunAtMs: dueAt,
|
||||
lastRunAtMs: completedAt,
|
||||
lastRunStatus,
|
||||
},
|
||||
},
|
||||
]);
|
||||
|
||||
const cron = createRestartCronService({
|
||||
storePath: store.storePath,
|
||||
enqueueSystemEvent,
|
||||
requestHeartbeat,
|
||||
runIsolatedAgentJob,
|
||||
nowMs: () => startNow,
|
||||
startupDeferredMissedAgentJobDelayMs: 120_000,
|
||||
});
|
||||
|
||||
try {
|
||||
await cron.start();
|
||||
|
||||
expect(runIsolatedAgentJob).not.toHaveBeenCalled();
|
||||
expect(enqueueSystemEvent).not.toHaveBeenCalled();
|
||||
expect(requestHeartbeat).not.toHaveBeenCalled();
|
||||
|
||||
const listedJobs = await cron.list({ includeDisabled: true });
|
||||
const updated = listedJobs.find(
|
||||
(job) => job.id === `startup-isolated-agent-already-${lastRunStatus}`,
|
||||
);
|
||||
expect(updated?.state.lastRunStatus).toBe(lastRunStatus);
|
||||
expect(updated?.state.nextRunAtMs).toBe(Date.parse("2025-12-14T09:10:00.000Z"));
|
||||
} finally {
|
||||
cron.stop();
|
||||
await store.cleanup();
|
||||
}
|
||||
},
|
||||
);
|
||||
|
||||
it("replays a newer missed cron slot behind a completed persisted slot", async () => {
|
||||
vi.setSystemTime(new Date("2025-12-13T04:10:00.000Z"));
|
||||
await withRestartedCron(
|
||||
[
|
||||
{
|
||||
id: "restart-completed-slot-newer-miss",
|
||||
name: "completed slot with newer miss",
|
||||
enabled: true,
|
||||
createdAtMs: Date.parse("2025-12-10T12:00:00.000Z"),
|
||||
updatedAtMs: Date.parse("2025-12-13T04:01:30.000Z"),
|
||||
schedule: { kind: "cron", expr: "* * * * *", tz: "UTC" },
|
||||
sessionTarget: "main",
|
||||
wakeMode: "next-heartbeat",
|
||||
payload: { kind: "systemEvent", text: "newer slot missed" },
|
||||
state: {
|
||||
nextRunAtMs: Date.parse("2025-12-13T04:01:00.000Z"),
|
||||
lastRunAtMs: Date.parse("2025-12-13T04:01:00.000Z"),
|
||||
lastRunStatus: "ok",
|
||||
},
|
||||
},
|
||||
],
|
||||
async ({ enqueueSystemEvent, requestHeartbeat }) => {
|
||||
expectQueuedSystemEvent(enqueueSystemEvent, "newer slot missed");
|
||||
expect(requestHeartbeat).toHaveBeenCalled();
|
||||
},
|
||||
);
|
||||
});
|
||||
|
||||
it("replays a cron slot due exactly at restart behind a completed persisted slot", async () => {
|
||||
vi.setSystemTime(new Date("2025-12-13T04:02:00.000Z"));
|
||||
await withRestartedCron(
|
||||
[
|
||||
{
|
||||
id: "restart-completed-slot-boundary-miss",
|
||||
name: "completed slot with boundary miss",
|
||||
enabled: true,
|
||||
createdAtMs: Date.parse("2025-12-10T12:00:00.000Z"),
|
||||
updatedAtMs: Date.parse("2025-12-13T04:01:30.000Z"),
|
||||
schedule: { kind: "cron", expr: "* * * * *", tz: "UTC" },
|
||||
sessionTarget: "main",
|
||||
wakeMode: "next-heartbeat",
|
||||
payload: { kind: "systemEvent", text: "boundary slot missed" },
|
||||
state: {
|
||||
nextRunAtMs: Date.parse("2025-12-13T04:01:00.000Z"),
|
||||
lastRunAtMs: Date.parse("2025-12-13T04:01:00.000Z"),
|
||||
lastRunStatus: "ok",
|
||||
},
|
||||
},
|
||||
],
|
||||
async ({ enqueueSystemEvent, requestHeartbeat }) => {
|
||||
expectQueuedSystemEvent(enqueueSystemEvent, "boundary slot missed");
|
||||
expect(requestHeartbeat).toHaveBeenCalled();
|
||||
},
|
||||
);
|
||||
});
|
||||
|
||||
it("marks interrupted recurring jobs failed instead of replaying them on startup", async () => {
|
||||
const dueAt = Date.parse("2025-12-13T16:00:00.000Z");
|
||||
const staleRunningAt = Date.parse("2025-12-13T16:30:00.000Z");
|
||||
|
||||
@@ -181,17 +181,32 @@ function computeStaggeredCronPreviousRunAtMs(job: CronJob, nowMs: number) {
|
||||
return undefined;
|
||||
}
|
||||
|
||||
function computeStaggeredCronPreviousRunAtOrBeforeMs(job: CronJob, nowMs: number) {
|
||||
const previous = computeStaggeredCronPreviousRunAtMs(job, nowMs);
|
||||
const probeMs = nowMs + 1_000;
|
||||
if (!Number.isFinite(probeMs)) {
|
||||
return previous;
|
||||
}
|
||||
|
||||
// Croner previous-run queries are strict-before and second-granular. Keep
|
||||
// the strict result, then probe past the current second to include a slot
|
||||
// exactly at now without losing the prior slot between boundaries.
|
||||
const boundary = computeStaggeredCronPreviousRunAtMs(job, probeMs);
|
||||
if (
|
||||
isFiniteTimestamp(boundary) &&
|
||||
boundary <= nowMs &&
|
||||
(!isFiniteTimestamp(previous) || boundary > previous)
|
||||
) {
|
||||
return boundary;
|
||||
}
|
||||
return previous;
|
||||
}
|
||||
|
||||
function isStaggeredCronRunAtMs(job: CronJob, runAtMs: number): boolean {
|
||||
if (job.schedule.kind !== "cron" || !isFiniteTimestamp(runAtMs)) {
|
||||
return false;
|
||||
}
|
||||
// Probe past the candidate second. Croner-style second-granular schedules
|
||||
// normalize a 1ms probe back to the candidate's second, so
|
||||
// `previousRuns(1, runAtMs + 1)` returns the slot before the candidate
|
||||
// rather than the candidate itself and exact-second slots get misclassified
|
||||
// as stale. A 1s probe lands past the candidate second, matching the cursor
|
||||
// step used elsewhere in this file (cf. #81691).
|
||||
const previous = computeStaggeredCronPreviousRunAtMs(job, runAtMs + 1_000);
|
||||
const previous = computeStaggeredCronPreviousRunAtOrBeforeMs(job, runAtMs);
|
||||
return previous === runAtMs;
|
||||
}
|
||||
|
||||
@@ -510,6 +525,15 @@ export function computeJobPreviousRunAtMs(job: CronJob, nowMs: number): number |
|
||||
return isFiniteTimestamp(previous) ? previous : undefined;
|
||||
}
|
||||
|
||||
/** Computes the latest effective cron timestamp at or before the supplied time. */
|
||||
export function computeJobPreviousRunAtOrBeforeMs(job: CronJob, nowMs: number): number | undefined {
|
||||
if (!isJobEnabled(job) || job.schedule.kind !== "cron") {
|
||||
return undefined;
|
||||
}
|
||||
const previous = computeStaggeredCronPreviousRunAtOrBeforeMs(job, nowMs);
|
||||
return isFiniteTimestamp(previous) ? previous : undefined;
|
||||
}
|
||||
|
||||
/** Maximum consecutive schedule errors before auto-disabling a job. */
|
||||
const MAX_SCHEDULE_ERRORS = 3;
|
||||
|
||||
|
||||
@@ -69,6 +69,7 @@ import {
|
||||
} from "./failure-alerts.js";
|
||||
import {
|
||||
DEFAULT_ERROR_BACKOFF_SCHEDULE_MS,
|
||||
computeJobPreviousRunAtOrBeforeMs,
|
||||
computeJobPreviousRunAtMs,
|
||||
computeJobNextRunAtMs,
|
||||
errorBackoffMs,
|
||||
@@ -1833,7 +1834,26 @@ function isRunnableJob(params: {
|
||||
return false;
|
||||
}
|
||||
if (hasScheduledNextRunAtMs(next) && nowMs >= next) {
|
||||
return true;
|
||||
const lastRunAtMs = job.state.lastRunAtMs;
|
||||
// Startup loads persisted state before maintenance recompute. Suppress a
|
||||
// completed stale slot, but still replay a newer slot due by restart time.
|
||||
const alreadyCompletedDueCronSlot =
|
||||
params.allowCronMissedRunByLastRun &&
|
||||
job.schedule.kind === "cron" &&
|
||||
(lastRunStatus === "ok" || lastRunStatus === "skipped") &&
|
||||
typeof lastRunAtMs === "number" &&
|
||||
Number.isFinite(lastRunAtMs) &&
|
||||
lastRunAtMs >= next;
|
||||
if (!alreadyCompletedDueCronSlot) {
|
||||
return true;
|
||||
}
|
||||
let latestRunAtMs: number | undefined;
|
||||
try {
|
||||
latestRunAtMs = computeJobPreviousRunAtOrBeforeMs(job, nowMs);
|
||||
} catch {
|
||||
return false;
|
||||
}
|
||||
return typeof latestRunAtMs === "number" && latestRunAtMs > lastRunAtMs;
|
||||
}
|
||||
if (!params.allowCronMissedRunByLastRun || job.schedule.kind !== "cron") {
|
||||
return false;
|
||||
|
||||
Reference in New Issue
Block a user