diff --git a/src/index.ts b/src/index.ts index e98ec8e532..2ed6eb1ab2 100644 --- a/src/index.ts +++ b/src/index.ts @@ -11,6 +11,7 @@ import { isGitHubBudgetBackgroundJob, queueSnapshotBacklog, queueSnapshotFromBinding, + scheduledEnqueueDelaySeconds, } from "./selfhost/queue-common"; import { isReviewExecutionJob, isSelfHostedReviewRuntime } from "./selfhost/review-runtime"; import type { JobMessage } from "./types"; @@ -186,5 +187,16 @@ async function enqueueScheduledJobs(env: Env, controller: ScheduledController): // set is byte-identical to today. if (selfHostedReviews && isRagEnabled(env)) jobs.push({ type: "rag-index-repo", requestedBy: "schedule" }); } - await Promise.all(jobs.map((job) => env.JOBS.send(job))); + // Phase-spread the enqueue (#1948): flushing every due job with run_after=now made the top-of-hour (and + // top-of-6h) tick fan out all the heavy per-repo maintenance parents in one instant, draining the shared REST + // bucket and tripping GitHub's secondary rate limit. Each job type gets a stable deterministic slot across the + // jitter window (the every-tick sweep/relay stay immediate); the enqueued SET is unchanged, only the timing. + await Promise.all( + jobs.map((job) => { + const delaySeconds = scheduledEnqueueDelaySeconds(job.type); + return delaySeconds > 0 + ? env.JOBS.send(job, { delaySeconds }) + : env.JOBS.send(job); + }), + ); } diff --git a/src/selfhost/queue-common.ts b/src/selfhost/queue-common.ts index c6e1b9603d..c681c0482c 100644 --- a/src/selfhost/queue-common.ts +++ b/src/selfhost/queue-common.ts @@ -17,6 +17,7 @@ import { extractPayloadType } from "./audit"; const DEFAULT_RATE_LIMIT_JITTER_MS = 5 * 60_000; const DEFAULT_STARTUP_JITTER_MS = 3 * 60_000; const DEFAULT_RECOVERY_JITTER_MS = 60_000; +const DEFAULT_SCHEDULED_ENQUEUE_JITTER_MS = 5 * 60_000; const DEFAULT_STARTUP_JITTER_MIN_JOBS = 8; const DEFAULT_PROCESSING_TIMEOUT_MS = 30 * 60_000; const DEFAULT_BACKGROUND_CONCURRENCY = 1; @@ -547,6 +548,33 @@ export function deterministicJitterMs(seed: string, maxJitterMs: number): number return Math.abs(h >>> 0) % (Math.floor(maxJitterMs) + 1); } +export function scheduledEnqueueJitterMs(): number { + return envDurationMs( + "SCHEDULED_ENQUEUE_JITTER_MS", + DEFAULT_SCHEDULED_ENQUEUE_JITTER_MS, + ); +} + +// The every-tick priority scheduled jobs enqueue immediately; the periodic maintenance jobs are deterministically +// phase-spread across the jitter window so a top-of-hour cron tick does not flush every heavy per-repo fan-out +// parent in the same instant (which drains the shared GitHub REST bucket and trips the secondary rate limit). The +// re-gate sweep and its Orb-relay retry run every ~2-min tick and drive timely merges/closes, so they stay +// immediate; everything else (the 30-min, hourly, and six-hourly maintenance set) is offset by a stable per-type +// slot. Deterministic (hash of the job type), so a type always lands in the same slot and the enqueued SET is +// unchanged — only the run_after timing is spread, and the per-repo children each parent fans out inherit that +// offset (their own index stagger is relative to when the parent runs). (#1948) +const IMMEDIATE_SCHEDULED_JOB_TYPES = new Set([ + "agent-regate-sweep", + "retry-orb-relay", +]); + +export function scheduledEnqueueDelaySeconds(jobType: string): number { + if (IMMEDIATE_SCHEDULED_JOB_TYPES.has(jobType)) return 0; + return Math.floor( + deterministicJitterMs(jobType, scheduledEnqueueJitterMs()) / 1000, + ); +} + export function jobCoalesceKey(payload: string): string | null { try { const message = JSON.parse(payload) as { diff --git a/test/unit/index.test.ts b/test/unit/index.test.ts index d4731924e7..3bc8c91859 100644 --- a/test/unit/index.test.ts +++ b/test/unit/index.test.ts @@ -1,6 +1,7 @@ import { afterEach, describe, expect, it, vi } from "vitest"; import worker from "../../src/index"; import { recordGitHubRateLimitObservation } from "../../src/db/repositories"; +import { scheduledEnqueueDelaySeconds } from "../../src/selfhost/queue-common"; import { createTestEnv } from "../helpers/d1"; describe("worker entrypoint", () => { @@ -467,6 +468,60 @@ describe("worker entrypoint", () => { ]); }); + it("phase-spreads the scheduled enqueue: sweep immediate, periodic maintenance jittered (#1948)", async () => { + const sent: Array<{ + message: import("../../src/types").JobMessage; + delaySeconds?: number; + }> = []; + const env = createTestEnv({ + JOBS: { + async send( + message: import("../../src/types").JobMessage, + options?: { delaySeconds?: number }, + ) { + sent.push({ + message, + ...(options?.delaySeconds === undefined + ? {} + : { delaySeconds: options.delaySeconds }), + }); + }, + } as unknown as Queue, + }); + const waitUntil: Promise[] = []; + + await worker.scheduled(controllerFor("2026-05-25T06:00:00.000Z"), env, executionContext(waitUntil)); + await Promise.all(waitUntil); + + // The enqueued SET is unchanged — jitter only spreads run_after timing, never which jobs are sent. + expect(sent.map((s) => s.message.type)).toEqual([ + "agent-regate-sweep", + "backfill-registered-repos", + "repair-data-fidelity", + "refresh-installation-health", + "refresh-registry", + "refresh-scoring-model", + "refresh-upstream-drift", + "rollup-product-usage", + "generate-signal-snapshots", + "build-burden-forecasts", + "build-contributor-evidence", + "build-contributor-decision-packs", + "file-upstream-drift-issues", + ]); + // Each captured job's delay matches the deterministic policy: the every-tick sweep is immediate (sent with no + // options), the periodic maintenance jobs carry their stable per-type jitter slot. + for (const s of sent) { + const expected = scheduledEnqueueDelaySeconds(s.message.type); + if (expected > 0) expect(s.delaySeconds).toBe(expected); + else expect(s.delaySeconds).toBeUndefined(); + } + // The priority sweep specifically stays immediate; at least one periodic job is actually deferred, so a + // top-of-6h tick no longer fires every heavy fan-out parent in the same instant. + expect(sent.find((s) => s.message.type === "agent-regate-sweep")?.delaySeconds).toBeUndefined(); + expect(sent.some((s) => (s.delaySeconds ?? 0) > 0)).toBe(true); + }); + it("enqueues the ops-alerts job hourly ONLY when GITTENSORY_REVIEW_OPS is ON (flag-OFF is byte-identical)", async () => { const sentFor = async (opsFlag?: string): Promise> => { const sent: Array = []; diff --git a/test/unit/selfhost-queue-common.test.ts b/test/unit/selfhost-queue-common.test.ts index de49974a31..8b5a0b7754 100644 --- a/test/unit/selfhost-queue-common.test.ts +++ b/test/unit/selfhost-queue-common.test.ts @@ -3,6 +3,7 @@ import { FOREGROUND_QUEUE_PRIORITY_FLOOR, buildSelfHostQueueSnapshot, consumingRetryDelayMs, + deterministicJitterMs, githubRateLimitAdmissionDelayMs, githubRateLimitAdmissionKeyScope, githubRateLimitAdmissionKeyForJob, @@ -23,6 +24,8 @@ import { queueSnapshotFromBinding, queueStartupJitterMinJobs, queueStartupJitterMs, + scheduledEnqueueDelaySeconds, + scheduledEnqueueJitterMs, } from "../../src/selfhost/queue-common"; import { clearGitHubResponseCacheForTest, githubRateLimitAdmissionKeyForInstallation, timeoutFetch } from "../../src/github/client"; import { RetryableJobError } from "../../src/queue/retryable"; @@ -812,4 +815,70 @@ describe("self-host queue common helpers", () => { else process.env.QUEUE_STARTUP_JITTER_MIN_JOBS = old; } }); + + it("parses the scheduled-enqueue jitter window with defensive fallbacks", () => { + const old = process.env.SCHEDULED_ENQUEUE_JITTER_MS; + try { + delete process.env.SCHEDULED_ENQUEUE_JITTER_MS; + expect(scheduledEnqueueJitterMs()).toBe(5 * 60_000); // default + process.env.SCHEDULED_ENQUEUE_JITTER_MS = "42000"; + expect(scheduledEnqueueJitterMs()).toBe(42000); + process.env.SCHEDULED_ENQUEUE_JITTER_MS = "-1"; // negative → fallback + expect(scheduledEnqueueJitterMs()).toBe(5 * 60_000); + process.env.SCHEDULED_ENQUEUE_JITTER_MS = "not-a-number"; // NaN → fallback + expect(scheduledEnqueueJitterMs()).toBe(5 * 60_000); + } finally { + if (old === undefined) delete process.env.SCHEDULED_ENQUEUE_JITTER_MS; + else process.env.SCHEDULED_ENQUEUE_JITTER_MS = old; + } + }); + + it("keeps the every-tick priority jobs immediate and phase-spreads the periodic maintenance jobs (#1948)", () => { + const old = process.env.SCHEDULED_ENQUEUE_JITTER_MS; + try { + delete process.env.SCHEDULED_ENQUEUE_JITTER_MS; // default 5-min window + // The timely-merge sweep and its Orb-relay retry run every ~2-min tick → never deferred. + expect(scheduledEnqueueDelaySeconds("agent-regate-sweep")).toBe(0); + expect(scheduledEnqueueDelaySeconds("retry-orb-relay")).toBe(0); + + // A periodic maintenance job gets a stable, in-window slot derived from the shared jitter helper. + const window = 5 * 60_000; + for (const type of [ + "refresh-registry", + "refresh-scoring-model", + "generate-signal-snapshots", + "build-contributor-evidence", + ]) { + const delay = scheduledEnqueueDelaySeconds(type); + expect(delay).toBe(Math.floor(deterministicJitterMs(type, window) / 1000)); + expect(delay).toBeGreaterThanOrEqual(0); + expect(delay).toBeLessThanOrEqual(window / 1000); + expect(scheduledEnqueueDelaySeconds(type)).toBe(delay); // deterministic + } + + // Distinct job types land in distinct slots → the enqueue is actually spread, not synchronized. + const slots = [ + "refresh-registry", + "refresh-scoring-model", + "refresh-upstream-drift", + "generate-signal-snapshots", + "build-burden-forecasts", + "build-contributor-evidence", + "build-contributor-decision-packs", + "file-upstream-drift-issues", + "rollup-product-usage", + ].map(scheduledEnqueueDelaySeconds); + expect(new Set(slots).size).toBeGreaterThan(1); + + // A sub-second window collapses every slot to an immediate send (covers the floor → 0 path). + process.env.SCHEDULED_ENQUEUE_JITTER_MS = "500"; + expect(scheduledEnqueueDelaySeconds("refresh-registry")).toBe(0); + // A zero window disables jitter entirely. + process.env.SCHEDULED_ENQUEUE_JITTER_MS = "0"; + expect(scheduledEnqueueDelaySeconds("generate-signal-snapshots")).toBe(0); + } finally { + if (old === undefined) delete process.env.SCHEDULED_ENQUEUE_JITTER_MS; + else process.env.SCHEDULED_ENQUEUE_JITTER_MS = old; + } + }); });