fix(host-service): bound concurrent GitHub lookups in the PR runtime

packages/host-service/src/runtime/pull-requests/pull-requests.test.ts

@@ -155,6 +155,7 @@ function createManager(
 			worktreePath: string,
 		) => Promise<WorkspaceRefsSnapshot>;
 		worktreeExists?: (worktreePath: string) => boolean;
+		octokitTimeoutMs?: number;
 	} = {},
 ) {
 	return new PullRequestRuntimeManager({
@@ -179,6 +180,7 @@ function createManager(
 		// Seeded worktree paths are fabricated; default the disk gate open so
 		// sync-path tests exercise the git read, not the missing-dir skip.
 		worktreeExists: overrides.worktreeExists ?? (() => true),
+		octokitTimeoutMs: overrides.octokitTimeoutMs,
 	});
 }
 
@@ -2490,3 +2492,196 @@ describe("PullRequestRuntimeManager GitHub traffic", () => {
 		expect(getWorkspace(db, "ws")?.pullRequestId).toBeNull();
 	});
 });
+
+describe("PullRequestRuntimeManager fleet sweep under load", () => {
+	const PROJECTS = 96;
+	const WORKSPACES_PER_PROJECT = 2;
+
+	function seedFleet(db: HostDb) {
+		const prsByHeadRef: Record<string, ReturnType<typeof makePrNode>> = {};
+		for (let p = 0; p < PROJECTS; p++) {
+			const projectId = `project-${p}`;
+			db.insert(schema.projects)
+				.values({
+					id: projectId,
+					repoPath: `/repo-${p}`,
+					createdAt: Date.now(),
+					repoProvider: "github",
+					repoOwner: REPO.owner,
+					repoName: REPO.name,
+					repoUrl: `https://github.com/${REPO.owner}/${REPO.name}.git`,
+					remoteName: "origin",
+				})
+				.run();
+			for (let w = 0; w < WORKSPACES_PER_PROJECT; w++) {
+				const branch = `p${p}-w${w}`;
+				db.insert(schema.workspaces)
+					.values({
+						id: `ws-${branch}`,
+						projectId,
+						worktreePath: `/repo-${p}/.worktrees/${branch}`,
+						branch,
+						createdAt: Date.now(),
+						headSha: `sha-${branch}`,
+						upstreamOwner: REPO.owner,
+						upstreamRepo: REPO.name,
+						upstreamBranch: branch,
+					})
+					.run();
+				prsByHeadRef[branch] = makePrNode({
+					number: Object.keys(prsByHeadRef).length + 1,
+					headRef: branch,
+					headSha: `sha-${branch}`,
+				});
+			}
+		}
+		return prsByHeadRef;
+	}
+
+	function killedByTimeout(args: string[]) {
+		return Object.assign(new Error(`Command failed: gh ${args.join(" ")}`), {
+			killed: true,
+			signal: "SIGTERM",
+			code: null,
+		});
+	}
+
+	// A machine that runs `slots` gh processes at once, each needing `callMs`.
+	// Like execFile, the timeout starts at spawn, so time spent waiting behind
+	// other spawns counts against it. Virtual time: nothing really waits.
+	function contendedGh(
+		answer: (args: string[]) => Promise<unknown>,
+		machine: { slots: number; callMs: number; timeoutMs: number },
+	) {
+		const stats = { spawns: 0, killed: 0 };
+		let inFlight = 0;
+		const execGh = async (args: string[]) => {
+			stats.spawns += 1;
+			const ahead = inFlight++;
+			try {
+				await new Promise((resolve) => setImmediate(resolve));
+				const elapsedMs =
+					(Math.floor(ahead / machine.slots) + 1) * machine.callMs;
+				if (elapsedMs > machine.timeoutMs) {
+					stats.killed += 1;
+					throw killedByTimeout(args);
+				}
+				return await answer(args);
+			} finally {
+				inFlight--;
+			}
+		};
+		return { execGh, stats };
+	}
+
+	function fleetSweep(manager: PullRequestRuntimeManager) {
+		const accessible = manager as unknown as {
+			refreshEligibleProjects(): Promise<void>;
+		};
+		return accessible.refreshEligibleProjects.bind(accessible);
+	}
+
+	test("a healthy GitHub stays reachable when the sweep covers many projects", async () => {
+		const db = createRealDb();
+		const prsByHeadRef = seedFleet(db);
+		const warns: string[] = [];
+		const originalWarn = console.warn;
+		console.warn = (...args: unknown[]) => {
+			warns.push(String(args[0]));
+		};
+		// The reporter's numbers: one PR call takes 0.4 s, gh times out at 10 s.
+		const gh = contendedGh(routeGh(prsByHeadRef), {
+			slots: 4,
+			callMs: 400,
+			timeoutMs: 10_000,
+		});
+		const manager = createManager(db, {
+			execGh: gh.execGh,
+			github: (async () => {
+				throw new Error("octokit must not run");
+			}) as never,
+		});
+		try {
+			await fleetSweep(manager)();
+
+			expect(manager.getGithubStatus()).toBeNull();
+			expect(warns.filter((w) => w.includes("GitHub unreachable"))).toEqual([]);
+			expect(gh.stats.killed).toBe(0);
+			expect(getWorkspace(db, "ws-p95-w1")?.pullRequestId).not.toBeNull();
+		} finally {
+			manager.stop();
+			console.warn = originalWarn;
+		}
+	});
+
+	test("an Octokit fallback that never answers frees its slot for an explicit refresh", async () => {
+		const db = createRealDb();
+		const prsByHeadRef = seedFleet(db);
+		const answer = routeGh(prsByHeadRef);
+		let ghAnswers = false;
+		let hungOctokitCalls = 0;
+		const manager = createManager(db, {
+			execGh: async (args) => {
+				await new Promise((resolve) => setImmediate(resolve));
+				if (ghAnswers) return answer(args);
+				throw new Error("gh: Not Found (HTTP 404)");
+			},
+			github: () => {
+				hungOctokitCalls += 1;
+				return new Promise<never>(() => {});
+			},
+			octokitTimeoutMs: 5,
+		});
+		let stuckTimer: ReturnType<typeof setTimeout> | undefined;
+		try {
+			await withSilencedWarnings(async () => {
+				const sweep = fleetSweep(manager)();
+				while (hungOctokitCalls < 8) {
+					await new Promise((resolve) => setImmediate(resolve));
+				}
+				ghAnswers = true;
+
+				const outcome = await Promise.race([
+					manager
+						.refreshPullRequestsByWorkspaces(["ws-p95-w1"])
+						.then(() => "refreshed"),
+					new Promise((resolve) => {
+						stuckTimer = setTimeout(resolve, 1_000, "stuck");
+					}),
+				]);
+
+				expect(outcome).toBe("refreshed");
+				expect(getWorkspace(db, "ws-p95-w1")?.pullRequestId).not.toBeNull();
+				await sweep;
+			});
+		} finally {
+			clearTimeout(stuckTimer);
+			manager.stop();
+		}
+	});
+
+	test("a GitHub that hangs on every call trips the gate without a spawn per project", async () => {
+		const db = createRealDb();
+		seedFleet(db);
+		let spawns = 0;
+		const manager = createManager(db, {
+			execGh: async (args) => {
+				spawns += 1;
+				await new Promise((resolve) => setImmediate(resolve));
+				throw killedByTimeout(args);
+			},
+			github: (async () => {
+				throw new Error("octokit must not run on a dead network");
+			}) as never,
+		});
+		try {
+			await withSilencedWarnings(() => fleetSweep(manager)());
+
+			expect(manager.getGithubStatus()?.reason).toBe("unreachable");
+			expect(spawns).toBeGreaterThan(0);
+			expect(spawns).toBeLessThan(PROJECTS);
+		} finally {
+			manager.stop();
+		}
+	});
+});

