perf(next-version): fetch PR VERSIONs concurrently; a timed-out fetch is an unknown claim

Co-authored-by: capy-ai[bot] <230910855+capy-ai[bot]@users.noreply.github.com>

bin/gstack-next-version

@@ -6,7 +6,9 @@
 // slot at the requested bump level, and emits the whole picture as JSON.
 //
 // Contract: util NEVER writes files or mutates state. Pure reader + reporter.
-// /ship consumes the JSON and decides what to do.
+// /ship consumes the JSON and decides what to do. `unknown_claims` lists the
+// open PRs whose VERSION could not be read (a 15 s timeout or a non-404
+// failure): their claims are unknown, so the picked slot may collide.
 //
 // Usage:
 //   gstack-next-version --base <branch> --bump <major|minor|patch|micro> \
@@ -34,7 +36,7 @@
 //       empty, unreadable or malformed
 //   3 — util bug (unexpected exception)
 
-import { execFileSync, spawnSync } from "node:child_process";
+import { execFileSync, spawn, spawnSync } from "node:child_process";
 import { existsSync, readFileSync, readdirSync, statSync } from "node:fs";
 import { homedir } from "node:os";
 import { join, resolve } from "node:path";
@@ -84,10 +86,12 @@ type Output = {
   active_siblings: Sibling[];
   reason: string;
   warnings: string[];
+  unknown_claims: number[];
 };
 
 const ACTIVE_SIBLING_MAX_AGE_S = 24 * 60 * 60;
 const GH_API_CONCURRENCY = 10;
+const VERSION_FETCH_TIMEOUT_MS = 15000;
 
 // Collision resolution: bump past the highest claimed within the same level.
 // Semantics: if my bump is MINOR and the queue claims 1.7.0.0, I advance to
@@ -117,6 +121,33 @@ function runCommand(cmd: string, args: string[], timeoutMs = 15000): { ok: boole
   };
 }
 
+// Async twin of runCommand for the per-PR VERSION fetches, so GH_API_CONCURRENCY
+// workers really overlap. A fetch past timeoutMs is killed and settles at once
+// as timedOut, without waiting on output pipes a descendant may still hold.
+function runCommandAsync(cmd: string, args: string[], timeoutMs: number): Promise<{ ok: boolean; stdout: string; stderr: string; timedOut: boolean }> {
+  return new Promise((done) => {
+    let stdout = "";
+    let stderr = "";
+    const child = spawn(cmd, args, { stdio: ["ignore", "pipe", "pipe"] });
+    const timer = setTimeout(() => {
+      child.kill("SIGKILL");
+      child.stdout.destroy();
+      child.stderr.destroy();
+      done({ ok: false, stdout, stderr, timedOut: true });
+    }, timeoutMs);
+    child.stdout.on("data", (chunk) => { stdout += chunk; });
+    child.stderr.on("data", (chunk) => { stderr += chunk; });
+    child.on("error", (error) => {
+      clearTimeout(timer);
+      done({ ok: false, stdout, stderr: stderr || String(error), timedOut: false });
+    });
+    child.on("close", (code) => {
+      clearTimeout(timer);
+      done({ ok: code === 0, stdout, stderr, timedOut: false });
+    });
+  });
+}
+
 // VERSION-path resolution for monorepos. Priority: CLI flag > .gstack/version-path
 // at repo root > "VERSION". Pure function; takes the repo root as an argument so
 // tests can drive it with a fixture dir without mocking git.
@@ -181,7 +212,13 @@ function readBaseVersion(base: string, versionPath: string, repoRoot: string, wa
   return v;
 }
 
