@@ -53,6 +53,7 @@ const MIN_REFIRE_GAP_MS = 2_000;
53535454const DEFAULT_MISSED_JOB_STAGGER_MS = 5_000;
5555const DEFAULT_MAX_MISSED_JOBS_PER_RESTART = 5;
56+const DEFAULT_STARTUP_DEFERRED_MISSED_AGENT_JOB_DELAY_MS = 2 * 60_000;
5657const DEFAULT_FAILURE_ALERT_AFTER = 2;
5758const DEFAULT_FAILURE_ALERT_COOLDOWN_MS = 60 * 60_000; // 1 hour
5859@@ -82,9 +83,14 @@ type StartupCatchupCandidate = {
8283job: CronJob;
8384};
848586+type StartupDeferredJob = {
87+jobId: string;
88+delayMs?: number;
89+};
90+8591type StartupCatchupPlan = {
8692candidates: StartupCatchupCandidate[];
87-deferredJobIds: string[];
93+deferredJobs: StartupDeferredJob[];
8894};
89959096export async function executeJobCoreWithTimeout(
@@ -1038,10 +1044,10 @@ function collectRunnableJobs(
1038104410391045export async function runMissedJobs(
10401046state: CronServiceState,
1041-opts?: { skipJobIds?: ReadonlySet<string> },
1047+opts?: { skipJobIds?: ReadonlySet<string>; deferAgentTurnJobs?: boolean },
10421048) {
10431049const plan = await planStartupCatchup(state, opts);
1044-if (plan.candidates.length === 0 && plan.deferredJobIds.length === 0) {
1050+if (plan.candidates.length === 0 && plan.deferredJobs.length === 0) {
10451051return;
10461052}
10471053@@ -1051,7 +1057,7 @@ export async function runMissedJobs(
1051105710521058async function planStartupCatchup(
10531059state: CronServiceState,
1054-opts?: { skipJobIds?: ReadonlySet<string> },
1060+opts?: { skipJobIds?: ReadonlySet<string>; deferAgentTurnJobs?: boolean },
10551061): Promise<StartupCatchupPlan> {
10561062const maxImmediate = Math.max(
105710630,
@@ -1060,7 +1066,7 @@ async function planStartupCatchup(
10601066return locked(state, async () => {
10611067await ensureLoaded(state, { skipRecompute: true });
10621068if (!state.store) {
1063-return { candidates: [], deferredJobIds: [] };
1069+return { candidates: [], deferredJobs: [] };
10641070}
1065107110661072const now = state.deps.nowMs();
@@ -1070,13 +1076,28 @@ async function planStartupCatchup(
10701076allowCronMissedRunByLastRun: true,
10711077});
10721078if (missed.length === 0) {
1073-return { candidates: [], deferredJobIds: [] };
1079+return { candidates: [], deferredJobs: [] };
10741080}
10751081const sorted = missed.toSorted(
10761082(a, b) => (a.state.nextRunAtMs ?? 0) - (b.state.nextRunAtMs ?? 0),
10771083);
1078-const startupCandidates = sorted.slice(0, maxImmediate);
1079-const deferred = sorted.slice(maxImmediate);
1084+const deferredAgentJobs = opts?.deferAgentTurnJobs
1085+ ? sorted.filter((job) => job.payload.kind === "agentTurn")
1086+ : [];
1087+const startupEligible = opts?.deferAgentTurnJobs
1088+ ? sorted.filter((job) => job.payload.kind !== "agentTurn")
1089+ : sorted;
1090+const startupCandidates = startupEligible.slice(0, maxImmediate);
1091+const deferredOverflow = startupEligible.slice(maxImmediate);
1092+const deferredAgentDelayMs = Math.max(
1093+0,
1094+state.deps.startupDeferredMissedAgentJobDelayMs ??
1095+DEFAULT_STARTUP_DEFERRED_MISSED_AGENT_JOB_DELAY_MS,
1096+);
1097+const deferred: StartupDeferredJob[] = [
1098+ ...deferredOverflow.map((job) => ({ jobId: job.id })),
1099+ ...deferredAgentJobs.map((job) => ({ jobId: job.id, delayMs: deferredAgentDelayMs })),
1100+];
10801101if (deferred.length > 0) {
10811102state.deps.log.info(
10821103{
@@ -1087,6 +1108,16 @@ async function planStartupCatchup(
10871108"cron: staggering missed jobs to prevent gateway overload",
10881109);
10891110}
1111+if (deferredAgentJobs.length > 0) {
1112+state.deps.log.info(
1113+{
1114+count: deferredAgentJobs.length,
1115+jobIds: deferredAgentJobs.map((job) => job.id),
1116+delayMs: deferredAgentDelayMs,
1117+},
1118+"cron: deferring missed agent jobs until after gateway startup",
1119+);
1120+}
10901121if (startupCandidates.length > 0) {
10911122state.deps.log.info(
10921123{ count: startupCandidates.length, jobIds: startupCandidates.map((j) => j.id) },
@@ -1101,7 +1132,7 @@ async function planStartupCatchup(
1101113211021133return {
11031134candidates: startupCandidates.map((job) => ({ jobId: job.id, job })),
1104-deferredJobIds: deferred.map((job) => job.id),
1135+deferredJobs: deferred,
11051136};
11061137});
11071138}
@@ -1182,14 +1213,20 @@ async function applyStartupCatchupOutcomes(
11821213applyOutcomeToStoredJob(state, result);
11831214}
118412151185-if (plan.deferredJobIds.length > 0) {
1216+if (plan.deferredJobs.length > 0) {
11861217const baseNow = state.deps.nowMs();
11871218let offset = staggerMs;
1188-for (const jobId of plan.deferredJobIds) {
1219+for (const deferred of plan.deferredJobs) {
1220+const jobId = deferred.jobId;
11891221const job = state.store.jobs.find((entry) => entry.id === jobId);
11901222if (!job || !isJobEnabled(job)) {
11911223continue;
11921224}
1225+if (typeof deferred.delayMs === "number") {
1226+job.state.nextRunAtMs = baseNow + deferred.delayMs + offset - staggerMs;
1227+offset += staggerMs;
1228+continue;
1229+}
11931230job.state.nextRunAtMs = baseNow + offset;
11941231offset += staggerMs;
11951232}
此内容由惯性聚合(RSS阅读器)自动聚合整理，仅供阅读参考。 原文来自 — 版权归原作者所有。