github/openclaw
1
0
Fork 0
mirror of https://github.com/openclaw/openclaw.git synced 2026-05-09 22:04:30 +00:00
Code Issues Packages Projects Releases Wiki Activity

refactor(cron): share job tick state normalization

Browse Source
This commit is contained in:
Peter Steinberger
2026-02-14 21:44:30 +00:00
parent 6aab89939f
commit 6b400eca5c
1 changed files with 43 additions and 40 deletions
Download Patch File Download Diff File Expand all files Collapse all files

59
src/cron/service/jobs.ts
View File

@@ -90,17 +90,18 @@ export function computeJobNextRunAtMs(job: CronJob, nowMs: number): number | und
/** Maximum consecutive schedule errors before auto-disabling a job. */ /** Maximum consecutive schedule errors before auto-disabling a job. */
const MAX_SCHEDULE_ERRORS = 3; const MAX_SCHEDULE_ERRORS = 3;
export function recomputeNextRuns(state: CronServiceState): boolean { function normalizeJobTickState(params: { state: CronServiceState; job: CronJob; nowMs: number }): {
if (!state.store) { changed: boolean;
return false; skip: boolean;
} } {
const { state, job, nowMs } = params;
let changed = false; let changed = false;
const now = state.deps.nowMs();
for (const job of state.store.jobs) {
if (!job.state) { if (!job.state) {
job.state = {}; job.state = {};
changed = true; changed = true;
} }
if (!job.enabled) { if (!job.enabled) {
if (job.state.nextRunAtMs !== undefined) { if (job.state.nextRunAtMs !== undefined) {
job.state.nextRunAtMs = undefined; job.state.nextRunAtMs = undefined;
@@ -110,10 +111,11 @@ export function recomputeNextRuns(state: CronServiceState): boolean {
job.state.runningAtMs = undefined; job.state.runningAtMs = undefined;
changed = true; changed = true;
} }
continue; return { changed, skip: true };
} }
const runningAt = job.state.runningAtMs; const runningAt = job.state.runningAtMs;
if (typeof runningAt === "number" && now - runningAt > STUCK_RUN_MS) { if (typeof runningAt === "number" && nowMs - runningAt > STUCK_RUN_MS) {
state.deps.log.warn( state.deps.log.warn(
{ jobId: job.id, runningAtMs: runningAt }, { jobId: job.id, runningAtMs: runningAt },
"cron: clearing stuck running marker", "cron: clearing stuck running marker",
@@ -121,6 +123,24 @@ export function recomputeNextRuns(state: CronServiceState): boolean {
job.state.runningAtMs = undefined; job.state.runningAtMs = undefined;
changed = true; changed = true;
} }
return { changed, skip: false };
}
export function recomputeNextRuns(state: CronServiceState): boolean {
if (!state.store) {
return false;
}
let changed = false;
const now = state.deps.nowMs();
for (const job of state.store.jobs) {
const tick = normalizeJobTickState({ state, job, nowMs: now });
if (tick.changed) {
changed = true;
}
if (tick.skip) {
continue;
}
// Only recompute if nextRunAtMs is missing or already past-due. // Only recompute if nextRunAtMs is missing or already past-due.
// Preserving a still-future nextRunAtMs avoids accidentally advancing // Preserving a still-future nextRunAtMs avoids accidentally advancing
// a job that hasn't fired yet (e.g. during restart recovery). // a job that hasn't fired yet (e.g. during restart recovery).
@@ -177,30 +197,13 @@ export function recomputeNextRunsForMaintenance(state: CronServiceState): boolea
let changed = false; let changed = false;
const now = state.deps.nowMs(); const now = state.deps.nowMs();
for (const job of state.store.jobs) { for (const job of state.store.jobs) {
if (!job.state) { const tick = normalizeJobTickState({ state, job, nowMs: now });
job.state = {}; if (tick.changed) {
changed = true;
}
if (!job.enabled) {
if (job.state.nextRunAtMs !== undefined) {
job.state.nextRunAtMs = undefined;
changed = true;
}
if (job.state.runningAtMs !== undefined) {
job.state.runningAtMs = undefined;
changed = true; changed = true;
} }
if (tick.skip) {
continue; continue;
} }
const runningAt = job.state.runningAtMs;
if (typeof runningAt === "number" && now - runningAt > STUCK_RUN_MS) {
state.deps.log.warn(
{ jobId: job.id, runningAtMs: runningAt },
"cron: clearing stuck running marker",
);
job.state.runningAtMs = undefined;
changed = true;
}
// Only compute missing nextRunAtMs, do NOT recompute existing ones. // Only compute missing nextRunAtMs, do NOT recompute existing ones.
// If a job was past-due but not found by findDueJobs, recomputing would // If a job was past-due but not found by findDueJobs, recomputing would
// cause it to be silently skipped. // cause it to be silently skipped.