From 6df71b5a24e0b2019f7ee5aa5ebc9777802b1403 Mon Sep 17 00:00:00 2001 From: real-venus Date: Sun, 12 Jul 2026 14:22:46 -0700 Subject: [PATCH] feat(miner): throttle discovery concurrency as rate-limit budget drops The fanout already recorded GitHub's x-ratelimit-remaining but never slowed its own concurrent fetching in response, so a discover run could sprint at full concurrency straight into a 403. Add a pure resolveThrottledConcurrency helper that maps the recorded remaining budget to an allowed in-flight concurrency (full speed when unknown or above the high-water mark, serialized at/below the low-water mark, scaling linearly in between), and make mapWithConcurrency park its pooled workers against that live limit so effective concurrency tapers off as the budget approaches zero. Thresholds are options-configurable with sensible defaults. Only decides how many requests run; never changes which docs are fetched or how a policy verdict is derived. Closes #4844 --- .../lib/discovery-throttle.d.ts | 8 + .../lib/discovery-throttle.js | 33 +++++ .../lib/opportunity-fanout.d.ts | 16 ++ .../lib/opportunity-fanout.js | 73 +++++++++- test/unit/miner-discovery-throttle.test.ts | 137 ++++++++++++++++++ 5 files changed, 261 insertions(+), 6 deletions(-) create mode 100644 packages/gittensory-miner/lib/discovery-throttle.d.ts create mode 100644 packages/gittensory-miner/lib/discovery-throttle.js create mode 100644 test/unit/miner-discovery-throttle.test.ts diff --git a/packages/gittensory-miner/lib/discovery-throttle.d.ts b/packages/gittensory-miner/lib/discovery-throttle.d.ts new file mode 100644 index 0000000000..60f40ebb1b --- /dev/null +++ b/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; diff --git a/packages/gittensory-miner/lib/discovery-throttle.js b/packages/gittensory-miner/lib/discovery-throttle.js new file mode 100644 index 0000000000..819a445559 --- /dev/null +++ b/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); +} diff --git a/packages/gittensory-miner/lib/opportunity-fanout.d.ts b/packages/gittensory-miner/lib/opportunity-fanout.d.ts index 6f4d408694..d179b9da14 100644 --- a/packages/gittensory-miner/lib/opportunity-fanout.d.ts +++ b/packages/gittensory-miner/lib/opportunity-fanout.d.ts @@ -31,12 +31,22 @@ export type CandidateIssueSummary = { warnings: CandidateIssueWarning[]; }; +export function mapWithConcurrency( + items: T[], + maxConcurrency: number, + worker: (item: T, index: number) => Promise, + resolveLimit: () => number, + sleepFn?: (ms: number) => Promise, +): Promise; + export function fetchCandidateIssuesWithSummary( targets: FanoutTarget[], githubToken: string, options?: { apiBaseUrl?: string; concurrency?: number; + rateLimitLowWaterMark?: number; + rateLimitHighWaterMark?: number; perPage?: number; sleepFn?: (ms: number) => Promise; }, @@ -48,6 +58,8 @@ export function fetchCandidateIssues( options?: { apiBaseUrl?: string; concurrency?: number; + rateLimitLowWaterMark?: number; + rateLimitHighWaterMark?: number; perPage?: number; sleepFn?: (ms: number) => Promise; }, @@ -59,6 +71,8 @@ export function searchCandidateIssuesWithSummary( options?: { apiBaseUrl?: string; concurrency?: number; + rateLimitLowWaterMark?: number; + rateLimitHighWaterMark?: number; perPage?: number; sleepFn?: (ms: number) => Promise; }, @@ -70,6 +84,8 @@ export function searchCandidateIssues( options?: { apiBaseUrl?: string; concurrency?: number; + rateLimitLowWaterMark?: number; + rateLimitHighWaterMark?: number; perPage?: number; sleepFn?: (ms: number) => Promise; }, diff --git a/packages/gittensory-miner/lib/opportunity-fanout.js b/packages/gittensory-miner/lib/opportunity-fanout.js index a1786f15d8..348aa87ded 100644 --- a/packages/gittensory-miner/lib/opportunity-fanout.js +++ b/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,20 +303,54 @@ 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: @@ -317,6 +358,20 @@ function normalizeOptions(options = {}) { ? 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 = []; diff --git a/test/unit/miner-discovery-throttle.test.ts b/test/unit/miner-discovery-throttle.test.ts new file mode 100644 index 0000000000..80db659cf2 --- /dev/null +++ b/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 + }); +});