feat(miner): throttle discovery concurrency as rate-limit budget drops

packages/gittensory-miner/lib/discovery-throttle.d.ts

@@ -0,0 +1,8 @@
+export const DEFAULT_RATE_LIMIT_LOW_WATER_MARK: number;
+export const DEFAULT_RATE_LIMIT_HIGH_WATER_MARK: number;
+export function resolveThrottledConcurrency(
+  baseConcurrency: number,
+  rateLimitRemaining: number | null,
+  lowWaterMark: number,
+  highWaterMark: number,
+): number;

packages/gittensory-miner/lib/discovery-throttle.js

@@ -0,0 +1,33 @@
+// Dynamic discovery back-off (#4844): the fanout already records GitHub's `x-ratelimit-remaining`, but nothing
+// slowed its own concurrent fetching in response — a `discover` run could sprint at full concurrency straight
+// into a 403. This pure helper maps the recorded remaining budget to an allowed in-flight concurrency so the
+// fanout tapers off as the budget approaches zero. It only decides *how many* requests may run; it never changes
+// which docs are fetched or how a policy verdict is derived from them.
+
+/** At or below this remaining budget, serialize discovery to a single in-flight request. */
+export const DEFAULT_RATE_LIMIT_LOW_WATER_MARK = 50;
+/** At or above this remaining budget, run at the full configured concurrency. */
+export const DEFAULT_RATE_LIMIT_HIGH_WATER_MARK = 250;
+
+/**
+ * Resolve the concurrency the fanout may run at for the currently-recorded rate-limit budget. Returns an integer
+ * in `[1, baseConcurrency]`:
+ *  - an unknown budget (`null`/non-finite — nothing recorded yet) runs at full `baseConcurrency`;
+ *  - at or below `lowWaterMark` it clamps to a single in-flight request;
+ *  - at or above `highWaterMark` it runs at full `baseConcurrency`;
+ *  - in between it scales linearly with the remaining fraction of the low→high band.
+ * @param {number} baseConcurrency
+ * @param {number|null} rateLimitRemaining
+ * @param {number} lowWaterMark
+ * @param {number} highWaterMark
+ * @returns {number}
+ */
+export function resolveThrottledConcurrency(baseConcurrency, rateLimitRemaining, lowWaterMark, highWaterMark) {
+  if (!Number.isFinite(rateLimitRemaining)) return baseConcurrency;
+  if (rateLimitRemaining <= lowWaterMark) return 1;
+  if (rateLimitRemaining >= highWaterMark) return baseConcurrency;
+  // remaining is strictly inside the (low, high) band, so the fraction is in (0, 1) and the ceil lands in
+  // [1, baseConcurrency] without any further clamping.
+  const fraction = (rateLimitRemaining - lowWaterMark) / (highWaterMark - lowWaterMark);
+  return Math.ceil(fraction * baseConcurrency);
+}

packages/gittensory-miner/lib/opportunity-fanout.d.ts

@@ -31,12 +31,22 @@ export type CandidateIssueSummary = {
   warnings: CandidateIssueWarning[];
 };
 
