fix(queue): phase-spread the scheduled cron enqueue (#1948)

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);
+    }),
+  );
 }

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<string>([
+  "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 {

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<unknown>[] = [];
+
+    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<Array<import("../../src/types").JobMessage>> => {
       const sent: Array<import("../../src/types").JobMessage> = [];

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;
+    }
+  });
 });