diff --git a/bin/gstack-next-version b/bin/gstack-next-version index df5473b6b..10b147df5 100755 --- a/bin/gstack-next-version +++ b/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 --bump \ @@ -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=` // 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) { diff --git a/test/gstack-next-version.test.ts b/test/gstack-next-version.test.ts index 83993db72..30cc97240 100644 --- a/test/gstack-next-version.test.ts +++ b/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[] = 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); +});