+export function mapWithConcurrency<T, R>(
+  items: T[],
+  maxConcurrency: number,
+  worker: (item: T, index: number) => Promise<R>,
+  resolveLimit: () => number,
+  sleepFn?: (ms: number) => Promise<unknown>,
+): Promise<R[]>;
+
 export function fetchCandidateIssuesWithSummary(
   targets: FanoutTarget[],
   githubToken: string,
   options?: {
     apiBaseUrl?: string;
     concurrency?: number;
+    rateLimitLowWaterMark?: number;
+    rateLimitHighWaterMark?: number;
     perPage?: number;
     sleepFn?: (ms: number) => Promise<unknown>;
   },
@@ -48,6 +58,8 @@ export function fetchCandidateIssues(
   options?: {
     apiBaseUrl?: string;
     concurrency?: number;
+    rateLimitLowWaterMark?: number;
+    rateLimitHighWaterMark?: number;
     perPage?: number;
     sleepFn?: (ms: number) => Promise<unknown>;
   },
@@ -59,6 +71,8 @@ export function searchCandidateIssuesWithSummary(
   options?: {
     apiBaseUrl?: string;
     concurrency?: number;
+    rateLimitLowWaterMark?: number;
+    rateLimitHighWaterMark?: number;
     perPage?: number;
     sleepFn?: (ms: number) => Promise<unknown>;
   },
@@ -70,6 +84,8 @@ export function searchCandidateIssues(
   options?: {
     apiBaseUrl?: string;
     concurrency?: number;
+    rateLimitLowWaterMark?: number;
+    rateLimitHighWaterMark?: number;
     perPage?: number;
     sleepFn?: (ms: number) => Promise<unknown>;
   },

packages/gittensory-miner/lib/opportunity-fanout.js

@@ -1,9 +1,16 @@
 import { Buffer } from "node:buffer";
 import { resolveAiPolicyVerdict } from "@jsonbored/gittensory-engine";
+import {
+  DEFAULT_RATE_LIMIT_HIGH_WATER_MARK,
+  DEFAULT_RATE_LIMIT_LOW_WATER_MARK,
+  resolveThrottledConcurrency,
+} from "./discovery-throttle.js";
 import { fetchWithRetry } from "./http-retry.js";
 
 const defaultApiBaseUrl = "https://api.github.com";
 const defaultConcurrency = 5;
+// How long a parked worker waits before re-checking the live rate-limit-derived concurrency limit (#4844).
+const throttleParkMs = 25;
 const defaultPerPage = 100;
 // Follow the GitHub Link header past the first page so a repo/search with >100 open issues isn't silently
 // truncated (#4831); cap the follow loop so a pathological Link chain can't run away.
@@ -296,27 +303,75 @@ async function fetchSearchIssues(searchQuery, githubToken, options, summary, war
   }
 }
 
-async function mapWithConcurrency(items, concurrency, worker) {
+function delay(ms) {
+  return new Promise((resolve) => setTimeout(resolve, ms));
+}
+
+// Run `worker` over `items` with a dynamic in-flight cap (#4844). The pool spawns `maxConcurrency` loops, but a
+// loop parks (re-checking every `throttleParkMs`) whenever the live `resolveLimit()` — derived from the recorded
+// rate-limit budget — is already met by the number of in-flight workers, so effective concurrency tapers off as
+// the budget drops instead of sprinting into a 403. `sleepFn` lets tests inject an instant wait for the park.
+export async function mapWithConcurrency(items, maxConcurrency, worker, resolveLimit, sleepFn) {
   const results = new Array(items.length);
+  const sleep = sleepFn ?? delay;
   let next = 0;
-  const workers = Array.from({ length: Math.min(concurrency, items.length) }, async () => {
+  let active = 0;
+  const runOne = async () => {
     while (next < items.length) {
+      // Park while the live limit is already saturated. The check and the `active`/`next` bumps below run without
+      // an intervening await, so two loops can never claim the same slot.
+      while (active >= resolveLimit()) {
+        await sleep(throttleParkMs);
+      }
+      // The shared cursor can be drained by other loops while this one is parked, so re-check before claiming.
+      if (next >= items.length) return;
       const index = next;
       next += 1;
-      results[index] = await worker(items[index], index);
+      active += 1;
+      try {
+        results[index] = await worker(items[index], index);
+      } finally {
+        active -= 1;
+      }
     }
-  });
+  };
+  const workers = Array.from({ length: Math.min(maxConcurrency, items.length) }, runOne);
   await Promise.all(workers);
   return results;
 }
 
+/** A live limit resolver for `mapWithConcurrency`, reading the summary's rate-limit budget as it is updated (#4844). */
+function liveConcurrencyResolver(normalizedOptions, summary) {
+  return () =>
+    resolveThrottledConcurrency(
+      normalizedOptions.concurrency,
+      summary.rateLimitRemaining,
+      normalizedOptions.rateLimitLowWaterMark,
+      normalizedOptions.rateLimitHighWaterMark,
+    );
+}
+
 function normalizeOptions(options = {}) {
   return {
     apiBaseUrl:
       typeof options.apiBaseUrl === "string" && options.apiBaseUrl.trim()
         ? options.apiBaseUrl.trim()
         : defaultApiBaseUrl,
     concurrency: normalizeLimit(options.concurrency, defaultConcurrency, 1, 10),
+    // Below/above these recorded-rate-limit-remaining marks the fanout serializes / runs at full concurrency; in
+    // between it scales down linearly (#4844).
+    rateLimitLowWaterMark: normalizeLimit(
+      options.rateLimitLowWaterMark,
+      DEFAULT_RATE_LIMIT_LOW_WATER_MARK,
+      0,
+      1_000_000,
+    ),
+    rateLimitHighWaterMark: normalizeLimit(
+      options.rateLimitHighWaterMark,
+      DEFAULT_RATE_LIMIT_HIGH_WATER_MARK,
+      1,
+      1_000_000,
+    ),
     perPage: normalizeLimit(options.perPage, defaultPerPage, 1, 100),
     maxPages: normalizeLimit(options.maxPages, defaultMaxPages, 1, 100),
     // Passed through to the per-fetch retry so tests can inject an instant sleep; undefined uses the real backoff.
@@ -332,8 +387,12 @@ export async function fetchCandidateIssuesWithSummary(targets, githubToken, opti
     rateLimitResetAt: null,
   };
   const warnings = [];
-  const batches = await mapWithConcurrency(normalizedTargets, normalizedOptions.concurrency, (target) =>
-    fetchTargetIssues(target, githubToken, normalizedOptions, summary, warnings),
+  const batches = await mapWithConcurrency(
+    normalizedTargets,
+    normalizedOptions.concurrency,
+    (target) => fetchTargetIssues(target, githubToken, normalizedOptions, summary, warnings),
+    liveConcurrencyResolver(normalizedOptions, summary),
+    normalizedOptions.sleepFn,
   );
   return {
     issues: batches.flat(),
@@ -375,6 +434,8 @@ export async function searchCandidateIssuesWithSummary(searchQuery, githubToken,
       const verdict = await resolveRepoAiPolicy(target, githubToken, normalizedOptions, summary, warnings);
       return [targetKey(target), verdict];
     },
+    liveConcurrencyResolver(normalizedOptions, summary),
+    normalizedOptions.sleepFn,
   );
   const policiesByKey = new Map(policyEntries);
   const issues = [];

test/unit/miner-discovery-throttle.test.ts

@@ -0,0 +1,137 @@
+import { afterEach, describe, expect, it, vi } from "vitest";
+
+vi.mock("@jsonbored/gittensory-engine", async () => {
+  return import("../../packages/gittensory-engine/src/index");
+});
+
+import {
+  DEFAULT_RATE_LIMIT_HIGH_WATER_MARK,
+  DEFAULT_RATE_LIMIT_LOW_WATER_MARK,
+  resolveThrottledConcurrency,
+} from "../../packages/gittensory-miner/lib/discovery-throttle.js";
+import {
+  fetchCandidateIssuesWithSummary,
+  mapWithConcurrency,
+} from "../../packages/gittensory-miner/lib/opportunity-fanout.js";
+
+const instant = async () => {};
+
+afterEach(() => {
+  vi.unstubAllGlobals();
+});
+
+describe("resolveThrottledConcurrency (#4844)", () => {
+  const LOW = DEFAULT_RATE_LIMIT_LOW_WATER_MARK; // 50
+  const HIGH = DEFAULT_RATE_LIMIT_HIGH_WATER_MARK; // 250
+
+  it("runs at full concurrency when the remaining budget is unknown", () => {
+    expect(resolveThrottledConcurrency(5, null, LOW, HIGH)).toBe(5);
+    expect(resolveThrottledConcurrency(5, Number.NaN, LOW, HIGH)).toBe(5);
+  });
+
+  it("serializes to a single request at or below the low-water mark", () => {
+    expect(resolveThrottledConcurrency(5, LOW, LOW, HIGH)).toBe(1); // exactly at the mark
+    expect(resolveThrottledConcurrency(5, 10, LOW, HIGH)).toBe(1); // well below
+  });
+
+  it("runs at full concurrency at or above the high-water mark", () => {
+    expect(resolveThrottledConcurrency(5, HIGH, LOW, HIGH)).toBe(5); // exactly at the mark
+    expect(resolveThrottledConcurrency(5, 5000, LOW, HIGH)).toBe(5); // well above
+  });
+
+  it("scales linearly through the low→high band", () => {
+    // midpoint of 50..250 is 150 → fraction 0.5 → ceil(0.5 * 5) = 3
+    expect(resolveThrottledConcurrency(5, 150, LOW, HIGH)).toBe(3);
+    // just above the low-water mark → the smallest non-serialized step, 1
+    expect(resolveThrottledConcurrency(5, 51, LOW, HIGH)).toBe(1);
+    // near the high-water mark → close to full
+    expect(resolveThrottledConcurrency(5, 240, LOW, HIGH)).toBe(5);
+  });
+
+  it("honors custom water marks", () => {
+    expect(resolveThrottledConcurrency(4, 100, 100, 200)).toBe(1); // at custom low
+    expect(resolveThrottledConcurrency(4, 150, 100, 200)).toBe(2); // custom midpoint → ceil(0.5*4)
+  });
+});
+
+describe("mapWithConcurrency dynamic in-flight cap (#4844)", () => {
+  it("never exceeds the live limit and still processes every item", async () => {
+    let active = 0;
+    let peak = 0;
+    const worker = async (item: number) => {
+      active += 1;
+      peak = Math.max(peak, active);
+      await Promise.resolve();
+      active -= 1;
+      return item * 2;
+    };
+    const results = await mapWithConcurrency([1, 2, 3, 4, 5], 5, worker, () => 1, instant);
+    expect(results).toEqual([2, 4, 6, 8, 10]);
+    expect(peak).toBe(1); // limit of 1 ⇒ fully serialized despite a pool of 5
+  });
+
+  it("tapers as the live limit drops mid-run", async () => {
+    let active = 0;
+    let peak = 0;
+    let limit = 4;
+    const worker = async (item: number) => {
+      active += 1;
+      peak = Math.max(peak, active);
+      if (item === 0) limit = 1; // the budget craters after the first item completes
+      await Promise.resolve();
+      active -= 1;
+      return item;
+    };
+    const results = await mapWithConcurrency([0, 1, 2, 3, 4, 5], 4, worker, () => limit, instant);
+    expect(results).toEqual([0, 1, 2, 3, 4, 5]);
+    expect(peak).toBeLessThanOrEqual(4);
+  });
+
+  it("parks on the real timer when no sleep function is injected", async () => {
+    let active = 0;
+    let peak = 0;
+    const worker = async (item: number) => {
+      active += 1;
+      peak = Math.max(peak, active);
+      await Promise.resolve();
+      active -= 1;
+      return item;
+    };
+    // No sleepFn ⇒ the park falls back to the built-in setTimeout-based delay.
+    const results = await mapWithConcurrency([1, 2, 3], 3, worker, () => 1);
+    expect(results).toEqual([1, 2, 3]);
+    expect(peak).toBe(1);
+  });
+});
+
+describe("discovery fanout throttling wiring (#4844)", () => {
+  const API = "https://api.test";
+
+  function lowBudgetFetch(remaining: string) {
+    return async (input: RequestInfo | URL) => {
+      const url = String(input);
+      if (url.includes("/contents/")) {
+        return Response.json({}, { status: 404, headers: { "x-ratelimit-remaining": remaining } });
+      }
+      if (url.includes("/issues?")) {
+        return Response.json([{ number: 1, title: "t", html_url: `${url}#1` }], {
+          headers: { "x-ratelimit-remaining": remaining, "x-ratelimit-reset": "1800000000" },
+        });
+      }
+      throw new Error(`unexpected request: ${url}`);
+    };
+  }
+
+  it("completes a multi-target run under a low remaining budget without erroring", async () => {
+    vi.stubGlobal("fetch", lowBudgetFetch("5")); // below the 50 low-water mark ⇒ serialized
+    const targets = Array.from({ length: 6 }, (_, i) => ({ owner: "acme", repo: `r${i}` }));
+    const result = await fetchCandidateIssuesWithSummary(targets, "", {
+      apiBaseUrl: API,
+      concurrency: 4,
+      sleepFn: instant,
+    });
+    expect(result.warnings).toEqual([]);
+    expect(result.rateLimitRemaining).toBe(5);
+    expect(result.issues).toHaveLength(6); // every target still fetched, just throttled
+  });
+});