@@ -57,6 +57,7 @@ import {
5757nextWakeAtMs,
5858recomputeNextRunsForMaintenance,
5959recordScheduleComputeError,
60+resolveJobErrorBackoffUntilMs,
6061resolveJobPayloadTextForMain,
6162} from "./jobs.js";
6263import { locked } from "./locked.js";
@@ -1467,6 +1468,7 @@ export async function onTimer(state: CronServiceState) {
14671468}
1468146914691470function isRunnableJob(params: {
1471+state: CronServiceState;
14701472job: CronJob;
14711473nowMs: number;
14721474skipJobIds?: ReadonlySet<string>;
@@ -1504,14 +1506,14 @@ function isRunnableJob(params: {
15041506return false;
15051507}
15061508const next = job.state.nextRunAtMs;
1509+if (isErrorBackoffPending(params.state, job, nowMs)) {
1510+// Error retry windows are anchored at run end; persisted start-based
1511+// retry timestamps from older state must not bypass active backoff.
1512+return false;
1513+}
15071514if (hasScheduledNextRunAtMs(next) && nowMs >= next) {
15081515return true;
15091516}
1510-if (hasScheduledNextRunAtMs(next) && next > nowMs && isErrorBackoffPending(job, nowMs)) {
1511-// Respect active retry backoff windows on restart, but allow missed-slot
1512-// replay once the backoff window has elapsed.
1513-return false;
1514-}
15151517if (!params.allowCronMissedRunByLastRun || job.schedule.kind !== "cron") {
15161518return false;
15171519}
@@ -1532,20 +1534,15 @@ function isRunnableJob(params: {
15321534return previousRunAtMs > lastRunAtMs;
15331535}
153415361535-function isErrorBackoffPending(job: CronJob, nowMs: number): boolean {
1537+function isErrorBackoffPending(state: CronServiceState, job: CronJob, nowMs: number): boolean {
15361538if (job.schedule.kind === "at" || job.state.lastStatus !== "error") {
15371539return false;
15381540}
1539-const lastRunAtMs = job.state.lastRunAtMs;
1540-if (typeof lastRunAtMs !== "number" || !Number.isFinite(lastRunAtMs)) {
1541-return false;
1542-}
1543-const consecutiveErrorsRaw = job.state.consecutiveErrors;
1544-const consecutiveErrors =
1545-typeof consecutiveErrorsRaw === "number" && Number.isFinite(consecutiveErrorsRaw)
1546- ? Math.max(1, Math.floor(consecutiveErrorsRaw))
1547- : 1;
1548-return nowMs < lastRunAtMs + errorBackoffMs(consecutiveErrors);
1541+const backoffUntilMs = resolveJobErrorBackoffUntilMs(
1542+job,
1543+state.deps.cronConfig?.retry?.backoffMs ?? DEFAULT_ERROR_BACKOFF_SCHEDULE_MS,
1544+);
1545+return backoffUntilMs !== undefined && nowMs < backoffUntilMs;
15491546}
1550154715511548function collectRunnableJobs(
@@ -1562,6 +1559,7 @@ function collectRunnableJobs(
15621559}
15631560return state.store.jobs.filter((job) =>
15641561isRunnableJob({
1562+ state,
15651563 job,
15661564 nowMs,
15671565skipJobIds: opts?.skipJobIds,
@@ -1571,6 +1569,55 @@ function collectRunnableJobs(
15711569);
15721570}
157315711572+function deferPendingBackoffMissedCronSlots(
1573+state: CronServiceState,
1574+nowMs: number,
1575+opts?: { skipJobIds?: ReadonlySet<string> },
1576+): boolean {
1577+if (!state.store) {
1578+return false;
1579+}
1580+let changed = false;
1581+for (const job of state.store.jobs) {
1582+if (
1583+!isJobEnabled(job) ||
1584+job.schedule.kind !== "cron" ||
1585+opts?.skipJobIds?.has(job.id) ||
1586+typeof job.state.runningAtMs === "number"
1587+) {
1588+continue;
1589+}
1590+const backoffUntilMs = resolveJobErrorBackoffUntilMs(
1591+job,
1592+state.deps.cronConfig?.retry?.backoffMs ?? DEFAULT_ERROR_BACKOFF_SCHEDULE_MS,
1593+);
1594+if (backoffUntilMs === undefined || nowMs >= backoffUntilMs) {
1595+continue;
1596+}
1597+let previousRunAtMs: number | undefined;
1598+try {
1599+previousRunAtMs = computeJobPreviousRunAtMs(job, nowMs);
1600+} catch {
1601+continue;
1602+}
1603+const lastRunAtMs = job.state.lastRunAtMs;
1604+if (
1605+typeof previousRunAtMs !== "number" ||
1606+!Number.isFinite(previousRunAtMs) ||
1607+typeof lastRunAtMs !== "number" ||
1608+!Number.isFinite(lastRunAtMs) ||
1609+previousRunAtMs <= lastRunAtMs
1610+) {
1611+continue;
1612+}
1613+if (job.state.nextRunAtMs !== backoffUntilMs) {
1614+job.state.nextRunAtMs = backoffUntilMs;
1615+changed = true;
1616+}
1617+}
1618+return changed;
1619+}
1620+15741621export async function runMissedJobs(
15751622state: CronServiceState,
15761623opts?: { skipJobIds?: ReadonlySet<string>; deferAgentTurnJobs?: boolean },
@@ -1599,12 +1646,18 @@ async function planStartupCatchup(
15991646}
1600164716011648const now = state.deps.nowMs();
1649+const deferredBackoffMissedSlot = deferPendingBackoffMissedCronSlots(state, now, {
1650+skipJobIds: opts?.skipJobIds,
1651+});
16021652const missed = collectRunnableJobs(state, now, {
16031653skipJobIds: opts?.skipJobIds,
16041654skipAtIfAlreadyRan: true,
16051655allowCronMissedRunByLastRun: true,
16061656});
16071657if (missed.length === 0) {
1658+if (deferredBackoffMissedSlot) {
1659+await persist(state);
1660+}
16081661return { candidates: [], deferredJobs: [] };
16091662}
16101663const sorted = missed.toSorted(
此内容由惯性聚合(RSS阅读器)自动聚合整理，仅供阅读参考。 原文来自 — 版权归原作者所有。