-async function fetchGithubClaimed(base: string, versionPath: string, excludePR: number | null, warnings: string[]): Promise<{ claimed: ClaimedPR[]; offline: boolean }> {
+async function fetchGithubClaimed(
+  base: string,
+  versionPath: string,
+  excludePR: number | null,
+  warnings: string[],
+  fetchTimeoutMs = VERSION_FETCH_TIMEOUT_MS,
+): Promise<{ claimed: ClaimedPR[]; offline: boolean; unknownClaims: number[] }> {
   const list = runCommand("gh", [
     "pr",
     "list",
@@ -196,7 +233,7 @@ async function fetchGithubClaimed(base: string, versionPath: string, excludePR:
   ]);
   if (!list.ok) {
     warnings.push(`gh pr list failed: ${list.stderr.trim().slice(0, 200)}`);
-    return { claimed: [], offline: true };
+    return { claimed: [], offline: true, unknownClaims: [] };
   }
   let prs: {
     number: number;
@@ -209,7 +246,7 @@ async function fetchGithubClaimed(base: string, versionPath: string, excludePR:
     prs = JSON.parse(list.stdout);
   } catch (e) {
     warnings.push(`gh pr list returned invalid JSON`);
-    return { claimed: [], offline: true };
+    return { claimed: [], offline: true, unknownClaims: [] };
   }
   // Determine our repo owner to filter out fork PRs. `gh api contents?ref=<branch>`
   // resolves to OUR repo regardless of where the PR originated, so fork PRs would
@@ -221,43 +258,56 @@ async function fetchGithubClaimed(base: string, versionPath: string, excludePR:
     : prs
   ).filter((p) => excludePR === null || p.number !== excludePR);
   // Fetch each PR's VERSION at its head in parallel (bounded concurrency).
+  // Workers finish in any order, so claims and warnings are sorted by PR number.
   const results: ClaimedPR[] = [];
+  const prWarnings: { pr: number; text: string }[] = [];
+  const unknownClaims: number[] = [];
   const queue = [...sameRepoPRs];
   const workers = Array.from({ length: Math.min(GH_API_CONCURRENCY, sameRepoPRs.length) }, async () => {
     while (queue.length) {
       const pr = queue.shift();
       if (!pr) return;
+      const warn = (text: string) => prWarnings.push({ pr: pr.number, text: `PR #${pr.number}: ${text}` });
       // gh passes branch name via argv, not shell — safe.
       // encodeURI handles spaces in subproject paths (e.g. "Tinas Second Brain/...")
       // while leaving "/" untouched so the GitHub Contents API gets the path intact.
-      const content = runCommand("gh", [
+      const content = await runCommandAsync("gh", [
         "api",
         `repos/{owner}/{repo}/contents/${encodeURI(versionPath)}?ref=${encodeURIComponent(pr.headRefName)}`,
         "-q",
         ".content",
-      ]);
+      ], fetchTimeoutMs);
+      if (content.timedOut) {
+        unknownClaims.push(pr.number);
+        warn(`VERSION fetch timed out after ${fetchTimeoutMs / 1000} s; claim unknown, the slot may collide; rerun, then ask the coordinator`);
+        continue;
+      }
       if (!content.ok) {
-        warnings.push(
-          `PR #${pr.number}: could not fetch ${versionPath} (fork, private, or wrong path — try --version-path or .gstack/version-path)`,
-        );
+        if (/HTTP 404|Not Found/i.test(content.stderr)) {
+          warn(`could not fetch ${versionPath} (fork, private, or wrong path — try --version-path or .gstack/version-path)`);
+        } else {
+          unknownClaims.push(pr.number);
+          warn(`VERSION fetch failed (${content.stderr.trim().split("\n")[0]?.slice(0, 200) || "no error output"}); claim unknown, the slot may collide; rerun, then ask the coordinator`);
+        }
         continue;
       }
       let versionStr: string;
       try {
         versionStr = extractVersion(Buffer.from(content.stdout.trim(), "base64").toString("utf8"), versionPath);
       } catch {
-        warnings.push(`PR #${pr.number}: VERSION is not valid base64`);
+        warn("VERSION is not valid base64");
         continue;
       }
       if (!parseVersion(versionStr)) {
-        warnings.push(`PR #${pr.number}: VERSION is malformed (${versionStr})`);
+        warn(`VERSION is malformed (${versionStr})`);
         continue;
       }
       results.push({ pr: pr.number, branch: pr.headRefName, version: versionStr, url: pr.url });
     }
   });
   await Promise.all(workers);
-  return { claimed: results, offline: false };
+  warnings.push(...prWarnings.sort((a, b) => a.pr - b.pr).map((w) => w.text));
+  return { claimed: results.sort((a, b) => a.pr - b.pr), offline: false, unknownClaims: unknownClaims.sort((a, b) => a - b) };
 }
 
 async function fetchGitlabClaimed(base: string, versionPath: string, excludePR: number | null, warnings: string[]): Promise<{ claimed: ClaimedPR[]; offline: boolean }> {
@@ -730,8 +780,9 @@ async function main() {
 
   let claimed: ClaimedPR[] = [];
   let offline = false;
+  let unknownClaims: number[] = [];
   if (host === "github") {
-    ({ claimed, offline } = await fetchGithubClaimed(args.base, versionPath, excludePR, warnings));
+    ({ claimed, offline, unknownClaims } = await fetchGithubClaimed(args.base, versionPath, excludePR, warnings));
   } else if (host === "gitlab") {
     ({ claimed, offline } = await fetchGitlabClaimed(args.base, versionPath, excludePR, warnings));
   } else {
@@ -803,6 +854,7 @@ async function main() {
     active_siblings: activeSiblings,
     reason: finalReason,
     warnings,
+    unknown_claims: unknownClaims,
   };
   process.stdout.write(JSON.stringify(out, null, 2) + "\n");
 }
@@ -811,7 +863,7 @@ async function main() {
 // from lib/version-source so existing importers of this module keep working
 // unchanged.
 export { parseVersion, fmtVersion, bumpVersion, cmpVersion, versionWidth, extractVersion };
-export { pickNextSlot, markActiveSiblings, resolveVersionPath, fetchGitClaimed };
+export { pickNextSlot, markActiveSiblings, resolveVersionPath, fetchGitClaimed, fetchGithubClaimed, VERSION_FETCH_TIMEOUT_MS };
 
 // Only run main() when invoked as a script, not when imported by tests.
 if (import.meta.main) {

test/gstack-next-version.test.ts

@@ -19,6 +19,8 @@ import {
   markActiveSiblings,
   resolveVersionPath,
   fetchGitClaimed,
+  fetchGithubClaimed,
+  VERSION_FETCH_TIMEOUT_MS,
 } from "../bin/gstack-next-version";
 
 describe("parseVersion", () => {
@@ -1217,3 +1219,134 @@ describe("no version source: one signal, never an invented slot (G1)", () => {
     expect(r.stderr).not.toContain("assuming");
   }, 30_000);
 });
+
+// W2-A3 (fix wave 3): VERSION fetches run through async workers bounded at
+// GH_API_CONCURRENCY, each with a 15 s timeout; claims and warnings are sorted
+// by PR number, and a timed-out or failed fetch is an unknown claim. A named
+// fake `gh` first on PATH answers every call with a per-PR delay; the live
+// GitHub API is never called.
+describe("concurrent VERSION fetches (W2-A3)", () => {
+  const OWNER = "fixture-owner";
+  type FakePR = { pr: number; version?: string; delay: number; reply?: "404" | "502" | "malformed" };
+  const prBranch = (pr: number) => `fixture/pr-${pr}`;
+  const prUrl = (pr: number) => `https://github.com/${OWNER}/repo/pull/${pr}`;
+  const claim = (pr: number, version: string) => ({ pr, branch: prBranch(pr), version, url: prUrl(pr) });
+
+  function delayedGh(dir: string, prs: FakePR[]): string {
+    const flight = join(dir, "flight.log");
+    const list = JSON.stringify(prs.map((p) => ({ number: p.pr, headRefName: prBranch(p.pr), headRepositoryOwner: { login: OWNER }, url: prUrl(p.pr), isDraft: false })));
+    const reply = (p: FakePR) => {
+      if (p.reply === "404") return `echo 'gh: Not Found (HTTP 404)' >&2; exit 1`;
+      if (p.reply === "502") return `echo 'gh: Server Error (HTTP 502)' >&2; exit 1`;
+      const body = p.reply === "malformed" ? "not-a-version\n" : `${p.version}\n`;
+      return `echo '${Buffer.from(body).toString("base64")}'; exit 0`;
+    };
+    writeFileSync(join(dir, "gh"), [
+      "#!/bin/sh",
+      'case "$*" in',
+      `  "pr list --state open --base main --limit 200 --json number,headRefName,headRepositoryOwner,url,isDraft") printf '%s\\n' '${list}'; exit 0 ;;`,
+      `  "repo view --json owner -q .owner.login") echo '${OWNER}'; exit 0 ;;`,
+      ...prs.map((p) => `  "api repos/{owner}/{repo}/contents/VERSION?ref=${encodeURIComponent(prBranch(p.pr))} -q .content") echo + >> '${flight}'; sleep ${p.delay}; echo - >> '${flight}'; ${reply(p)} ;;`),
+      "esac",
+      'echo "FAKE_GH_UNHANDLED $*" >&2',
+      "exit 2",
+      "",
+    ].join("\n"), { mode: 0o755 });
+    return flight;
+  }
+
+  async function fetchWith(prs: FakePR[], fetchTimeoutMs?: number) {
+    const dir = mkdtempSync(join(tmpdir(), "nextver-concurrency-"));
+    const flight = delayedGh(dir, prs);
+    const savedPath = process.env.PATH;
+    process.env.PATH = `${dir}:${savedPath ?? ""}`;
+    const warnings: string[] = [];
+    const started = performance.now();
+    try {
+      const result: { claimed: { pr: number }[]; offline: boolean; unknownClaims: number[] } = await fetchGithubClaimed("main", "VERSION", null, warnings, fetchTimeoutMs);
+      let running = 0;
+      let maxInFlight = 0;
+      for (const event of readFileSync(flight, "utf8").trim().split("\n")) {
+        running += event === "+" ? 1 : -1;
+        maxInFlight = Math.max(maxInFlight, running);
+      }
+      return { ...result, warnings, ms: performance.now() - started, maxInFlight };
+    } finally {
+      process.env.PATH = savedPath;
+      rmSync(dir, { recursive: true, force: true });
+    }
+  }
+
+  test("8 PRs at 0.5 s each finish in under 2 s with every claim, sorted by PR number", async () => {
+    const r = await fetchWith([48, 41, 47, 42, 46, 43, 45, 44].map((pr) => ({ pr, version: `1.6.${pr}.0`, delay: 0.5 })));
+    expect(r.ms).toBeLessThan(2000); // gstack-timing-ok: the plan's contract (8 x 0.5 s serial took 4,160 ms)
+    expect(r.offline).toBe(false);
+    expect(r.claimed).toEqual([41, 42, 43, 44, 45, 46, 47, 48].map((pr) => claim(pr, `1.6.${pr}.0`)));
+    expect(r.warnings).toEqual([]);
+    expect(r.unknownClaims).toEqual([]);
+    expect(r.maxInFlight).toBeGreaterThan(1);
+  }, 30_000);
+
+  test("randomized delays give the same claims, warnings and unknown claims every run, within the concurrency bound", async () => {
+    let seed = 7;
+    const random = () => { seed = (seed * 1103515245 + 12345) % 2 ** 31; return seed / 2 ** 31; };
+    const base: Omit<FakePR, "delay">[] = Array.from({ length: 14 }, (_, i) => ({ pr: 60 + i, version: `2.0.${i}.0` }));
+    base[2]!.reply = "404";
+    base[5]!.reply = "malformed";
+    base[9]!.reply = "502";
+    const runs = [];
+    for (let run = 0; run < 3; run++) runs.push(await fetchWith(base.map((p) => ({ ...p, delay: Number((random() * 0.3).toFixed(2)) }))));
+    for (const r of runs) {
+      expect(r.claimed).toEqual(runs[0]!.claimed);
+      expect(r.warnings).toEqual(runs[0]!.warnings);
+      expect(r.unknownClaims).toEqual([69]);
+      expect(r.maxInFlight).toBeLessThanOrEqual(10);
+    }
+    expect(runs[0]!.claimed.map((c) => c.pr)).toEqual([60, 61, 63, 64, 66, 67, 68, 70, 71, 72, 73]);
+    expect(runs[0]!.warnings).toEqual([
+      "PR #62: could not fetch VERSION (fork, private, or wrong path — try --version-path or .gstack/version-path)",
+      "PR #65: VERSION is malformed (not-a-version)",
+      "PR #69: VERSION fetch failed (gh: Server Error (HTTP 502)); claim unknown, the slot may collide; rerun, then ask the coordinator",
+    ]);
+  }, 60_000);
+
+  test("a timed-out fetch is killed, keeps its cause and counts as an unknown claim", async () => {
+    expect(VERSION_FETCH_TIMEOUT_MS).toBe(15000);
+    const r = await fetchWith([{ pr: 51, version: "1.7.1.0", delay: 0.1 }, { pr: 52, version: "1.7.2.0", delay: 10 }], 500);
+    expect(r.ms).toBeLessThan(5000); // gstack-timing-ok: the killed fetch must not wait out its 10 s sleeper
+    expect(r.claimed).toEqual([claim(51, "1.7.1.0")]);
+    expect(r.unknownClaims).toEqual([52]);
+    expect(r.warnings).toEqual(["PR #52: VERSION fetch timed out after 0.5 s; claim unknown, the slot may collide; rerun, then ask the coordinator"]);
+  }, 30_000);
+
+  test("the CLI JSON carries unknown_claims", () => {
+    const dir = mkdtempSync(join(tmpdir(), "nextver-unknown-cli-"));
+    try {
+      delayedGh(dir, [{ pr: 81, version: "1.6.4.0", delay: 0 }, { pr: 82, delay: 0, reply: "502" }]);
+      // A local origin whose path contains github.com pins host:"github" by the URL sniff, so no auth probe or network call runs.
+      const bare = join(dir, "github.com", "origin.git");
+      const work = join(dir, "work");
+      const git = (cwd: string, ...args: string[]) => Bun.spawnSync(["git", "-c", "user.email=t@t", "-c", "user.name=t", ...args], { cwd, timeout: 30_000 });
+      mkdirSync(bare, { recursive: true });
+      mkdirSync(work);
+      git(bare, "init", "-q", "--bare", "-b", "main");
+      git(work, "init", "-q", "-b", "main");
+      writeFileSync(join(work, "VERSION"), "1.6.3.0\n");
+      git(work, "add", "-A");
+      git(work, "commit", "-qm", "v1.6.3.0 chore: base");
+      git(work, "remote", "add", "origin", bare);
+      git(work, "push", "-q", "origin", "main");
+      const proc = Bun.spawnSync(["bun", "run", join(import.meta.dir, "..", "bin", "gstack-next-version"), "--base", "main", "--bump", "patch",
+        "--current-version", "1.6.3.0", "--workspace-root", "null", "--exclude-pr", "999"],
+      { cwd: work, timeout: 30_000, env: { ...process.env, PATH: `${dir}:${process.env.PATH ?? ""}` } });
+      const out = JSON.parse(proc.stdout.toString());
+      expect(out.host, proc.stderr.toString()).toBe("github");
+      expect(out.unknown_claims).toEqual([82]);
+      expect(out.claimed.map((c: { pr: number }) => c.pr)).toEqual([81]);
+      expect(out.version).toBe("1.6.5.0");
+      expect(out.warnings).toContain("PR #82: VERSION fetch failed (gh: Server Error (HTTP 502)); claim unknown, the slot may collide; rerun, then ask the coordinator");
+    } finally {
+      rmSync(dir, { recursive: true, force: true });
+    }
+  }, 30_000);
+});