packages/host-service/src/runtime/pull-requests/pull-requests.ts

@@ -13,7 +13,10 @@ import {
 } from "../../db/schema";
 import type { EventBus } from "../../events/event-bus";
 import type { GitWatcher } from "../../events/git-watcher";
-import type { ExecGh } from "../../trpc/router/workspace-creation/utils/exec-gh";
+import {
+	type ExecGh,
+	GH_TIMEOUT_MS,
+} from "../../trpc/router/workspace-creation/utils/exec-gh";
 import { type GitFactory, resolveDefaultBranchName } from "../git";
 import { ConditionalGh } from "./utils/conditional-gh";
 import {
@@ -59,6 +62,8 @@ import {
 	type WorkspaceRefsSnapshot,
 } from "./utils/workspace-refs";
 
+// gh's timeout runs from spawn, so an unbounded burst times out healthy calls.
+const MAX_CONCURRENT_GITHUB_LOOKUPS = 8;
 // Long-cadence sweep that catches anything `GitWatcher` might miss
 // (overflow, fs.watch errors, transient watcher failures). Steady-state
 // branch syncs are event-driven via `GitWatcher.onChanged`; this is a
@@ -151,6 +156,7 @@ export interface PullRequestRuntimeManagerOptions {
 	readWorkspaceRefs?: (worktreePath: string) => Promise<WorkspaceRefsSnapshot>;
 	/** Test seam for the missing-worktree gate. Defaults to `existsSync`. */
 	worktreeExists?: (worktreePath: string) => boolean;
+	octokitTimeoutMs?: number;
 }
 
 interface NormalizedRepoIdentity {
@@ -180,6 +186,33 @@ export interface CheckoutPullRequestMetadata {
 	isCrossRepository: boolean;
 }
 
+// ponytail: the request is not aborted, only abandoned. The ETIMEDOUT trips the
+// unreachable hold, so no new requests start; thread an AbortSignal into the
+// github-query helpers if abandoned sockets ever matter.
+async function rejectAfter<T>(
+	work: Promise<T>,
+	ms: number,
+	what: string,
+): Promise<T> {
+	let timer: ReturnType<typeof setTimeout> | undefined;
+	const deadline = new Promise<never>((_, reject) => {
+		timer = setTimeout(
+			() =>
+				reject(
+					Object.assign(new Error(`${what} timed out after ${ms}ms`), {
+						code: "ETIMEDOUT",
+					}),
+				),
+			ms,
+		);
+	});
+	try {
+		return await Promise.race([work, deadline]);
+	} finally {
+		clearTimeout(timer);
+	}
+}
+
 function mapCheckoutPullRequestState(
 	state: CheckoutPullRequestMetadata["state"],
 	isDraft: boolean,
@@ -236,6 +269,9 @@ export class PullRequestRuntimeManager {
 	// appear once the limit resets" is true within seconds rather than at
 	// the next 5-minute sweep.
 	private holdRecoveryTimer: ReturnType<typeof setTimeout> | null = null;
+	private readonly octokitTimeoutMs: number;
+	private githubLookupsRunning = 0;
+	private readonly githubLookupQueue: (() => void)[] = [];
 	private readonly workspaceSyncState = new Map<
 		string,
 		{ running: Promise<void>; rerunPending: boolean; bypassCache: boolean }
@@ -293,6 +329,7 @@ export class PullRequestRuntimeManager {
 			options.readWorkspaceRefs ??
 			(async (worktreePath) => readWorkspaceRefs(await this.git(worktreePath)));
 		this.worktreeExists = options.worktreeExists ?? existsSync;
+		this.octokitTimeoutMs = options.octokitTimeoutMs ?? GH_TIMEOUT_MS;
 	}
 
 	start() {
@@ -1281,34 +1318,61 @@ export class PullRequestRuntimeManager {
 		viaOctokit: () => Promise<T>,
 		options: { probe?: boolean } = {},
 	): Promise<T> {
-		// An explicit refresh (PR just created, user asked) is allowed through a
-		// hold: it is one call, and its success is what reopens the gate early
-		// once the network is back.
 		if (!options.probe) this.githubGate.assertReachable();
-		try {
-			const result = await viaGh();
-			this.githubGate.recordSuccess();
-			return result;
-		} catch (ghError) {
-			if (isGitHubUnreachableError(ghError)) {
-				this.noteGitHubFailure(ghError);
-				throw ghError;
-			}
-			console.warn(
-				`[host-service:pull-request-runtime] gh ${what} failed; falling back to Octokit`,
-				{ ...context, error: ghError },
-			);
+		if (this.githubLookupsRunning < MAX_CONCURRENT_GITHUB_LOOKUPS) {
+			this.githubLookupsRunning += 1;
+		} else {
+			await this.waitForGitHubSlot(options.probe === true);
 		}
 		try {
-			const result = await viaOctokit();
-			this.githubGate.recordSuccess();
-			return result;
-		} catch (octokitError) {
-			this.noteGitHubFailure(octokitError);
-			throw octokitError;
+			// An explicit refresh (PR just created, user asked) is allowed through a
+			// hold: it is one call, and its success is what reopens the gate early
+			// once the network is back.
+			if (!options.probe) this.githubGate.assertReachable();
+			try {
+				const result = await viaGh();
+				this.githubGate.recordSuccess();
+				return result;
+			} catch (ghError) {
+				if (isGitHubUnreachableError(ghError)) {
+					this.noteGitHubFailure(ghError);
+					throw ghError;
+				}
+				console.warn(
+					`[host-service:pull-request-runtime] gh ${what} failed; falling back to Octokit`,
+					{ ...context, error: ghError },
+				);
+			}
+			try {
+				const result = await rejectAfter(
+					viaOctokit(),
+					this.octokitTimeoutMs,
+					`Octokit ${what}`,
+				);
+				this.githubGate.recordSuccess();
+				return result;
+			} catch (octokitError) {
+				this.noteGitHubFailure(octokitError);
+				throw octokitError;
+			}
+		} finally {
+			this.releaseGitHubSlot();
 		}
 	}
 
+	private waitForGitHubSlot(urgent: boolean): Promise<void> {
+		return new Promise((resolve) => {
+			if (urgent) this.githubLookupQueue.unshift(resolve);
+			else this.githubLookupQueue.push(resolve);
+		});
+	}
+
+	private releaseGitHubSlot(): void {
+		const next = this.githubLookupQueue.shift();
+		if (next) next();
+		else this.githubLookupsRunning -= 1;
+	}
+
 	/** Trips the gate on a host-wide failure and says so once per hold. */
 	private noteGitHubFailure(error: unknown): void {
 		const hold = this.githubGate.recordFailure(error);

packages/host-service/src/trpc/router/workspace-creation/utils/exec-gh.ts

@@ -4,6 +4,8 @@ import { getToolEnvironment } from "../../../../terminal/clean-shell-env";
 
 const execFileAsync = promisify(execFile);
 
+export const GH_TIMEOUT_MS = 10_000;
+
 export interface ExecGhOptions {
 	cwd?: string;
 	timeout?: number;
@@ -28,7 +30,7 @@ export const execGh: ExecGh = async (args, options) => {
 	const env = await getToolEnvironment();
 	const child = execFileAsync("gh", args, {
 		encoding: "utf8",
-		timeout: options?.timeout ?? 10_000,
+		timeout: options?.timeout ?? GH_TIMEOUT_MS,
 		// Node's 1MB default dies on large REST payloads (open-PR sweeps of
 		// busy repos exceed it even paginated).
 		maxBuffer: options?.maxBuffer ?? 10 * 1024 * 1024,